}

    }};

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

        // ignore

    }};

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

    }};

}

    // this pair acts as SubtreeId -> HashSet<Predicate> which is friendlier to iteration

    subtree_solutions: Vec<HashSet<Predicate>>,

    subtree_id_to_index: HashMap<SubtreeId, usize>,

}

#[derive(Debug)]

struct BranchingProtoComponent {

    branches: HashMap<Predicate, ProtoComponentBranch>,

}

#[derive(Debug, Clone)]

struct ProtoComponentBranch {

    state: ComponentState,

    inner: ProtoComponentBranchInner,

        &mut self,

        port: PortId,

        expect_polarity: Polarity,

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

        use PortOpError as Poe;

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

            return Err(Poe::PortUnavailable);

        }

        }

        match phased {

            ConnectorPhased::Setup { .. } => Err(Poe::NotConnected),

            ConnectorPhased::Communication(comm) => {

                let batch = comm.native_batches.last_mut().unwrap(); // length >= 1 is invariant

                Ok(batch)

        log!(@BENCH, cu.inner.logger, "");

        // 1. run all proto components to Nonsync blockers

        // iterate

        let mut branching_proto_components =

            HashMap::<ComponentId, BranchingProtoComponent>::default();

            .proto_components

            .iter()

            .map(|(&proto_id, proto)| (proto_id, proto.clone()))

            .collect();

        log!(cu.inner.logger, "Nonsync running {} proto components...", unrun_components.len());

        // drains unrun_components, and populates branching_proto_components.

                unrun_components.len()

            );

            let mut ctx = NonsyncProtoContext {

                cu_inner: &mut cu.inner,

                proto_component_id,

                unrun_components: &mut unrun_components,

            };

            log!(

                cu.inner.logger,

                "proto component {:?} ran to nonsync blocker {:?}",

                proto_component_id,

                &blocker

            );

                    cu.inner.logger,

                    "Children in subtree are: {:?}",

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

                );

                SolutionStorage::new(subtree_id_iter)

            },

            getter_buffer: Default::default(),

            deadline: timeout.map(|to| Instant::now() + to),

        };

        log!(cu.inner.logger, "Round context structure initialized");

        log!(@BENCH, cu.inner.logger, "");

        'native_branches: for ((native_branch, index), branch_spec_val) in

            comm.native_batches.drain(..).zip(0..).zip(SpecVal::iter_domain())

        {

            let NativeBatch { to_get, to_put } = native_branch;

            let predicate = {

                let mut predicate = Predicate::default();

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

                }

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

                        log!(cu.inner.logger, "Native branch index={} contains internal inconsistency wrt. {:?}. Skipping", index, var);

                        continue 'native_branches;

                    }

                }

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

                    cu.inner.logger,

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

                    index,

                    &msg,

                    putter

                );

                rctx.getter_buffer.putter_add(cu, putter, msg);

            }

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

            if branch.is_ended() {

                log!(

                    cu.inner.logger,

        log!(

            cu.inner.logger,

            "Running all {} proto components to their sync blocker...",

            branching_proto_components.len()

        );

        for (&proto_component_id, proto_component) in branching_proto_components.iter_mut() {

            // must reborrow to constrain the lifetime of pcb_temps to inside the loop

            let (swap, pcb_temps) = pcb_temps.reborrow().split_first_mut();

            let (blocked, _pcb_temps) = pcb_temps.split_first_mut();

            // initially, no components have .ended==true

            // drain from branches --> blocked

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

            // swap the blocked branches back

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

            if branches.is_empty() {

                log!(cu.inner.logger, "{:?} has become inconsistent!", proto_component_id);

                if let Some(parent) = comm.neighborhood.parent {

                    if already_requested_failure.replace_with_true() {

            log!(

                cu.inner.logger,

                "Decision loop! have {} messages to recv",

                rctx.getter_buffer.len()

            );

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

                log!(

                    cu.inner.logger,

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

                    &send_payload_msg,

                    getter,

                );

                        let udp_endpoint_ext =

                        let SendPayloadMsg { predicate, payload } = send_payload_msg;

                        log!(cu.inner.logger, "Delivering to udp endpoint index={}", index);

                        udp_endpoint_ext.outgoing_payloads.insert(predicate, payload);

                    }

                        log!(@MARK, cu.inner.logger, "sending payload");

                        let msg = Msg::CommMsg(CommMsg {

                            round_index: comm.round_index,

                            contents: CommMsgContents::SendPayload(send_payload_msg),

                        });

                    }

                        branching_native.feed_msg(

                            cu,

                            rctx,

                            getter,

                            &send_payload_msg,

                            MapTempGuard::new(&mut bn_temp_owner),

                        )

                    }

                            branching_component.feed_msg(

                                cu,

                                rctx,

                                cid,

                                getter,

                                &send_payload_msg,

                        } else {

                            log!(

                                cu.inner.logger,

                                "Delivery to getter {:?} msg {:?} failed because {:?} isn't here",

                                getter,

                                &send_payload_msg,

                            );

                        }

                    }

                }

            }

            log!(cu.inner.logger, "No decision yet. Let's recv an endpoint msg...");

            {

                let (net_index, comm_ctrl_msg): (usize, CommCtrlMsg) = match comm

                    .endpoint_manager

                    .try_recv_any_comms(

                    &mut *cu.inner.logger,

                    rctx,

                    comm.round_index,

                )? {

                    CommRecvOk::NewControlMsg { net_index, msg } => (net_index, msg),

                    CommRecvOk::NewPayloadMsgs => continue 'undecided,

                    CommRecvOk::TimeoutWithoutNew => {

        round_ctx: &mut RoundCtx,

        getter: PortId,

        send_payload_msg: &SendPayloadMsg,

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

    ) {

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

        let mut draining = bn_temp;

        let finished = &mut self.branches;

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

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

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

            // check if this branch expects to receive it

            let mut feed_branch = |branch: &mut NativeBranch, predicate: &Predicate| {

                branch.to_get.remove(&getter);

                let was = branch.gotten.insert(getter, send_payload_msg.payload.clone());

                assert!(was.is_none());

                if branch.is_ended() {

                    log!(

impl BranchingProtoComponent {

    fn drain_branches_to_blocked(

        cd: CyclicDrainer<Predicate, ProtoComponentBranch>,

        cu: &mut ConnectorUnphased,

        rctx: &mut RoundCtx,

        proto_component_id: ComponentId,

    ) -> Result<(), UnrecoverableSyncError> {

        cd.cyclic_drain(|mut predicate, mut branch, mut drainer| {

            let mut ctx = SyncProtoContext {

                cu_inner: &mut cu.inner,

                predicate: &predicate,

                branch_inner: &mut branch.inner,

                    // move to "blocked"

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

                    drainer.add_output(predicate, branch);

                }

                B::CouldntCheckFiring(port) => {

                    // sanity check

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

                    // keep forks in "unblocked"

                    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

                    // overwrite assignment

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

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

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

                            proto_component_id, putter, var);

                        // discard forever

                        drop((predicate, branch));

                        rctx.getter_buffer.putter_add(cu, putter, msg);

                        drainer.add_input(predicate, branch);

                    }

                }

                B::SyncBlockEnd => {

                    // make concrete all variables

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

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

                        let did_fire = val == SpecVal::FIRING;

                        if did_fire != should_have_fired {

                                port, var, val, did_fire, should_have_fired);

                            // IMPLICIT inconsistency

                            drop((predicate, branch));

                            return Ok(());

                        }

                    }

                    drainer.add_output(predicate, branch);

                }

            }

            Ok(())

        })

    }

    fn feed_msg(

        &mut self,

        cu: &mut ConnectorUnphased,

        rctx: &mut RoundCtx,

        proto_component_id: ComponentId,

        getter: PortId,

            logger,

            "feeding proto component {:?} getter {:?} {:?}",

            proto_component_id,

            getter,

            &send_payload_msg

        );

        let (mut unblocked, pcb_temps) = pcb_temps.split_first_mut();

        let (mut blocked, pcb_temps) = pcb_temps.split_first_mut();

        // partition drain from branches -> {unblocked, blocked}

        log!(logger, "visiting {} blocked branches...", branches.len());

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

            if branch.ended {

            }

        }

        log!(logger, "blocked {:?} unblocked {:?}", blocked.len(), unblocked.len());

        // drain from unblocked --> blocked

        let (swap, _pcb_temps) = pcb_temps.split_first_mut();

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

        // swap the blocked branches back

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

        log!(cu.inner.logger, "component settles down with branches: {:?}", branches.keys());

        Ok(())

    }

    fn insert_branch_merging(

                for (k, v) in branch.inner.inbox.drain() {

                    old.inner.inbox.insert(k, v);

                }

            }

        }

    }

        for (branch_predicate, branch) in branches {

            if branch.ended && branch_predicate.assigns_subset(solution_predicate) {

                let ProtoComponentBranch { state, .. } = branch;

            }

        }

        panic!("ProtoComponent had no branches matching pred {:?}", solution_predicate);

    }

        let branch = ProtoComponentBranch { state, inner: Default::default(), ended: false };

    }

}

