SyncBlockEnd,

    CouldntReadMsg(PortId),

    CouldntCheckFiring(PortId),

    PutMsg(PortId, Payload),

    NondetChoice { n: u16 },

}

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

impl IdParts for Id {

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

        (self.connector_id, self.u32_suffix)

    }

}

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

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

        //     let _ = writeln!(w, $($arg)*);

        // }

    }};

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

    }};

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

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

        //     let _ = writeln!(w, $($arg)*);

        // }

    }};

        &mut self,

        port: PortId,

        expect_polarity: Polarity,

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

        }

        if info.polarity != expect_polarity {

            return Err(Poe::WrongPolarity);

        }

                    cu.logger(),

                    "New native with firing ports {:?}",

                    firing_iter.clone().collect::<Vec<_>>()

                );

                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;

                    }

                }

                // this branch is consistent. distinguish it with a unique var:val mapping and proceed

                predicate.inserted(native_spec_var, branch_spec_val)

            };

                    "Native branch {} sending msg {:?} with putter {:?}",

                    index,

                    &msg,

                    putter

                );

                // sanity check

                rctx.putter_push(cu, putter, msg);

            }

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

            if branch.is_ended() {

                // empty to_get set => already corresponds with a component solution

                log!(

        comm.endpoint_manager.undelay_all();

        'undecided: loop {

            // handle all buffered messages, sending them through endpoints / feeding them to components

            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!(

                    cu.logger(),

                    "Routing msg {:?} to {:?} via {:?}",

                    &send_payload_msg,

        rctx: &mut RoundCtx,

        getter: PortId,

        send_payload_msg: &SendPayloadMsg,

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

        // Visit all native's branches, and feed those whose current predicates are

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

                // the existing branch asserts the getter port is SILENT

                log!(

                    cu.logger(),

                    assert!(!branch.inner.inbox.contains_key(&port)); 

                    // This branch hit a proper blocker: progress awaits the receipt of some message. Exit the cycle.

                    drainer.add_output(predicate, branch);

                }

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

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

                }

                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.

                        log!(cu.logger(), "Proto component {:?} tried to PUT on port {:?} when pred said var {:?}==Some(false). inconsistent!",

                            proto_component_id, putter, var);

                        drop((predicate, branch));

                        drainer.add_input(predicate, branch);

                    }

                }

                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;

                        if actually_exchanged != speculated_to_fire {

                            log!(cu.logger(), "Inconsistent wrt. port {:?} var {:?} val {:?} actually_exchanged={}, speculated_to_fire={}",

                                port, var, val, actually_exchanged, speculated_to_fire);

        // drain from unblocked --> blocked

        let (swap, _pcb_temps) = pcb_temps.split_first_mut(); // peel off ONE temp storage map

        let cd = CyclicDrainer::new(unblocked.0, swap.0, blocked.0);

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

    }

    // Insert a new speculate branch into the given storage,

    // MERGING it with an existing branch if their predicate keys clash.

    fn insert_branch_merging(

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

            );

        }

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

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

        log!(

            self.logger,

            &state,

            &moved_ports

        );

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

        // Update the ownership of the moved ports

        for port in moved_ports.iter() {

        }

    }

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

            o,

            PortInfo {

                route: Route::LocalComponent,

                peer: Some(i),

                polarity: Putter,

                owner: self.proto_component_id,

            },

        );

            i,

            PortInfo {

                route: Route::LocalComponent,

                peer: Some(o),

                polarity: Getter,

                owner: self.proto_component_id,

    }

}

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

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

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

            match res {

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

                Ok(0) => break 'read_loop,

            @ENDPT,

            logger,

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

            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,

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

                    self.inbox.len() + msg_size,

                    msg_size,

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

                );

                Ok(Some(msg))

            }

            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,

    ) -> Result<(), NetEndpointError> {

        use bincode::config::Options;

        use NetEndpointError as Nee;

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

    }

    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;

        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,

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

        fn map_trane(

            trane: TryRecvAnyNetError,

            net_endpoint_store: &EndpointStore<NetEndpointExt>,

        ) -> ConnectError {

            ConnectError::NetEndpointSetupError(

                net_endpoint_store.endpoint_exts[trane.index]

                    .stream

                    .local_addr()

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

                trane.error,

            )

        }

        // try yield undelayed net message

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

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

            return Ok(tup);

        }

        loop {

        &mut self,

        cu: &mut impl CuUndecided,

        rctx: &mut RoundCtx,

        round_index: usize,

    ) -> Result<CommRecvOk, UnrecoverableSyncError> {

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

        impl EndpointManager {

            fn handle_msg(

                &mut self,

                cu: &mut impl CuUndecided,

                rctx: &mut RoundCtx,

                net_index: usize,

                msg: Msg,

                round_index: usize,

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

                                comm_msg.round_index,

                                round_index,

                            );

                            return None;

                        }

                        Ordering::Greater => {

                            log!(

                                cu.logger(),

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

                                comm_msg.round_index,

                                round_index,

                            );

                            self.delayed_messages.push((net_index, Msg::CommMsg(comm_msg)));

                            return None;

                        }

                    },

                };

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

                        *some_message_enqueued = true;

                        None

                    }

                }

            }

        }

        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)

            {

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

            }

        }

        loop {

            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,

                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 predicate = Predicate::singleton(port_spec_var, SpecVal::FIRING);

                        rctx.getter_push(

                            ee.getter_for_incoming,

                            SendPayloadMsg { payload, predicate },

                        );

                        some_message_enqueued = true;

                Ok(()) => {} // continue looping

                Err(Pape::Timeout) => return Ok(CommRecvOk::TimeoutWithoutNew),

                Err(Pape::PollFailed) => return Err(Use::PollFailed),

            }

        }

    }

    fn try_recv_undrained_net(

        &mut self,

        logger: &mut dyn Logger,

    ) -> Result<Option<(usize, Msg)>, TryRecvAnyNetError> {

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

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

                }

                return Ok(Some((index, msg)));

            }

        }

        Ok(None)

    }

    fn poll_and_populate(

        &mut self,

        logger: &mut dyn Logger,

        deadline: &Option<Instant>,

    ) -> Result<(), PollAndPopulateError> {

        use PollAndPopulateError as Pape;

            None

        };

        // Yes we do! Poll with remaining time as poll deadline

        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!(

                        @ENDPT,

                        logger,

                        "RECV poll event {:?} for NET endpoint index {:?}. undrained: {:?}",

                }

            }

        }

        self.events.clear();

        Ok(())

    }

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

        if self.undelayed_messages.is_empty() {

            // fast path

            std::mem::swap(&mut self.delayed_messages, &mut self.undelayed_messages);

            return;

        }

        // slow path

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

    }

    pub(super) fn udp_endpoints_round_end(

        &mut self,

        logger: &mut dyn Logger,

        decision: &Decision,

    ) -> Result<(), UnrecoverableSyncError> {

        // retain received_from_this_round for use in pseudo_socket_api::recv_from

            logger,

            "Ending round for {} udp endpoints",

            self.udp_endpoint_store.endpoint_exts.len()

        );

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

                        ee.sock.send(payload.as_slice()).map_err(|e| {

                            println!("{:?}", e);

                            Use::BrokenUdpEndpoint { index }

use super::*;

fn secs_since_unix_epoch() -> f64 {

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

        .duration_since(std::time::UNIX_EPOCH)

        .map(|dur| dur.as_secs_f64())

        .unwrap_or(0.)

}

// A data structure encoding the state of a connector, passed around

// during the session optimization procedure.

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

struct SessionInfo {

    serde_proto_description: SerdeProtocolDescription,

    endpoint_incoming_to_getter: Vec<PortId>,

    proto_components: HashMap<ComponentId, ComponentState>,

}

// Newtype wrapper for an Arc<ProtocolDescription>,

// such that it can be (de)serialized for transmission over the network.

struct MyPortInfo {

    polarity: Polarity,

    port: PortId,

    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,

}

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

#[derive(Debug)]

struct ConnectorCommunication {

// the way it identifies the evented structre it corresponds to.

// See runtime/setup for methods converting between TokenTarget and mio::Token

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

enum TokenTarget {

    NetEndpoint { index: usize },

    UdpEndpoint { index: usize },

}

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

// such that it can know when to continue polling, and when to block.

enum CommRecvOk {

    TimeoutWithoutNew,

        self.vec.iter()

    }

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

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

        })

    }

}

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

            .into()

    }

}

