#include <time.h>

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

#include "../utility.c"

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

	int i, cid;

	cid = atoi(argv[1]);

	printf("cid %d\n", cid);

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

	unsigned char pdl[] = "";

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

	Connector * c = connector_new_with_id(pd, cid);

	bool seen_delim = false;

	for(i=2; i<argc; i++) {

		EndpointPolarity ep;

		Polarity p;

		if(argv[i][0] == '.') {

			seen_delim = true;

			continue;

		} else if(seen_delim) {

			p = Polarity_Getter;

			ep = EndpointPolarity_Passive;

		} else {

			p = Polarity_Putter;

			ep = EndpointPolarity_Active;

		}

		FfiSocketAddr addr = {{127, 0, 0, 1}, atoi(argv[i])};

		printf("@%d\n", addr.port);

		connector_add_net_port(c, NULL, addr, p, ep);

	}

	printf("Added all ports!\n");

	connector_connect(c, -1);

	clock_t begin = clock();

		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;

}

	ports_used = atoi(argv[5]);

	do_puts = argv[6][0]; // 't' or 'f'

	do_gets = argv[7][0];

	printf("cid %d, min_putter %d, min_getter %d, ports_tot %d, ports_used %d, do_puts %c, do_gets %c\n",

		cid, min_putter, min_getter, ports_tot, ports_used, do_puts, do_gets);

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

	unsigned char pdl[] = "";

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

	Connector * c = connector_new_with_id(pd, cid);

	PortId putters[ports_tot], getters[ports_tot];

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

		connector_add_net_port(c, &putters[i],

			(FfiSocketAddr){{127, 0, 0, 1}, min_putter+i},

			Polarity_Putter, EndpointPolarity_Active);

		connector_add_net_port(c, &getters[i],

			(FfiSocketAddr){{127, 0, 0, 1}, min_getter+i},

			Polarity_Getter, EndpointPolarity_Active);

	}

	connector_connect(c, -1);

	printf("connect ok!\n");

	clock_t begin = clock();

	char msg[] = "Hello, world!";

		for(j=0; j<ports_used; j++) {

			if(do_gets=='y') connector_get(c, getters[j]);

			if(do_puts=='y') connector_put_bytes(c, putters[j], msg, sizeof(msg)-1);

		}

		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;

}

#include <time.h>

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

#include "../utility.c"

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

	forwards = atoi(argv[1]);

	unsigned char pdl[] = ""; 

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

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

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

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

	PortId native_putter, native_getter;

	connector_add_port_pair(c, &native_putter, &native_getter);

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

		PortId putter, getter;

		connector_add_port_pair(c, &putter, &getter);

		// native ports: {native_putter, native_getter, putter, getter}

		char ident[] = "forward"; // defined in reowolf's stdlib 

		// thread a forward component onto native_tail

		connector_add_component(c, ident, sizeof(ident)-1, (PortId[]){native_getter, putter}, 2);

		// native ports: {native_putter, getter}

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

		native_getter = getter;

	}

	connector_connect(c, -1);

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

	clock_t begin = clock();

		connector_get(c, native_getter);

		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;

}

use super::*;

enum PollAndPopulateError {

    PollFailed,

    Timeout,

}

struct TryRecvAnyNetError {

    error: NetEndpointError,

    index: usize,

}

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

impl NetEndpoint {

    // Returns the bincode configuration the NetEndpoint uses pervasively

    // for configuration on ser/de operations.

    fn bincode_opts() -> impl bincode::config::Options {

        // uses variable-length encoding everywhere; great!

        bincode::config::DefaultOptions::default()

    }

    // Attempt to return some deserializable T-type from

    // the inbox or network stream

    pub(super) fn try_recv<T: serde::de::DeserializeOwned>(

        &mut self,

        logger: &mut dyn Logger,

    ) -> Result<Option<T>, NetEndpointError> {

        use NetEndpointError as Nee;

        // populate inbox with bytes as much as possible (mio::TcpStream is nonblocking)

        let before_len = self.inbox.len();

        'read_loop: loop {

            let res = self.stream.read_to_end(&mut self.inbox);

            match res {

                Err(e) if err_would_block(&e) => break 'read_loop,

                Ok(0) => break 'read_loop,

                Ok(_) => (),

                Err(_e) => return Err(Nee::BrokenNetEndpoint),

            }

        }

        log!(

            @ENDPT,

            logger,

            "Inbox bytes [{:x?}| {:x?}]",

            DenseDebugHex(&self.inbox[..before_len]),

            DenseDebugHex(&self.inbox[before_len..]),

        );

