#include <time.h>

#include "../../reowolf.h"

#include "../utility.c"

int main(int argc, char** argv) {

	int i, proto_components;

	proto_components = atoi(argv[1]);

	printf("proto_components: %d\n", proto_components);

	const unsigned char pdl[] = 

	"primitive trivial_loop() {   "

	"    while(true) synchronous{}"

	"}                            "

	;

	Arc_ProtocolDescription * pd = protocol_description_parse(pdl, sizeof(pdl)-1);

	char logpath[] = "./bench_4.txt";

	Connector * c = connector_new_logging(pd, logpath, sizeof(logpath)-1);

	for (i=0; i<proto_components; i++) {

		printf("Error str `%s`\n", reowolf_error_peek(NULL));

	}

	connector_connect(c, -1);

	printf("Error str `%s`\n", reowolf_error_peek(NULL));

	clock_t begin = clock();

	for (i=0; i<1000000; i++) {

		connector_sync(c, -1);

	}

	clock_t end = clock();

	double time_spent = (double)(end - begin) / CLOCKS_PER_SEC;

	printf("Time taken: %f\n", time_spent);

	return 0;

}

pub unsafe extern "C" fn connector_new(pd: &Arc<ProtocolDescription>) -> *mut Connector {

    let c = Connector::new(Box::new(DummyLogger), pd.clone(), Connector::random_id());

    Box::into_raw(Box::new(c))

}

/// Destroys the given a pointer to the connector on the heap, freeing its resources.

/// Usable in {setup, communication} states.

#[no_mangle]

pub unsafe extern "C" fn connector_destroy(connector: *mut Connector) {

    drop(Box::from_raw(connector))

}

/// Given an initialized connector in setup or connecting state,

/// - Creates a new directed port pair with logical channel putter->getter,

/// - adds the ports to the native component's interface,

/// - and returns them using the given out pointers.

/// Usable in {setup, communication} states.

#[no_mangle]

pub unsafe extern "C" fn connector_add_port_pair(

    connector: &mut Connector,

    out_putter: *mut PortId,

    out_getter: *mut PortId,

) {

    let [o, i] = connector.new_port_pair();

        out_putter.write(o);

        out_getter.write(i);

    }

/// Given

/// - an initialized connector in setup or connecting state,

/// - a string slice for the component's identifier in the connector's configured protocol description,

/// - a set of ports (represented as a slice; duplicates are ignored) in the native component's interface,

/// the connector creates a new (internal) protocol component C, such that the set of native ports are moved to C.

/// Usable in {setup, communication} states.

#[no_mangle]

