PutMsg(PortId, Payload),

    NondetChoice { n: u16 },

}

///////////////////// IMPL /////////////////////

impl IdParts for Id {

    fn id_parts(self) -> (ConnectorId, U32Suffix) {

        // }

    }};

    (@ENDPT, $logger:expr, $($arg:tt)*) => {{

    }};

    ($logger:expr, $($arg:tt)*) => {{

        // if let Some(w) = $logger.line_writer() {

    ) -> Result<&mut NativeBatch, PortOpError> {

        use PortOpError as Poe;

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

        if info.owner != cu.native_component_id {

            return Err(Poe::PortUnavailable);

        }

                );

                let firing_ports: HashSet<PortId> = firing_iter.clone().collect();

                for port in firing_iter {

                    predicate.assigned.insert(var, SpecVal::FIRING);

                }

                // all silent ports have SpecVal::SILENT

                    if firing_ports.contains(port) {

                        // this one is FIRING

                        continue;

                    }

                    if let Some(SpecVal::FIRING) = predicate.assigned.insert(var, SpecVal::SILENT) {

                        continue 'native_branches;

                    }

                }

                    putter

                );

                // sanity check

                rctx.putter_push(cu, putter, msg);

            }

            let branch = NativeBranch { index, gotten: Default::default(), to_get };

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

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

                log!(@MARK, cu.logger(), "handling payload msg for getter {:?} of {:?}", getter, &send_payload_msg);

                let cid = getter_info.owner; // the id of the component owning `getter` port

                assert_eq!(Getter, getter_info.polarity); // sanity check

                log!(

        bn_temp: MapTempGuard<'_, Predicate, NativeBranch>,

    ) {

        log!(cu.logger(), "feeding native getter {:?} {:?}", getter, &send_payload_msg);

        let mut draining = bn_temp;

        let finished = &mut self.branches;

        std::mem::swap(draining.0, finished);

        // consistent with that of the received message.

        for (predicate, mut branch) in draining.drain() {

            log!(cu.logger(), "visiting native branch {:?} with {:?}", &branch, &predicate);

            if predicate.query(var) != Some(SpecVal::FIRING) {

                // optimization. Don't bother trying this branch,

                // because the resulting branch would have an inconsistent predicate.

                }

                B::CouldntCheckFiring(port) => {

                    // sanity check: `CouldntCheckFiring` returned IFF the variable is speculatively assigned

                    assert!(predicate.query(var).is_none());

                    // speculate on the two possible values of `var`. Schedule both branches to be rerun.

                    drainer.add_input(predicate.clone().inserted(var, SpecVal::SILENT), branch.clone());

                }

                B::PutMsg(putter, payload) => {

                    // sanity check: The given port indeed has `Putter` polarity

                    // assign FIRING to this port's associated firing variable

                    let was = predicate.assigned.insert(var, SpecVal::FIRING);

                    if was == Some(SpecVal::SILENT) {

                        // Discard the branch, as it clearly has contradictory requirements for this value.

                B::SyncBlockEnd => {

                    // This branch reached the end of it's synchronous block

                    // assign all variables of owned ports that DIDN'T fire to SILENT

                        let actually_exchanged = branch.inner.did_put_or_get.contains(port);

                        let val = *predicate.assigned.entry(var).or_insert(SpecVal::SILENT);

                        let speculated_to_fire = val == SpecVal::FIRING;

        BranchingProtoComponent::drain_branches_to_blocked(cd, cu, rctx, proto_component_id)?;

        // swap the blocked branches back

        std::mem::swap(blocked.0, &mut self.branches);

        Ok(())

    }

        for port in moved_ports.iter() {

            assert_eq!(

                self.proto_component_id,

            );

        }

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

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

        // Update the ownership of the moved ports

        for port in moved_ports.iter() {

        }

    }

        // 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()];

            o,

            PortInfo {

                route: Route::LocalComponent,

                owner: self.proto_component_id,

            },

        );

            i,

            PortInfo {

                route: Route::LocalComponent,

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

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

    }

use super::*;

struct MonitoredReader<R: Read> {

    bytes: usize,

    r: R,

}

enum PollAndPopulateError {

    PollFailed,

    Timeout,

}

struct TryRecvAnyNetError {

    error: NetEndpointError,

    index: usize,

}

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

impl NetEndpoint {

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

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

    }

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

        &mut self,

        logger: &mut dyn Logger,

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

        use NetEndpointError as Nee;

        let before_len = self.inbox.len();

        'read_loop: loop {

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

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

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

        );

        let mut monitored = MonitoredReader::from(&self.inbox[..]);

        use bincode::config::Options;

        match Self::bincode_opts().deserialize_from(&mut monitored) {

            Ok(msg) => {

                let msg_size = monitored.bytes_read();

                log!(

                    @ENDPT,

                    logger,

            }

            Err(e) => match *e {

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

                    Ok(None)

                }

                _ => Err(Nee::MalformedMessage),

            },

        }

    }

    pub(super) fn send<T: serde::ser::Serialize>(

        &mut self,

        msg: &T,

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

        self.net_endpoint_store.endpoint_exts.len()

    }

    pub(super) fn send_to_comms(

        &mut self,

        index: usize,

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

        net_endpoint.send(msg).map_err(|_| Use::BrokenNetEndpoint { index })

    }

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

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

        logger: &mut dyn Logger,

        deadline: &Option<Instant>,

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

        fn map_trane(

            trane: TryRecvAnyNetError,

            net_endpoint_store: &EndpointStore<NetEndpointExt>,

                trane.error,

            )

        }

        // try yield undelayed net message

        if let Some(tup) = self.undelayed_messages.pop() {

            log!(@ENDPT, logger, "RECV undelayed_msg {:?}", &tup);

        round_index: usize,

    ) -> Result<CommRecvOk, UnrecoverableSyncError> {

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

        impl EndpointManager {

            fn handle_msg(

                &mut self,

                cu: &mut impl CuUndecided,

                some_message_enqueued: &mut bool,

            ) -> Option<(usize, CommCtrlMsg)> {

                let comm_msg_contents = match msg {

                    Msg::CommMsg(comm_msg) => match comm_msg.round_index.cmp(&round_index) {

                        Ordering::Less => {

                            log!(

                                cu.logger(),

                                "We are in round {}, but msg is for round {}. Discard",

                            return None;

                        }

                        Ordering::Greater => {

                            log!(

                                cu.logger(),

                                "We are in round {}, but msg is for round {}. Buffer",

                        }

                    },

                };

                match comm_msg_contents {

                    CommMsgContents::SendPayload(send_payload_msg) => {

                        let getter =

                            self.net_endpoint_store.endpoint_exts[net_index].getter_for_incoming;

                        rctx.getter_push(getter, send_payload_msg);

        use {PollAndPopulateError as Pape, UnrecoverableSyncError as Use};

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

        let mut some_message_enqueued = false;

        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)

            }

        }

        loop {

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

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

                    cu,

                    self.udp_endpoint_store.polled_undrained.insert(index);

                    if !ee.received_this_round {

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

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

                        rctx.getter_push(

                            ee.getter_for_incoming,

            }

        }

    }

    fn try_recv_undrained_net(

        &mut self,

        logger: &mut dyn Logger,

        }

        Ok(None)

    }

    fn poll_and_populate(

        &mut self,

        logger: &mut dyn Logger,

        self.poll.poll(&mut self.events, remaining).map_err(|_| Pape::PollFailed)?;

        for event in self.events.iter() {

            match TokenTarget::from(event.token()) {

                TokenTarget::NetEndpoint { index } => {

                    self.net_endpoint_store.polled_undrained.insert(index);

                    log!(

        self.events.clear();

        Ok(())

    }

    pub(super) fn undelay_all(&mut self) {

        if self.undelayed_messages.is_empty() {

            // fast path

        // slow path

        self.undelayed_messages.extend(self.delayed_messages.drain(..));

    }

    pub(super) fn udp_endpoints_round_end(

        &mut self,

        logger: &mut dyn Logger,

        );

        use UnrecoverableSyncError as Use;

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

            for (index, ee) in self.udp_endpoint_store.endpoint_exts.iter_mut().enumerate() {

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

                    if payload_predicate.assigns_subset(solution_predicate) {

use super::*;

fn secs_since_unix_epoch() -> f64 {

    std::time::SystemTime::now()

        .duration_since(std::time::UNIX_EPOCH)

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

struct SessionInfo {

    serde_proto_description: SerdeProtocolDescription,

    endpoint_incoming_to_getter: Vec<PortId>,

    proto_components: HashMap<ComponentId, ComponentState>,

}

    owner: ComponentId,

}

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

    id_manager: IdManager,

}

enum TokenTarget {

    NetEndpoint { index: usize },

    UdpEndpoint { index: usize },

}

// Returned by the endpoint manager as a result of comm_recv, telling the connector what happened,

        self.vec.pop()

    }

}

    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.

        SpecVar(match info.polarity {

            Getter => port,

            Putter => info.peer.unwrap(),

}

impl Drop for Connector {

    fn drop(&mut self) {

    }

}

// Given a slice of ports, return the first, if any, port is present repeatedly

        // - they are each others' peers

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

        // - polarity putter, getter respectively

            o,

            PortInfo {

                route: Route::LocalComponent,

                polarity: Putter,

            },

        );

            i,

            PortInfo {

                route: Route::LocalComponent,

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

        }

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

            if info.owner != cu.native_component_id {

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

            }

            .insert(new_cid, cu.proto_description.new_main_component(identifier, ports));

        // update the ownership of moved ports

        for port in ports.iter() {

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

                None => unreachable!(),

            }

    // buffer a message along with the ID of the putter who sent it

    fn putter_push(&mut self, cu: &mut impl CuUndecided, putter: PortId, msg: SendPayloadMsg) {

            log!(cu.logger(), "Putter add (putter:{:?} => getter:{:?})", putter, getter);

            self.getter_push(getter, msg);

        } else {

use crate::common::*;

use crate::runtime::*;

impl TokenTarget {

    const HALFWAY_INDEX: usize = usize::MAX / 2;

    const MAX_INDEX: usize = usize::MAX;

}

impl From<Token> for TokenTarget {

    fn from(Token(index): Token) -> Self {

            TokenTarget::UdpEndpoint { index: shifted }

        } else {

            TokenTarget::NetEndpoint { index }

impl Into<Token> for TokenTarget {

    fn into(self) -> Token {

        match self {

            TokenTarget::UdpEndpoint { index } => Token(index + Self::HALFWAY_INDEX),

            TokenTarget::NetEndpoint { index } => Token(index),

        }

            })),

        }

    }

    /// Conceptually, this returning [p0, g1] is sugar for:

    /// 1. create port pair [p0, g0]

    /// 2. create port pair [p1, g1]

            ConnectorPhased::Setup(setup) => {

                let udp_index = setup.udp_endpoint_setups.len();

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

                let mut npid = || cu.ips.id_manager.new_port_id();

                let [nin, nout, uin, uout] = [npid(), npid(), npid(), npid()];

                    nout,

                    PortInfo {

                        route: Route::LocalComponent,

                        owner: cu.native_component_id,

                    },

                );

                    uin,

                    PortInfo {

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

                        owner: udp_cid,

                    },

                );

                    uout,

                    PortInfo {

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

                        owner: udp_cid,

                    },

                );

                setup.udp_endpoint_setups.push(UdpEndpointSetup {

                    local_addr,

                    peer_addr,

                    getter_for_incoming: nin,

                });

                Ok([nout, nin])

            }

        }

        match phased {

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

            ConnectorPhased::Setup(setup) => {

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

                    new_pid,

                    PortInfo {

                        route: Route::LocalComponent,

                    &sock_addr,

                    endpoint_polarity

                );

                setup.net_endpoint_setups.push(NetEndpointSetup {

                    sock_addr,

                    endpoint_polarity,

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

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

                // connect all endpoints in parallel; send and receive peer ids through ports

                    &mut *cu.logger,

                    &setup.net_endpoint_setups,

                    &setup.udp_endpoint_setups,

                    &deadline,

                )?;

                log!(

                    &cu.ips.port_info,

                    &endpoint_manager,

                );

                let neighborhood = init_neighborhood(

                    cu.ips.id_manager.connector_id,

                    &mut *cu.logger,

                    &deadline,

                )?;

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

                let mut comm = ConnectorCommunication {

                    round_index: 0,

                    endpoint_manager,

                    neighborhood,

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

                };

                if cfg!(feature = "session_optimization") {

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

                }

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

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

                Ok(())

            }

        }

    }

}

    logger: &mut dyn Logger,

    net_endpoint_setups: &[NetEndpointSetup],

    udp_endpoint_setups: &[UdpEndpointSetup],

    deadline: &Option<Instant>,

    use ConnectError as Ce;

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

        // becomes completed once sent_local_port && recv_peer_port.is_some()

        // we send local port if we haven't already and we receive a writable event

        // we recv peer port if we haven't already and we receive a readbale event

        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 {

        // becomes completed once we receive our first writable event

        getter_for_incoming: PortId,

        sock: UdpSocket,

    }

    }

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

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

    let mut events =

        Events::with_capacity((net_endpoint_setups.len() + udp_endpoint_setups.len()) * 2 + 4);

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

    let mut delayed_messages = vec![];

    let mut net_todos = net_endpoint_setups

        .iter()

        .enumerate()

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

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

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

            } 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();

            };