        use bincode::config::Options;

            Ok(msg) => {

                // inbox[..msg_size] was deserialized into one message!

                self.inbox.drain(..msg_size);

                log!(

                    @ENDPT,

                    logger,

                    "Yielding msg. Inbox len {}-{}=={}: [{:?}]",

                    self.inbox.len() + msg_size,

                    msg_size,

                    self.inbox.len(),

                    DenseDebugHex(&self.inbox[..]),

                );

                Ok(Some(msg))

            }

            Err(e) => match *e {

                bincode::ErrorKind::Io(k) if k.kind() == std::io::ErrorKind::UnexpectedEof => {

                    // Contents of inbox insufficient for deserializing a message

                    Ok(None)

                }

                _ => Err(Nee::MalformedMessage),

            },

        }

    }

    // Send the given serializable type into the stream

        &mut self,

        msg: &T,

    ) -> Result<(), NetEndpointError> {

        use bincode::config::Options;

        use NetEndpointError as Nee;

            .map_err(|_| Nee::BrokenNetEndpoint)?;

        let _ = self.stream.flush();

        Ok(())

    }

}

impl EndpointManager {

    pub(super) fn index_iter(&self) -> Range<usize> {

        0..self.num_net_endpoints()

    }

    pub(super) fn num_net_endpoints(&self) -> usize {

        self.net_endpoint_store.endpoint_exts.len()

    }

    // Setup-phase particular send procedure.

    // Used pervasively, allows for some brevity with the ? operator.

    pub(super) fn send_to_setup(&mut self, index: usize, msg: &Msg) -> Result<(), ConnectError> {

        let net_endpoint = &mut self.net_endpoint_store.endpoint_exts[index].net_endpoint;

            ConnectError::NetEndpointSetupError(net_endpoint.stream.local_addr().unwrap(), err)

        })

    }

    // Communication-phase particular send procedure.

    // Used pervasively, allows for some brevity with the ? operator.

    pub(super) fn send_to_comms(

        &mut self,

        index: usize,

        msg: &Msg,

    ) -> Result<(), UnrecoverableSyncError> {

        use UnrecoverableSyncError as Use;

        let net_endpoint = &mut self.net_endpoint_store.endpoint_exts[index].net_endpoint;

    }

    /// Receive the first message of any kind at all.

    /// Why not return SetupMsg? Because often this message will be forwarded to several others,

    /// and by returning a Msg, it can be serialized in-place (NetEndpoints allow the sending of Msg types!)

    pub(super) fn try_recv_any_setup(

        &mut self,

        logger: &mut dyn Logger,

        deadline: &Option<Instant>,

    ) -> Result<(usize, Msg), ConnectError> {

        // helper function, mapping a TryRecvAnySetup type error

        // into a ConnectError

        fn map_trane(

            trane: TryRecvAnyNetError,

            net_endpoint_store: &EndpointStore<NetEndpointExt>,

        ) -> ConnectError {

            ConnectError::NetEndpointSetupError(

                net_endpoint_store.endpoint_exts[trane.index]

                    .net_endpoint

                    .stream

                    .local_addr()

                    .unwrap(), // stream must already be connected

                trane.error,

            )

        // pop undelayed messages, handling them. Return the first CommCtrlMsg popped

        while let Some((net_index, msg)) = self.undelayed_messages.pop() {

            if let Some((net_index, msg)) =

                self.handle_msg(cu, rctx, net_index, msg, round_index, &mut some_message_enqueued)

            {

                return Ok(CommRecvOk::NewControlMsg { net_index, msg });

            }

        }

        loop {

            // drain endpoints of incoming messages (without blocking)

            // return first CommCtrlMsg received

            while let Some((net_index, msg)) = self.try_recv_undrained_net(cu.logger())? {

                if let Some((net_index, msg)) = self.handle_msg(

                    cu,

                    rctx,

                    net_index,

                    msg,

                    round_index,

                    &mut some_message_enqueued,

                ) {

                    return Ok(CommRecvOk::NewControlMsg { net_index, msg });

                }

            }

            // try receive a udp message

            while let Some(index) = self.udp_endpoint_store.polled_undrained.pop() {

                let ee = &mut self.udp_endpoint_store.endpoint_exts[index];

                if let Some(bytes_written) = ee.sock.recv(recv_buffer).ok() {

                    // I received a payload!

                    self.udp_endpoint_store.polled_undrained.insert(index);

                    if !ee.received_this_round {

                        let payload = Payload::from(&recv_buffer[..bytes_written]);

                        let port_spec_var = rctx.ips.port_info.spec_var_for(ee.getter_for_incoming);

                        let predicate = Predicate::singleton(port_spec_var, SpecVal::FIRING);

                        rctx.getter_push(

                            ee.getter_for_incoming,

                            SendPayloadMsg { payload, predicate },

                        );

                        some_message_enqueued = true;

                        ee.received_this_round = true;

                    } else {

                        // lose the message!

                    }

                }

            }

            if some_message_enqueued {

                return Ok(CommRecvOk::NewPayloadMsgs);

            }

            // poll if time remains

                }

                ee.received_this_round = false;

            }

        }

        Ok(())

    }

}