pub unsafe extern "C" fn connector_add_component(

    connector: &mut Connector,

    ident_ptr: *const u8,

    ident_len: usize,

    ports_ptr: *const PortId,

    ports_len: usize,

) -> c_int {

    StoredError::tl_clear();

    match connector.add_component(

        &*slice_from_raw_parts(ident_ptr, ident_len),

        &*slice_from_raw_parts(ports_ptr, ports_len),

    ) {

        Ok(()) => RW_OK,

        Err(err) => {

            StoredError::tl_debug_store(&err);

            RW_TL_ERR

    // of the new component.

    pub(crate) fn new_component(&mut self, moved_ports: HashSet<PortId>, state: ComponentState) {

        // Sanity check! The moved ports are owned by this component to begin with

        for port in moved_ports.iter() {

            assert_eq!(

                self.proto_component_id,

                self.ips.port_info.map.get(port).unwrap().owner

            );

        }

        // Create the new component, and schedule it to be run

        let new_cid = self.ips.id_manager.new_component_id();

        log!(

            self.logger,

            "Component {:?} added new component {:?} with state {:?}, moving ports {:?}",

            self.proto_component_id,

            new_cid,

            &state,

            &moved_ports

        );

        self.unrun_components.push((new_cid, state));

        // Update the ownership of the moved ports

        for port in moved_ports.iter() {

            self.ips.port_info.map.get_mut(port).unwrap().owner = new_cid;

        }

    }

    // Facilitates callback from the component to the connector runtime,

    // creating a new port-pair connected by an memory channel

    pub(crate) fn new_port_pair(&mut self) -> [PortId; 2] {

        // adds two new associated ports, related to each other, and exposed to the proto component

        let mut new_cid_fn = || self.ips.id_manager.new_port_id();

        let [o, i] = [new_cid_fn(), new_cid_fn()];

        self.ips.port_info.map.insert(

            o,

            PortInfo {

                route: Route::LocalComponent,

                peer: Some(i),

                polarity: Putter,

                owner: self.proto_component_id,

            },

        );

        self.ips.port_info.map.insert(

            i,

            PortInfo {

                route: Route::LocalComponent,

                peer: Some(o),

                polarity: Getter,

                owner: self.proto_component_id,

            },

        );

        log!(

            self.logger,

            "Component {:?} port pair (out->in) {:?} -> {:?}",

            self.proto_component_id,

            o,

            i

        );

        [o, i]

    }

}

impl SyncProtoContext<'_> {

    // The component calls the runtime back, inspecting whether it's associated

    // preidcate has already determined a (speculative) value for the given port's firing variable.

    pub(crate) fn is_firing(&mut self, port: PortId) -> Option<bool> {

        let var = self.rctx.ips.port_info.spec_var_for(port);

        self.predicate.query(var).map(SpecVal::is_firing)

    }

    // The component calls the runtime back, trying to inspect a port's message

    pub(crate) fn read_msg(&mut self, port: PortId) -> Option<&Payload> {

        let maybe_msg = self.branch_inner.inbox.get(&port);

        if maybe_msg.is_some() {

            // Make a note that this component has received

            // this port's message 1+ times this round

    polled_undrained: VecSet<usize>,

}

// The information associated with a port identifier, designed for local storage.

#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]

struct PortInfo {

    owner: ComponentId,

    peer: Option<PortId>,

    polarity: Polarity,

    route: Route,

}

// Similar to `PortInfo`, but designed for communication during the setup procedure.

#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]

struct MyPortInfo {

    polarity: Polarity,

    port: PortId,

    owner: ComponentId,

}

// Newtype around port info map, allowing the implementation of some

// useful methods

#[derive(Default, Debug, Clone, serde::Serialize, serde::Deserialize)]

struct PortInfoMap {

    map: HashMap<PortId, PortInfo>,

}

// A convenient substructure for containing port info and the ID manager.

// Houses the bulk of the connector's persistent state between rounds.

// It turns out several situations require access to both things.

#[derive(Debug, Clone)]

struct IdAndPortState {

    port_info: PortInfoMap,

    id_manager: IdManager,

}

// A component's setup-phase-specific data

#[derive(Debug)]

struct ConnectorCommunication {

    round_index: usize,

    endpoint_manager: EndpointManager,

    neighborhood: Neighborhood,

    native_batches: Vec<NativeBatch>,

    round_result: Result<Option<RoundEndedNative>, SyncError>,

}

// A component's data common to both setup and communication phases

#[derive(Debug)]

struct ConnectorUnphased {

        Self { vec }

    }

    fn contains(&self, element: &T) -> bool {

        self.vec.binary_search(element).is_ok()

    }

    // Insert the given element. Returns whether it was already present.

    fn insert(&mut self, element: T) -> bool {

        match self.vec.binary_search(&element) {

            Ok(_) => false,

            Err(index) => {

                self.vec.insert(index, element);

                true

            }

        }

    }

    fn iter(&self) -> std::slice::Iter<T> {

        self.vec.iter()

    }

    fn pop(&mut self) -> Option<T> {

        self.vec.pop()

    }

}

impl PortInfoMap {

    fn ports_owned_by(&self, owner: ComponentId) -> impl Iterator<Item = &PortId> {

    }

    fn spec_var_for(&self, port: PortId) -> SpecVar {

        // Every port maps to a speculative variable

        // Two distinct ports map to the same variable

        // IFF they are two ends of the same logical channel.

        let info = self.map.get(&port).unwrap();

        SpecVar(match info.polarity {

            Getter => port,

            Putter => info.peer.unwrap(),

        })

    }

}

impl SpecVarStream {

    fn next(&mut self) -> SpecVar {

        let phantom_port: PortId =

            Id { connector_id: self.connector_id, u32_suffix: self.port_suffix_stream.next() }

                .into();

        SpecVar(phantom_port)

    }

}

impl IdManager {

    fn new(connector_id: ConnectorId) -> Self {

        Self {

            connector_id,

            port_suffix_stream: Default::default(),

            component_suffix_stream: Default::default(),

        }

    }

    fn new_spec_var_stream(&self) -> SpecVarStream {

        // Spec var stream starts where the current port_id stream ends, with gap of SKIP_N.

        // This gap is entirely unnecessary (i.e. 0 is fine)

        // It's purpose is only to make SpecVars easier to spot in logs.

        // E.g. spot the spec var: { v0_0, v1_2, v1_103 }

        const SKIP_N: u32 = 100;

        let port_suffix_stream = self.port_suffix_stream.clone().n_skipped(SKIP_N);

        // allocate two fresh port identifiers

        let [o, i] = [new_cid(), new_cid()];

        // store info for each:

        // - they are each others' peers

        // - they are owned by a local component with id `cid`

        // - polarity putter, getter respectively

        cu.ips.port_info.map.insert(

            o,

            PortInfo {

                route: Route::LocalComponent,

                peer: Some(i),

                owner: cu.native_component_id,

                polarity: Putter,

            },

        );

        cu.ips.port_info.map.insert(

            i,

            PortInfo {

                route: Route::LocalComponent,

                peer: Some(o),

                owner: cu.native_component_id,

                polarity: Getter,

            },

        );

        log!(cu.logger, "Added port pair (out->in) {:?} -> {:?}", o, i);

        [o, i]

    }

    /// Instantiates a new component for the connector runtime to manage, and passing

    /// the given set of ports from the interface of the native component, to that of the

    /// newly created component (passing their ownership).

    /// # Errors

    /// Error is returned if the moved ports are not owned by the native component,

    /// if the given component name is not defined in the connector's protocol,

    /// the given sequence of ports contains a duplicate port,

    /// or if the component is unfit for instantiation with the given port sequence.

    /// # Panics

    /// This function panics if the connector's (large) component id space is exhausted.

    pub fn add_component(

        &mut self,

        identifier: &[u8],

        ports: &[PortId],

    ) -> Result<(), AddComponentError> {

        // Check for error cases first before modifying `cu`

        use AddComponentError as Ace;

        let cu = &self.unphased;

        if let Some(port) = duplicate_port(ports) {

            return Err(Ace::DuplicatePort(port));

        if expected_polarities.len() != ports.len() {

            return Err(Ace::WrongNumberOfParamaters { expected: expected_polarities.len() });

        }

        for (&expected_polarity, &port) in expected_polarities.iter().zip(ports.iter()) {

            let info = cu.ips.port_info.map.get(&port).ok_or(Ace::UnknownPort(port))?;

            if info.owner != cu.native_component_id {

                return Err(Ace::UnknownPort(port));

            }

            if info.polarity != expected_polarity {

                return Err(Ace::WrongPortPolarity { port, expected_polarity });

            }

        }

        // No errors! Time to modify `cu`

        // create a new component and identifier

        let cu = &mut self.unphased;

        let new_cid = cu.ips.id_manager.new_component_id();

        cu.proto_components.insert(new_cid, cu.proto_description.new_component(identifier, ports));

        // update the ownership of moved ports

        for port in ports.iter() {

            match cu.ips.port_info.map.get_mut(port) {

                Some(port_info) => port_info.owner = new_cid,

                None => unreachable!(),

            }

        }

        Ok(())

    }

}

impl Predicate {

    #[inline]

    pub fn singleton(k: SpecVar, v: SpecVal) -> Self {

        Self::default().inserted(k, v)

    }

    #[inline]

    pub fn inserted(mut self, k: SpecVar, v: SpecVal) -> Self {

        self.assigned.insert(k, v);

        self

    }

    // Return true whether `self` is a subset of `maybe_superset`

    pub fn assigns_subset(&self, maybe_superset: &Self) -> bool {

        for (var, val) in self.assigned.iter() {

            match maybe_superset.assigned.get(var) {

                Some(val2) if val2 == val => {}

                _ => return false, // var unmapped, or mapped differently

            }

        }

        // `maybe_superset` mirrored all my assignments!

        true

                    uout,

                    PortInfo {

                        route: Route::UdpEndpoint { index: udp_index },

                        polarity: Putter,

                        peer: Some(uin),

                        owner: udp_cid,

                    },

                );

                cu.ips.port_info.map.insert(

                    nin,

                    PortInfo {

                        route: Route::LocalComponent,

                        polarity: Getter,

                        peer: Some(uout),

                        owner: cu.native_component_id,

                    },

                );

                // allocate the two ports owned by the UdpMediator component

                // Remember to setup this UdpEndpoint setup during `connect` later.

                setup.udp_endpoint_setups.push(UdpEndpointSetup {

                    local_addr,

                    peer_addr,

                    getter_for_incoming: nin,

                });

                // Return the native's output, input port pair

                Ok([nout, nin])

            }

        }

    }

    /// Adds a "dangling" port to the connector in the setup phase,

    /// to be formed into channel during the connect procedure with the given

    /// transport layer information.

    pub fn new_net_port(

        &mut self,

        polarity: Polarity,

        sock_addr: SocketAddr,

        endpoint_polarity: EndpointPolarity,

    ) -> Result<PortId, WrongStateError> {

        let Self { unphased: cu, phased } = self;

        match phased {

            ConnectorPhased::Communication(..) => Err(WrongStateError),

            ConnectorPhased::Setup(setup) => {

                // allocate a single dangling port with a `None` peer (for now)

                let new_pid = cu.ips.id_manager.new_port_id();

                cu.ips.port_info.map.insert(

                    new_pid,

                    PortInfo {

                        route: Route::LocalComponent,

                        peer: None,

                        owner: cu.native_component_id,

                        polarity,

                    },

                );

                log!(

                    cu.logger,

                    "Added net port {:?} with polarity {:?} addr {:?} endpoint_polarity {:?}",

                    new_pid,

                    polarity,

                    &sock_addr,

                    endpoint_polarity

                );

                // Remember to setup this NetEndpoint setup during `connect` later.

                setup.net_endpoint_setups.push(NetEndpointSetup {

                    sock_addr,

                    endpoint_polarity,

                    getter_for_incoming: new_pid,

                });

                Ok(new_pid)

            }

        }

    }

    /// Finalizes the connector's setup procedure and forms a distributed system with

    /// all other connectors reachable through network channels. This procedure represents

    /// a synchronization barrier, and upon successful return, the connector can no longer add new network ports,

    /// but is ready to begin the first communication round.

    /// Initially, the connector has a singleton set of _batches_, the only element of which is empty.

    /// This single element starts off selected. The selected batch is modified with `put` and `get`,

    /// and new batches are added and selected with `next_batch`. See `sync` for an explanation of the

    /// purpose of these batches.

    pub fn connect(&mut self, timeout: Option<Duration>) -> Result<(), ConnectError> {

        use ConnectError as Ce;

        let Self { unphased: cu, phased } = self;

        match &phased {

            ConnectorPhased::Communication { .. } => {

                log!(cu.logger, "Call to connecting in connected state");

                Err(Ce::AlreadyConnected)

            }

            ConnectorPhased::Setup(setup) => {

                log!(cu.logger, "~~~ CONNECT called timeout {:?}", timeout);

                let deadline = timeout.map(|to| Instant::now() + to);

                let neighborhood = init_neighborhood(

                    cu.ips.id_manager.connector_id,

                    &mut *cu.logger,

                    &mut endpoint_manager,

                    &deadline,

                )?;

                log!(cu.logger, "Successfully created neighborhood {:?}", &neighborhood);

                // Put it all together with an initial round index of zero.

                let mut comm = ConnectorCommunication {

                    round_index: 0,

                    endpoint_manager,

                    neighborhood,

                    native_batches: vec![Default::default()],

                    round_result: Ok(None), // no previous round yet

                };

                if cfg!(feature = "session_optimization") {

                    // Perform the session optimization procedure, which may modify the

                    // internals of the connector, rerouting ports, moving around connectors etc.

                    session_optimize(cu, &mut comm, &deadline)?;

                }

                log!(cu.logger, "connect() finished. setup phase complete");

                // Connect procedure successful! Commit changes by...

                // ... commiting new port info for ConnectorUnphased

                for (port, info) in extra_port_info.info.drain() {

                    cu.ips.port_info.map.insert(port, info);

                }

                for (port, peer) in extra_port_info.peers.drain() {

                    cu.ips.port_info.map.get_mut(&port).unwrap().peer = Some(peer);

                }

                // ... replacing the connector's phase to "communication"

                *phased = ConnectorPhased::Communication(Box::new(comm));

                Ok(())

            }

        }

    }

}