impl SolutionStorage {

    fn new(subtree_ids: impl Iterator<Item = SubtreeId>) -> Self {

        let mut subtree_id_to_index: HashMap<SubtreeId, usize> = Default::default();

        let mut subtree_solutions = vec![];

        self.getters_and_sends.pop()

    }

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

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

    }

    fn putter_add(&mut self, cu: &mut ConnectorUnphased, putter: PortId, msg: SendPayloadMsg) {

            self.getter_add(getter, msg);

        } else {

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

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

        }

    }

}

impl SyncProtoContext<'_> {

    pub(crate) fn is_firing(&mut self, port: PortId) -> Option<bool> {

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

    }

    pub(crate) fn read_msg(&mut self, port: PortId) -> Option<&Payload> {

        self.branch_inner.did_put_or_get.insert(port);

        self.branch_inner.inbox.get(&port)

    }

            self.cu_inner.logger,

            "Component {:?} added new component with state {:?}, moving ports {:?}",

            self.proto_component_id,

            &state,

            &moved_ports

        );

        for port in moved_ports.iter() {

        }

        // 3. create a new component

    }

    pub fn new_port_pair(&mut self) -> [PortId; 2] {

        // adds two new associated ports, related to each other, and exposed to the proto component

        log!(

            self.cu_inner.logger,

            "Component {:?} port pair (out->in) {:?} -> {:?}",

            self.proto_component_id,

            o,

            i

    // enqueues all current round SendPayload messages using round_ctx.getter_add

    // returns the first comm_ctrl_msg encountered

    // only polls until SOME message is enqueued

    pub(super) fn try_recv_any_comms(

        &mut self,

        logger: &mut dyn Logger,

        round_ctx: &mut impl RoundCtxTrait,

        round_index: usize,

    ) -> Result<CommRecvOk, UnrecoverableSyncError> {

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

        impl EndpointManager {

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

                        round_ctx.getter_add(

                            ee.getter_for_incoming,

                            SendPayloadMsg { payload, predicate },

                        );

                        some_message_enqueued = true;

#[derive(Debug)]

pub struct VecLogger(ConnectorId, Vec<u8>);

#[derive(Debug)]

pub struct DummyLogger;

#[derive(Debug)]

pub struct FileLogger(ConnectorId, std::fs::File);

pub(crate) struct NonsyncProtoContext<'a> {

    cu_inner: &'a mut ConnectorUnphasedInner, // persists between rounds

    proto_component_id: ComponentId,          // KEY in id->component map

}

pub(crate) struct SyncProtoContext<'a> {

    cu_inner: &'a mut ConnectorUnphasedInner, // persists between rounds

    branch_inner: &'a mut ProtoComponentBranchInner, // sub-structure of component branch

    predicate: &'a Predicate,                 // KEY in pred->branch map

struct VecSet<T: std::cmp::Ord> {

    // invariant: ordered, deduplicated

    vec: Vec<T>,

}

#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]

enum Route {

    NetEndpoint { index: usize },

    UdpEndpoint { index: usize },

}

#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]

enum SubtreeId {

    LocalComponent(ComponentId),

    NetEndpoint { index: usize },

}

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

struct MyPortInfo {

    polarity: Polarity,

    port: PortId,

}

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

enum Decision {

    Failure,

    Success(Predicate),

}

    SessionGather { unoptimized_map: HashMap<ConnectorId, SessionInfo> },

    SessionScatter { optimized_map: HashMap<ConnectorId, SessionInfo> },

}

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

struct SessionInfo {

    serde_proto_description: SerdeProtocolDescription,

    endpoint_incoming_to_getter: Vec<PortId>,

}

#[derive(Debug, Clone)]

struct SerdeProtocolDescription(Arc<ProtocolDescription>);

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

struct CommMsg {

    round_index: usize,

    Nonexistant,

}

struct NetEndpoint {

    inbox: Vec<u8>,

    stream: TcpStream,

}

#[derive(Debug, Clone)]

struct NetEndpointSetup {

    getter_for_incoming: PortId,

    sock_addr: SocketAddr,

    endpoint_polarity: EndpointPolarity,

}

}

#[derive(Debug)]

struct EndpointStore<T> {

    endpoint_exts: Vec<T>,

    polled_undrained: VecSet<usize>,

}

struct PortInfo {

}

#[derive(Debug)]

struct ConnectorCommunication {

    round_index: usize,

    endpoint_manager: EndpointManager,

    neighborhood: Neighborhood,

    native_batches: Vec<NativeBatch>,

    round_result: Result<Option<RoundOk>, SyncError>,

}

#[derive(Debug)]

struct ConnectorUnphased {

    proto_description: Arc<ProtocolDescription>,

    inner: ConnectorUnphasedInner,

}

#[derive(Debug)]

struct ConnectorUnphasedInner {

    logger: Box<dyn Logger>,

    native_component_id: ComponentId,

}

#[derive(Debug)]

struct ConnectorSetup {

    net_endpoint_setups: Vec<NetEndpointSetup>,

    udp_endpoint_setups: Vec<UdpEndpointSetup>,

        self.vec.iter()

    }

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

        self.vec.pop()

    }

}

    fn spec_var_for(&self, port: PortId) -> SpecVar {

            Getter => port,

        })

    }

}

