'undecided: loop {

            // drain payloads_to_get, sending them through endpoints / feeding them to components

            log!(cu.logger, "Decision loop! have {} messages to recv", rctx.getter_buffer.len());

            while let Some((getter, send_payload_msg)) = rctx.getter_buffer.pop() {

                assert!(cu.port_info.polarities.get(&getter).copied() == Some(Getter));

                let route = cu.port_info.routes.get(&getter);

                match route {

                    Some(Route::UdpEndpoint { index }) => {

                    }

                    Some(Route::NetEndpoint { index }) => {

                        let msg = Msg::CommMsg(CommMsg {

                            round_index: comm.round_index,

                            contents: CommMsgContents::SendPayload(send_payload_msg),

                        });

            'endpoint_loop: for (index, ee) in

                self.udp_endpoint_store.endpoint_exts.iter_mut().enumerate()

            {

                ee.incoming_round_spec_var = None; // shouldn't be accessed before its overwritten next round; still adding for clarity.

                for (payload_predicate, payload) in ee.outgoing_payloads.drain() {

                    if payload_predicate.assigns_subset(solution_predicate) {

                        log!(

                            logger,

                            "Sent payload {:?} with pred {:?} through Udp endpoint {}",

                            &payload,

                            &payload_predicate,

                            index

use crate::common::*;

#[derive(Debug)]

pub enum ConnectError {

    BindFailed(SocketAddr),

    PollInitFailed,

    Timeout,

    PollFailed,

    AcceptFailed(SocketAddr),

    AlreadyConnected,

    PortPeerPolarityMismatch(PortId),

struct ProtoComponent {

    state: ComponentState,

    ports: HashSet<PortId>,

}

#[derive(Debug, Clone)]

struct NetEndpointSetup {

    sock_addr: SocketAddr,

    endpoint_polarity: EndpointPolarity,

}

#[derive(Debug, Clone)]

struct UdpEndpointSetup {

    local_addr: SocketAddr,

    peer_addr: SocketAddr,

}

#[derive(Debug)]

struct NetEndpointExt {

    net_endpoint: NetEndpoint,

    id_manager: IdManager,

    native_ports: HashSet<PortId>,

    port_info: PortInfo,

}

#[derive(Debug)]

struct ConnectorSetup {

    surplus_sockets: u16,

}

#[derive(Debug)]

enum ConnectorPhased {

    Setup(Box<ConnectorSetup>),

    Communication(Box<ConnectorCommunication>),

#[derive(Debug, Default)]

struct NativeBatch {

    // invariant: putters' and getters' polarities respected

    to_put: HashMap<PortId, Payload>,

    to_get: HashSet<PortId>,

}

enum TokenTarget {

    NetEndpoint { index: usize },

    UdpEndpoint { index: usize },

    Waker,

}

trait RoundCtxTrait {

        peer_addr: SocketAddr,

    ) -> Result<PortId, WrongStateError> {

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

        match phased {

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

            ConnectorPhased::Setup(setup) => {

                let udp_index = setup.udp_endpoint_setups.len();

                let [port_nat, port_udp] =

                    [cu.id_manager.new_port_id(), cu.id_manager.new_port_id()];

                cu.native_ports.insert(port_nat);

                cu.port_info.peers.insert(port_nat, port_udp);

                cu.port_info.peers.insert(port_udp, port_nat);

                cu.port_info.routes.insert(port_nat, Route::LocalComponent(ComponentId::Native));

                cu.port_info.routes.insert(port_udp, Route::UdpEndpoint { index: udp_index });

                cu.port_info.polarities.insert(port_nat, polarity);

                cu.port_info.polarities.insert(port_udp, !polarity);

                Ok(port_nat)

            }

        }

    }

    pub fn new_net_port(

        &mut self,

        endpoint_polarity: EndpointPolarity,

    ) -> Result<PortId, WrongStateError> {

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

        match phased {

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

            ConnectorPhased::Setup(setup) => {

                // {polarity, route} known. {peer} unknown.

                log!(

                    cu.logger,

                    polarity,

                );

            }

        }

    }

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

        use ConnectError as Ce;

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

            }

        }

    }

}

fn new_endpoint_manager(

    logger: &mut dyn Logger,

    port_info: &mut PortInfo,

    deadline: &Option<Instant>,

) -> Result<EndpointManager, ConnectError> {

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

    use std::sync::atomic::AtomicBool;

    use ConnectError as Ce;

    const BOTH: Interest = Interest::READABLE.add(Interest::WRITABLE);

    struct Todo {

        todo_endpoint: TodoEndpoint,

        endpoint_setup: NetEndpointSetup,

        sent_local_port: bool,          // true <-> I've sent my local port

        recv_peer_port: Option<PortId>, // Some(..) <-> I've received my peer's port

    }

    struct UdpTodo {

        local_port: PortId,

        sock: UdpSocket,

    let mut delayed_messages = vec![];

    // 2. create a registered (TcpListener/Endpoint) for passive / active respectively

    let mut net_todos = net_endpoint_setups

        .iter()

        .enumerate()

            let token = TokenTarget::NetEndpoint { index }.into();

            let todo_endpoint = if let EndpointPolarity::Active = endpoint_setup.endpoint_polarity {

                let mut stream = TcpStream::connect(endpoint_setup.sock_addr)

                    .expect("mio::TcpStream connect should not fail!");

                poll.registry().register(&mut stream, token, BOTH).unwrap();

                TodoEndpoint::NetEndpoint(NetEndpoint { stream, inbox: vec![] })

                    .map_err(|_| Ce::BindFailed(endpoint_setup.sock_addr))?;

                poll.registry().register(&mut listener, token, BOTH).unwrap();

                TodoEndpoint::Accepting(listener)

            };

            Ok(Todo {

                todo_endpoint,

                sent_local_port: false,

                recv_peer_port: None,

                endpoint_setup: endpoint_setup.clone(),

            })

        })

        .collect::<Result<Vec<Todo>, ConnectError>>()?;

    let udp_todos = udp_endpoint_setups

        .iter()

        .enumerate()

            let mut sock = UdpSocket::bind(endpoint_setup.local_addr)

                .map_err(|_| Ce::BindFailed(endpoint_setup.local_addr))?;

            poll.registry()

                .register(&mut sock, TokenTarget::UdpEndpoint { index }.into(), Interest::WRITABLE)

                .unwrap();

        })

        .collect::<Result<Vec<UdpTodo>, ConnectError>>()?;

    // 3. Using poll to drive progress:

    //    - accept an incoming connection for each TcpListener (turning them into endpoints too)

    //    - for each endpoint, send the local PortId

                        // event wasn't ERROR

                        if connect_failed.contains(&index) {

                            // spurious wakeup

                            continue;

                        }

                        let local_polarity =

                        if event.is_writable() && !net_todo.sent_local_port {

                            // can write and didn't send setup msg yet? Do so!

                            let msg = Msg::SetupMsg(SetupMsg::MyPortInfo(MyPortInfo {

                                polarity: local_polarity,

                            }));

                            net_endpoint

                                .send(&msg)

                                .map_err(|e| {

                                    Ce::NetEndpointSetupError(

                                        net_endpoint.stream.local_addr().unwrap(),

                                        "endpoint[{}] got peer info {:?}",

                                        index,

                                        peer_info

                                    );

                                    if peer_info.polarity == local_polarity {

                                        return Err(ConnectError::PortPeerPolarityMismatch(

                                        ));

                                    }

                                    net_todo.recv_peer_port = Some(peer_info.port);

                                    // 1. finally learned the peer of this port!

                                    // 2. learned the info of this peer port

                                    port_info.polarities.insert(peer_info.port, peer_info.polarity);

                                    if let Some(route) = port_info.routes.get(&peer_info.port) {

                                        // check just for logging purposes

                                        log!(

                                            logger,

                                            "Special case! Route to peer {:?} already known to be {:?}. Leave untouched",

                                            peer_info.port,

        // TODO leave the waker registered?

    }

    let net_endpoint_exts = net_todos

        .into_iter()

        .enumerate()

            net_endpoint: match todo_endpoint {

                TodoEndpoint::NetEndpoint(mut net_endpoint) => {

                    let token = TokenTarget::NetEndpoint { index }.into();

                    poll.registry()

                        .reregister(&mut net_endpoint.stream, token, Interest::READABLE)

                        .unwrap();

                    net_endpoint

                }

                _ => unreachable!(),

            },

        })

        .collect();

    let udp_endpoint_exts = udp_todos

        .into_iter()

        .enumerate()

        .map(|(index, udp_todo)| {

";

lazy_static::lazy_static! {

    static ref MINIMAL_PROTO: Arc<ProtocolDescription> = {

        Arc::new(reowolf::ProtocolDescription::parse(MINIMAL_PDL).unwrap())

    };

}

lazy_static::lazy_static! {

    static ref TEST_MSG: Payload = {

    };

}

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

#[test]

}

#[test]

fn new_net_port() {

    let test_log_path = Path::new("./logs/new_net_port");

    let mut c = file_logged_connector(0, test_log_path);

}

#[test]

fn trivial_connect() {

    let test_log_path = Path::new("./logs/trivial_connect");

    let mut c = file_logged_connector(0, test_log_path);

    c.connect(SEC1).unwrap();

}

#[test]

fn single_node_connect() {

    let test_log_path = Path::new("./logs/single_node_connect");

    let mut c = file_logged_connector(0, test_log_path);

    c.connect(SEC1).unwrap();

}

#[test]

fn minimal_net_connect() {

    let test_log_path = Path::new("./logs/minimal_net_connect");

    scope(|s| {

        s.spawn(|_| {

            let mut c = file_logged_connector(0, test_log_path);

            c.connect(SEC1).unwrap();

        });

        s.spawn(|_| {

            let mut c = file_logged_connector(1, test_log_path);

            c.connect(SEC1).unwrap();

        });

    })

    .unwrap();

}

    c.sync(SEC1).unwrap();

}

#[test]

fn two_natives_msg() {

    let test_log_path = Path::new("./logs/two_natives_msg");

    scope(|s| {

        s.spawn(|_| {

            let mut c = file_logged_connector(0, test_log_path);

            c.connect(SEC1).unwrap();

            c.get(g).unwrap();

            c.sync(SEC1).unwrap();

            c.gotten(g).unwrap();

        });

        s.spawn(|_| {

            let mut c = file_logged_connector(1, test_log_path);

            c.connect(SEC1).unwrap();

            c.put(p, TEST_MSG.clone()).unwrap();

            c.sync(SEC1).unwrap();

        });

    })

    .unwrap();

    c.gotten(i).unwrap_err();

}

#[test]

fn connector_pair_nondet() {

    let test_log_path = Path::new("./logs/connector_pair_nondet");

    scope(|s| {

        s.spawn(|_| {

            let mut c = file_logged_connector(0, test_log_path);

            c.connect(SEC1).unwrap();

            c.next_batch().unwrap();

            c.get(g).unwrap();

            assert_eq!(1, c.sync(SEC1).unwrap());

            c.gotten(g).unwrap();

        });

        s.spawn(|_| {

            let mut c = file_logged_connector(1, test_log_path);

            c.connect(SEC1).unwrap();

            c.put(p, TEST_MSG.clone()).unwrap();

            c.sync(SEC1).unwrap();

        });

    })

    .unwrap();

    }

}

#[test]

fn parent_timeout() {

    let test_log_path = Path::new("./logs/parent_timeout");

    scope(|s| {

        s.spawn(|_| {

            // parent; times out

            let mut c = file_logged_connector(999, test_log_path);

            c.connect(SEC1).unwrap();

            c.sync(MS300).unwrap_err(); // timeout

        });

        s.spawn(|_| {

            // child

            let mut c = file_logged_connector(000, test_log_path);

            c.connect(SEC1).unwrap();

            c.get(g).unwrap(); // not matched by put

            c.sync(None).unwrap_err(); // no timeout

        });

    })

    .unwrap();

}

#[test]

fn child_timeout() {

    let test_log_path = Path::new("./logs/child_timeout");

    scope(|s| {

        s.spawn(|_| {

            // child; times out

            let mut c = file_logged_connector(000, test_log_path);

            c.connect(SEC1).unwrap();

            c.sync(MS300).unwrap_err(); // timeout

        });

        s.spawn(|_| {

            // parent

            let mut c = file_logged_connector(999, test_log_path);

            c.connect(SEC1).unwrap();

            c.get(g).unwrap(); // not matched by put

            c.sync(None).unwrap_err(); // no timeout

        });

    })

    .unwrap();

    .unwrap();

}

#[test]

fn net_self_loop() {

    let test_log_path = Path::new("./logs/net_self_loop");

    let mut c = file_logged_connector(0, test_log_path);

    c.connect(SEC1).unwrap();

    c.put(p, TEST_MSG.clone()).unwrap();

    c.get(g).unwrap();

    c.sync(MS300).unwrap();

}

#[test]

fn nobody_connects_active() {

    let test_log_path = Path::new("./logs/nobody_connects_active");

    let mut c = file_logged_connector(0, test_log_path);

    c.connect(Some(Duration::from_secs(5))).unwrap_err();

}

#[test]

fn nobody_connects_passive() {

    let test_log_path = Path::new("./logs/nobody_connects_passive");

    let mut c = file_logged_connector(0, test_log_path);

    c.connect(Some(Duration::from_secs(5))).unwrap_err();

}

#[test]

fn together() {

    let test_log_path = Path::new("./logs/together");

            }

        });

    })

    .unwrap();

}