impl Debug for NetEndpoint {

    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {

        struct DebugStream<'a>(&'a TcpStream);

        impl Debug for DebugStream<'_> {

            fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {

                f.debug_struct("Endpoint")

                    .field("local_addr", &self.0.local_addr())

                    .field("peer_addr", &self.0.peer_addr())

                    .finish()

            }

        }

        f.debug_struct("Endpoint")

            .field("inbox", &self.inbox)

            .field("stream", &DebugStream(&self.stream))

            .finish()

    }

}

impl Into<Msg> for SetupMsg {

    fn into(self) -> Msg {

        Msg::SetupMsg(self)

    }

}

impl From<PollAndPopulateError> for ConnectError {

    fn from(pape: PollAndPopulateError) -> ConnectError {

        use {ConnectError as Ce, PollAndPopulateError as Pape};

        match pape {

            Pape::PollFailed => Ce::PollFailed,

            Pape::Timeout => Ce::Timeout,

        }

    }

}

impl From<TryRecvAnyNetError> for UnrecoverableSyncError {

    fn from(trane: TryRecvAnyNetError) -> UnrecoverableSyncError {

        let TryRecvAnyNetError { index, .. } = trane;

        UnrecoverableSyncError::BrokenNetEndpoint { index }

    }

}

#[derive(Debug)]

struct UdpEndpointExt {

    sock: UdpSocket, // already bound and connected

    received_this_round: bool,

    outgoing_payloads: HashMap<Predicate, Payload>,

    getter_for_incoming: PortId,

}

// Meta-data for the connector: its role in the consensus tree.

#[derive(Debug)]

struct Neighborhood {

    parent: Option<usize>,

    children: VecSet<usize>,

}

// Manages the connector's ID, and manages allocations for connector/port IDs.

#[derive(Debug, Clone)]

struct IdManager {

    connector_id: ConnectorId,

    port_suffix_stream: U32Stream,

    component_suffix_stream: U32Stream,

}

// Newtype wrapper around a byte buffer, used for UDP mediators to receive incoming datagrams.

    byte_vec: Vec<u8>,

}

// A generator of speculative variables. Created on-demand during the synchronous round

// by the IdManager.

#[derive(Debug)]

struct SpecVarStream {

    connector_id: ConnectorId,

    port_suffix_stream: U32Stream,

}

// Manages the messy state of the various endpoints, pollers, buffers, etc.

#[derive(Debug)]

struct EndpointManager {

    // invariants:

    // 1. net and udp endpoints are registered with poll with tokens computed with TargetToken::into

    // 2. Events is empty

    poll: Poll,

    events: Events,

    delayed_messages: Vec<(usize, Msg)>,

    undelayed_messages: Vec<(usize, Msg)>, // ready to yield

    net_endpoint_store: EndpointStore<NetEndpointExt>,

    udp_endpoint_store: EndpointStore<UdpEndpointExt>,

}

// A storage of endpoints, which keeps track of which components have raised

// an event during poll(), signifying that they need to be checked for new incoming data

#[derive(Debug)]

struct EndpointStore<T> {

    endpoint_exts: Vec<T>,

    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,

    }

}

impl Debug for SpecVar {

    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {

        let (a, b) = self.id_parts();

        write!(f, "v{}_{}", a, b)

    }

}

impl SpecVal {

    const FIRING: Self = SpecVal(1);

    const SILENT: Self = SpecVal(0);

    fn is_firing(self) -> bool {

        self == Self::FIRING

        // all else treated as SILENT

    }

    fn iter_domain() -> impl Iterator<Item = Self> {

        (0..).map(SpecVal)

    }

}

impl Debug for SpecVal {

    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {

        self.0.fmt(f)

    }

}

    fn default() -> Self {

        let mut byte_vec = Vec::with_capacity(Self::CAPACITY);

        unsafe {

            // safe! this vector is guaranteed to have sufficient capacity