// Given a set of net_ and udp_ endpoints to setup,

// port information to flesh out (by discovering peers through channels)

// and a deadline in which to do it,

// try to return:

// - An EndpointManager, containing all the set up endpoints

// - new information about ports acquired through the newly-created channels

fn setup_endpoints_and_pair_ports(

    logger: &mut dyn Logger,

    net_endpoint_setups: &[NetEndpointSetup],

    udp_endpoint_setups: &[UdpEndpointSetup],

    port_info: &PortInfoMap,

    logger: &mut dyn Logger,

    unoptimized_map: HashMap<ConnectorId, SessionInfo>,

) -> Result<HashMap<ConnectorId, SessionInfo>, ConnectError> {

    log!(logger, "Session map optimize START");

    // currently, it's the identity function

    log!(logger, "Session map optimize END");

    Ok(unoptimized_map)

}

// Modify the given connector's internals to reflect

// the given session info

fn apply_my_optimizations(

    cu: &mut ConnectorUnphased,

    comm: &mut ConnectorCommunication,

    session_info: SessionInfo,

) -> Result<(), ConnectError> {

    let SessionInfo {

        proto_components,

        port_info,

        serde_proto_description,

        endpoint_incoming_to_getter,

    } = session_info;

    // simply overwrite the contents

    cu.ips.port_info = port_info;

    cu.proto_components = proto_components;

    cu.proto_description = serde_proto_description.0;

    for (ee, getter) in comm

        .endpoint_manager

        .net_endpoint_store

        .endpoint_exts

        .iter_mut()

        .zip(endpoint_incoming_to_getter)

    {

        ee.getter_for_incoming = getter;

    }

    Ok(())

}