impl Drop for Connector {

    fn drop(&mut self) {

    }

}

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

fn duplicate_port(slice: &[PortId]) -> Option<PortId> {

    let mut vec = Vec::with_capacity(slice.len());

    for port in slice.iter() {

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

            o,

            PortInfo {

                route: Route::LocalComponent,

                peer: Some(i),

                owner: cu.native_component_id,

                polarity: Putter,

            },

        );

            i,

            PortInfo {

                route: Route::LocalComponent,

                peer: Some(o),

                owner: cu.native_component_id,

                polarity: Getter,

        }

        let expected_polarities = cu.proto_description.component_polarities(identifier)?;

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

            if info.owner != cu.native_component_id {

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

            }

            if info.polarity != expected_polarity {

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

            }

        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_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!(),

            }

        }

        Ok(())

    }

    fn getter_push(&mut self, getter: PortId, msg: SendPayloadMsg) {

        self.payload_inbox.push((getter, msg));

    }

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

            log!(cu.logger(), "Putter {:?} has no known peer!", putter);

            panic!("Putter {:?} has no known peer!");

        }

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

        }

    }

}

impl Connector {

            phased: ConnectorPhased::Setup(Box::new(ConnectorSetup {

                net_endpoint_setups: Default::default(),

                udp_endpoint_setups: Default::default(),

            })),

        }

    }

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

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

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

    /// 3. create udp component with interface of moved ports [p1, g0]

    /// 4. return [p0, g1]

    pub fn new_udp_mediator_component(

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

        match phased {

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

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