}

        }

        log!(cu.logger, "All proto components are blocked");

        log!(cu.logger, "Entering decision loop...");

        comm.endpoint_manager.undelay_all();

        '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);

                log!(

                    cu.logger,

                    getter,

                );

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

                    }

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

                        let msg = Msg::CommMsg(CommMsg {

                            round_index: comm.round_index,

                            contents: CommMsgContents::SendPayload(send_payload_msg),

                        });

                        comm.endpoint_manager.send_to_comms(*index, &msg)?;

                    }

                    Some(Route::LocalComponent(ComponentId::Native)) => branching_native.feed_msg(

                        cu,

                        &mut rctx.solution_storage,

                        getter,

            logger,

            "Ending round for {} udp endpoints",

            self.udp_endpoint_store.endpoint_exts.len()

        );

        use UnrecoverableSyncError as Use;

        if let Decision::Success(solution_predicate) = decision {

            '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

                        );

                        continue 'endpoint_loop; // send at most one payload per endpoint per round

                    }

                }

                log!(logger, "Sent no message through Udp endpoint {}", index);

            }

use crate::common::*;

#[derive(Debug)]

pub enum ConnectError {

    BindFailed(SocketAddr),

    PollInitFailed,

    Timeout,

    PollFailed,

    AcceptFailed(SocketAddr),

    AlreadyConnected,

    PortPeerPolarityMismatch(PortId),

    NetEndpointSetupError(SocketAddr, NetEndpointError),

    SetupAlgMisbehavior,

}

#[derive(Eq, PartialEq, Copy, Clone, Debug)]

pub enum AddComponentError {

    NoSuchComponent,

}

struct NetEndpoint {

    inbox: Vec<u8>,

    stream: TcpStream,

}

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

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,

    getter_for_incoming: PortId,

}

#[derive(Debug)]

struct Neighborhood {

    parent: Option<usize>,

    children: VecSet<usize>,

}

#[derive(Debug)]

struct ConnectorUnphased {

    proto_description: Arc<ProtocolDescription>,

    proto_components: HashMap<ProtoComponentId, ProtoComponent>,

    logger: Box<dyn Logger>,

    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(Default, Clone, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]

struct Predicate {

    assigned: BTreeMap<SpecVar, SpecVal>,

}

#[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 {

    fn get_deadline(&self) -> &Option<Instant>;

    fn getter_add(&mut self, getter: PortId, msg: SendPayloadMsg);

}

enum CommRecvOk {

    TimeoutWithoutNew,

    NewPayloadMsgs,

        }

    }

    pub fn new_udp_port(

        &mut self,

        polarity: Polarity,

        local_addr: SocketAddr,

        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,

        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) => {

                // {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;

        match &phased {

            ConnectorPhased::Communication { .. } => {

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

                Err(Ce::AlreadyConnected)

            }

            ConnectorPhased::Setup(setup) => {

                if cfg!(feature = "session_optimization") {

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

                }

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

                self.phased = ConnectorPhased::Communication(Box::new(comm));

                Ok(())

            }

        }

    }

}

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,

    }

    enum TodoEndpoint {

        Accepting(TcpListener),

        NetEndpoint(NetEndpoint),

    }

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

    }

    let mut waker_state: Option<Arc<WakerState>> = None;

    let mut poll = Poll::new().map_err(|_| Ce::PollInitFailed)?;

    let mut events = Events::with_capacity(net_endpoint_setups.len() * 2 + 4);

    let [mut net_polled_undrained, udp_polled_undrained] = [VecSet::default(), VecSet::default()];

    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![] })

            } else {

                let mut listener = TcpListener::bind(endpoint_setup.sock_addr)

                    .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

    //    - for each endpoint, recv the peer's PortId, and

    // all in connect_failed are NOT registered with Poll

    let mut connect_failed: HashSet<usize> = Default::default();

    // TODO register udps, and all them to incomplete list

                                        let _ = moved_arc.waker.wake();

                                    }

                                });

                            }

                            continue;

                        }

                        // 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(),

                                        e,

                                    )

                                })

                                .unwrap();

                            log!(logger, "endpoint[{}] sent msg {:?}", index, &msg);

                            net_todo.sent_local_port = true;

                            }

                            match maybe_msg {

                                None => {} // msg deserialization incomplete

                                Some(Msg::SetupMsg(SetupMsg::MyPortInfo(peer_info))) => {

                                    log!(

                                        logger,

                                        "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,

                                            route

                                        );

                                    }

                                    port_info

                                        .routes

                                        .entry(peer_info.port)

        events.clear();

    }

    log!(logger, "Endpoint setup complete! Cleaning up and building structures");

    if let Some(arc) = waker_state {

        log!(logger, "Sending waker the stop signal");

        arc.continue_signal.store(false, std::sync::atomic::Ordering::SeqCst);

        // 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)| {

            let UdpTodo { mut sock, local_port } = udp_todo;

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

            poll.registry().reregister(&mut sock, token, Interest::READABLE).unwrap();

            UdpEndpointExt {

                sock,

                outgoing_payloads: Default::default(),

    if(fires(ia)) {

      put(oa, get(ia));

      put(ob, get(ib));

    }

  } 

}

";

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]

fn basic_connector() {

    Connector::new(Box::new(DummyLogger), MINIMAL_PROTO.clone(), 0, 0);

}

#[test]

fn basic_logged_connector() {

#[test]

fn new_sync() {

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

    let mut c = file_logged_connector(0, test_log_path);

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

    c.add_component(b"sync", &[i, o]).unwrap();

}

#[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();

}

#[test]

fn put_no_sync() {

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

    let mut c = file_logged_connector(0, test_log_path);

    let [o, _] = c.new_port_pair();

    c.connect(SEC1).unwrap();

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

    let mut c = file_logged_connector(0, test_log_path);

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

    c.connect(SEC1).unwrap();

    c.get(i).unwrap();

    c.put(o, TEST_MSG.clone()).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();

}

#[test]

fn trivial_nondet() {

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

    let mut c = file_logged_connector(0, test_log_path);

    c.connect(SEC1).unwrap();

    c.get(i).unwrap();

    // getting 0 batch

    c.next_batch().unwrap();

    // silent 1 batch

    assert_eq!(1, c.sync(SEC1).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 native_immediately_inconsistent() {

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

    let mut c = file_logged_connector(0, test_log_path);

    let [_, g] = c.new_port_pair();

    c.connect(SEC1).unwrap();

    c.get(g).unwrap();

    match c.sync(MS300) {

        Err(SyncError::RoundFailure) => {}

        res => panic!("expeted timeout. but got {:?}", res),

    }

}

#[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();

}

#[test]

fn chain_connect() {

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

    let sock_addrs = [next_test_addr(), next_test_addr(), next_test_addr(), next_test_addr()];

        s.spawn(|_| {

            let mut c = file_logged_connector(1, test_log_path);

            c.new_net_port(Getter, sock_addrs[3], Active).unwrap();

            c.connect(SEC5).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");

    let sock_addrs = [next_test_addr(), next_test_addr()];

    scope(|s| {

        s.spawn(|_| {

            let mut c = file_logged_connector(0, test_log_path);

            let [p0, p1] = c.new_port_pair();

            let p2 = c.new_net_port(Getter, sock_addrs[0], Passive).unwrap();

            c.connect(SEC1).unwrap();

            for succeeds in success_iter.clone() {

                c.get(p0).unwrap();

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

                let res = c.sync(MS300);

                assert_eq!(res.is_ok(), succeeds);

            }

        });

    })

    .unwrap();

}