use std::{fs::File, net::SocketAddr, path::Path, sync::Arc, time::Duration};

//////////////////////////////////////////

const MS100: Option<Duration> = Some(Duration::from_millis(100));

const MS300: Option<Duration> = Some(Duration::from_millis(300));

const SEC1: Option<Duration> = Some(Duration::from_secs(1));

const SEC5: Option<Duration> = Some(Duration::from_secs(5));

const SEC15: Option<Duration> = Some(Duration::from_secs(15));

fn next_test_addr() -> SocketAddr {

    use std::{

        net::{Ipv4Addr, SocketAddrV4},

        sync::atomic::{AtomicU16, Ordering::SeqCst},

    };

    static TEST_PORT: AtomicU16 = AtomicU16::new(5_000);

    let port = TEST_PORT.fetch_add(1, SeqCst);

    SocketAddrV4::new(Ipv4Addr::LOCALHOST, port).into()

}

fn file_logged_connector(connector_id: ConnectorId, dir_path: &Path) -> Connector {

    file_logged_configured_connector(connector_id, dir_path, MINIMAL_PROTO.clone())

}

fn file_logged_configured_connector(

    connector_id: ConnectorId,

    dir_path: &Path,

    pd: Arc<ProtocolDescription>,

) -> Connector {

    let path = dir_path.join(format!("cid_{:?}.txt", connector_id));