impl SpecVarStream {

    fn next(&mut self) -> SpecVar {

        let phantom_port: PortId =

    pub fn get_logger(&mut self) -> &mut dyn Logger {

        &mut *self.unphased.inner.logger

    }

    pub fn new_port_pair(&mut self) -> [PortId; 2] {

        let cu = &mut self.unphased;

        // adds two new associated ports, related to each other, and exposed to the native

        log!(cu.inner.logger, "Added port pair (out->in) {:?} -> {:?}", o, i);

        [o, i]

    }

    pub fn add_component(

        &mut self,

        identifier: &[u8],

        ports: &[PortId],

    ) -> Result<(), AddComponentError> {

        // called by the USER. moves ports owned by the NATIVE

        use AddComponentError as Ace;

        // 1. check if this is OK

        let cu = &mut self.unphased;

        }

            }

            }

        }

        for port in ports.iter() {

        Ok(())

    }

}

impl Predicate {

    #[inline]

    pub fn singleton(k: SpecVar, v: SpecVal) -> Self {

        mut logger: Box<dyn Logger>,

        proto_description: Arc<ProtocolDescription>,

        connector_id: ConnectorId,

    ) -> Self {

        log!(&mut *logger, "Created with connector_id {:?}", connector_id);

        let mut id_manager = IdManager::new(connector_id);

        Self {

            unphased: ConnectorUnphased {