Arc_ProtocolDescription * pd = protocol_description_parse("", 0);

	char logpath[] = "./3_16_getter.txt";

	Connector * c = connector_new_logging_with_id(pd, logpath, sizeof(logpath)-1, 0);

	rw_err_peek(c);

	PortId getter;

	rw_err_peek(c);

	connector_add_net_port(c, &getter, addr, Polarity_Getter, EndpointPolarity_Passive);

	connector_connect(c, -1);

	rw_err_peek(c);

	int i;

	Arc_ProtocolDescription * pd = protocol_description_parse("", 0);

	char logpath[] = "./3_16_putter.txt";

	Connector * c = connector_new_logging_with_id(pd, logpath, sizeof(logpath)-1, 1);

	rw_err_peek(c);

	PortId putter;

	rw_err_peek(c);

	connector_add_net_port(c, &putter, addr, Polarity_Putter, EndpointPolarity_Active);

	connector_connect(c, -1);

	rw_err_peek(c);

	// Prepare a message to send

}

pub type ConnectorId = u32;

pub type U32Suffix = u32;

#[derive(

    Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,

)]

#[derive(

    Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,

)]

#[repr(C)]

pub struct Id {

    pub(crate) connector_id: ConnectorId,

}

impl IdParts for PortId {

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

        self.0.id_parts()

    }

}

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

        self.0.id_parts()

    }

}

impl U32Stream {

    pub(crate) fn next(&mut self) -> u32 {

}

impl From<Id> for PortId {

    fn from(id: Id) -> PortId {

        Self(id)

    }

}

        Self(id)

    }

}

impl From<&[u8]> for Payload {

    fn from(s: &[u8]) -> Payload {

        Payload(Arc::new(s.to_vec()))

impl Debug for PortId {

    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {

        let (a, b) = self.id_parts();

        write!(f, "pid{}_{}", a, b)

    }

}

    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {

        let (a, b) = self.id_parts();

        write!(f, "cid{}_{}", a, b)

    }

}

impl Debug for Payload {

/*

Change the definition of these macros to control the logging level statically

*/

macro_rules! log {

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

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

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

        }

    }};

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

        // ignore

                    BinaryOperator::Inequality => Ok(left.neq(&right)),

                    BinaryOperator::LessThan => Ok(left.lt(&right)),

                    BinaryOperator::LessThanEqual => Ok(left.lte(&right)),

                    BinaryOperator::GreaterThan => Ok(left.gt(&right)),

                    BinaryOperator::GreaterThanEqual => Ok(left.gte(&right)),

                    BinaryOperator::Remainder => Ok(left.modulus(&right)),

                    _ => unimplemented!("{:?}", expr.operation),

                }

            }

            Expression::Unary(expr) => {

                let mut value = self.eval(h, ctx, expr.expression)?;

                match expr.operation {

        comm: &mut ConnectorCommunication,

        timeout: Option<Duration>,

    ) -> Result<Option<RoundOk>, SyncError> {

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

        use SyncError as Se;

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

        log!(

            cu.inner.logger,

            "~~~ SYNC called with timeout {:?}; starting round {}",

            &timeout,

            comm.round_index

        );

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

        // 1. run all proto components to Nonsync blockers

        let mut branching_proto_components =

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

        // drains unrun_components, and populates branching_proto_components.

        while let Some((proto_component_id, mut component)) = unrun_components.pop() {

            log!(

                cu.inner.logger,

                "Nonsync running proto component with ID {:?}. {} to go after this",

                proto_component_id,

                unrun_components.len()

            );

        );

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

        // Create temp structures needed for the synchronous phase of the round

        let mut rctx = RoundCtx {

            solution_storage: {

                let e = comm

                    .neighborhood

                    .children

                    .iter()

                    .map(|&index| SubtreeId::NetEndpoint { index });

                let subtree_id_iter = n.chain(c).chain(e);

                    "Native submitting solution for batch {} with {:?}",

                    index,

                    &predicate

                );

                rctx.solution_storage.submit_and_digest_subtree_solution(

                    &mut *cu.inner.logger,

                    predicate.clone(),

                );

            }

            if let Some(_) = branching_native.branches.insert(predicate, branch) {

                // thanks to the native_spec_var, each batch has a distinct predicate

                unreachable!()

            cu,

            comm,

            &mut branching_native,

            &mut branching_proto_components,

            &mut rctx,

        )?;

        log!(cu.inner.logger, "Committing to decision {:?}!", &decision);

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

        comm.endpoint_manager.udp_endpoints_round_end(&mut *cu.inner.logger, &decision)?;

        // propagate the decision to children

        let msg = Msg::CommMsg(CommMsg {

        log!(

            cu.inner.logger,

            "Announcing decision {:?} through child endpoints {:?}",

            &msg,

            &comm.neighborhood.children

        );

        for &child in comm.neighborhood.children.iter() {

            comm.endpoint_manager.send_to_comms(child, &msg)?;

        }

        let ret = match decision {

            Decision::Failure => {

                // dropping {branching_proto_components, branching_native}

    }

    fn sync_reach_decision(

        cu: &mut ConnectorUnphased,

        comm: &mut ConnectorCommunication,

        branching_native: &mut BranchingNative,

        rctx: &mut RoundCtx,

    ) -> Result<Decision, UnrecoverableSyncError> {

        let mut already_requested_failure = false;

        if branching_native.branches.is_empty() {

            log!(cu.inner.logger, "Native starts with no branches! Failure!");

            match comm.neighborhood.parent {

                Some(parent) => {

                    if already_requested_failure.replace_with_true() {

                None => {

                    log!(cu.inner.logger, "No parent. Deciding on failure");

                    return Ok(Decision::Failure);

                }

            }

        }

        let mut pcb_temps_owner = <[HashMap<Predicate, ProtoComponentBranch>; 3]>::default();

        let mut pcb_temps = MapTempsGuard(&mut pcb_temps_owner);

        let mut bn_temp_owner = <HashMap<Predicate, NativeBranch>>::default();

        // run all proto components to their sync blocker

        log!(

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

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

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

                log!(

                    cu.inner.logger,

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

                    &send_payload_msg,

                            &mut comm.endpoint_manager.udp_endpoint_store.endpoint_exts[*index];

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

                    }

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

                    }

                            branching_component.feed_msg(

                                cu,

                                rctx,

                                getter,

                                &send_payload_msg,

                                pcb_temps.reborrow(),

                            )?;

                            if branching_component.branches.is_empty() {

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

                                    if already_requested_failure.replace_with_true() {

                                        Self::request_failure(cu, comm, parent)?

                                    } else {

                                        log!(cu.inner.logger, "Already requested failure");

                                    }

            }

            // check if we have a solution yet

            log!(cu.inner.logger, "Check if we have any local decisions...");

            for solution in rctx.solution_storage.iter_new_local_make_old() {

                log!(cu.inner.logger, "New local decision with solution {:?}...", &solution);

                match comm.neighborhood.parent {

                    Some(parent) => {

                        log!(cu.inner.logger, "Forwarding to my parent {:?}", parent);

                        let suggestion = Decision::Success(solution);

                        let msg = Msg::CommMsg(CommMsg {

                            round_index: comm.round_index,

                    log!(

                        cu.inner.logger,

                        "new native solution with {:?} is_ended() with gotten {:?}",

                        &predicate,

                        &branch.gotten

                    );

                    round_ctx.solution_storage.submit_and_digest_subtree_solution(

                        &mut *cu.inner.logger,

                        subtree_id,

                        predicate.clone(),

                    );

                } else {

}

impl BranchingProtoComponent {

    fn drain_branches_to_blocked(

        cd: CyclicDrainer<Predicate, ProtoComponentBranch>,

        cu: &mut ConnectorUnphased,

        rctx: &mut RoundCtx,

        ports: &HashSet<PortId>,

    ) -> Result<(), UnrecoverableSyncError> {

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

            let mut ctx = SyncProtoContext {

                cu_inner: &mut cu.inner,

                predicate: &predicate,

                    // sanity check

                    assert_eq!(Some(&Putter), cu.inner.port_info.polarities.get(&putter));

                    // overwrite assignment

                    let var = cu.inner.port_info.spec_var_for(putter);

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

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

                        // discard forever

                        drop((predicate, branch));

                    } else {

                        // keep in "unblocked"

                        branch.inner.did_put_or_get.insert(putter);

                        let msg = SendPayloadMsg { predicate: predicate.clone(), payload };

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

                        drainer.add_input(predicate, branch);

                    }

                }

                B::SyncBlockEnd => {

                    for port in ports.iter() {

                        let var = cu.inner.port_info.spec_var_for(*port);

                        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 {

                            // IMPLICIT inconsistency

                            drop((predicate, branch));

                            return Ok(());

                        }

                    }

                    // submit solution for this component

                    rctx.solution_storage.submit_and_digest_subtree_solution(

                        &mut *cu.inner.logger,

                        subtree_id,

                        predicate.clone(),

                    );

                    branch.ended = true;

    //     a

    // }

    fn feed_msg(

        &mut self,

        cu: &mut ConnectorUnphased,

        rctx: &mut RoundCtx,

        getter: PortId,

        send_payload_msg: &SendPayloadMsg,

        pcb_temps: MapTempsGuard<'_, Predicate, ProtoComponentBranch>,

    ) -> Result<(), UnrecoverableSyncError> {

        let logger = &mut *cu.inner.logger;

        log!(

    }

}

impl<'a, K: Eq + Hash, V> CyclicDrainInner<'a, K, V> {

    fn add_input(&mut self, k: K, v: V) {

        self.swap.insert(k, v);

    }

    fn add_output(&mut self, k: K, v: V) {

        self.output.insert(k, v);

    }

}

impl NonsyncProtoContext<'_> {

    pub fn new_component(&mut self, moved_ports: HashSet<PortId>, state: ComponentState) {

            self.cu_inner.logger,

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

            self.proto_component_id,

            &state,

            &moved_ports

        );

        // 2. remove ports from old component & update port->route

        for port in moved_ports.iter() {

            self.proto_component_ports.remove(port);

        }

        // 3. create a new component

        self.unrun_components.push((new_id, ProtoComponent { state, ports: moved_ports }));

    }

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

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

        self.proto_component_ports.insert(i);

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

        self.cu_inner.port_info.polarities.insert(o, Putter);

        self.cu_inner.port_info.polarities.insert(i, Getter);

        self.cu_inner.port_info.peers.insert(o, i);

        self.cu_inner.port_info.peers.insert(i, o);

        self.cu_inner.port_info.routes.insert(o, route);

        self.cu_inner.port_info.routes.insert(i, route);

        log!(

            self.cu_inner.logger,

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

            self.proto_component_id,

        msg: &T,

    ) -> Result<(), NetEndpointError> {

        use bincode::config::Options;

        use NetEndpointError as Nee;

        Self::bincode_opts()

            .serialize_into(&mut self.stream, msg)

    }

}

impl EndpointManager {

    pub(super) fn index_iter(&self) -> Range<usize> {

        0..self.num_net_endpoints()

        }

        Ok(())

    }

}

impl Debug for NetEndpoint {

    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {

    }

}

impl<R: Read> From<R> for MonitoredReader<R> {

    fn from(r: R) -> Self {

        Self { r, bytes: 0 }

    }

    BrokenUdpEndpoint { index: usize },

    MalformedStateError(MalformedStateError),

}

#[derive(Debug, Clone)]

pub enum SyncError {

    NotConnected,

    RoundFailure,

    Unrecoverable(UnrecoverableSyncError),

}

#[derive(Debug, Clone)]

pub enum MalformedStateError {

    PortCannotPut(PortId),

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_ports: &'a mut HashSet<PortId>, // sub-structure of component

}

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

}

#[derive(Default)]

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 {

    LocalComponent(ComponentId),

    NetEndpoint { index: usize },

    UdpEndpoint { index: usize },

}

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

}

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

struct SessionInfo {

    serde_proto_description: SerdeProtocolDescription,

    port_info: PortInfo,

    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,

    children: VecSet<usize>,

}

#[derive(Debug)]

struct IdManager {

    connector_id: ConnectorId,

    port_suffix_stream: U32Stream,

}

struct UdpInBuffer {

    byte_vec: Vec<u8>,

}

#[derive(Debug)]

struct SpecVarStream {

    connector_id: ConnectorId,

struct EndpointStore<T> {

    endpoint_exts: Vec<T>,

    polled_undrained: VecSet<usize>,

}

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

struct PortInfo {

    polarities: HashMap<PortId, Polarity>,

    peers: HashMap<PortId, PortId>,

    routes: HashMap<PortId, Route>,

}

#[derive(Debug)]

struct ConnectorCommunication {

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

    id_manager: IdManager,

    native_ports: HashSet<PortId>,

    port_info: PortInfo,

}

#[derive(Debug)]

struct ConnectorSetup {

    net_endpoint_setups: Vec<NetEndpointSetup>,

    udp_endpoint_setups: Vec<UdpEndpointSetup>,

}

}

impl IdManager {

    fn new(connector_id: ConnectorId) -> Self {

        Self {

            connector_id,

            port_suffix_stream: Default::default(),

        }

    }

    fn new_spec_var_stream(&self) -> SpecVarStream {

        // Spec var stream starts where the current port_id stream ends, with gap of SKIP_N.

        // This gap is entirely unnecessary (i.e. 0 is fine)

        // It's purpose is only to make SpecVars easier to spot in logs.

        let port_suffix_stream = self.port_suffix_stream.clone().n_skipped(SKIP_N);

        SpecVarStream { connector_id: self.connector_id, port_suffix_stream }

    }

    fn new_port_id(&mut self) -> PortId {

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

    }

    }

}

impl Drop for Connector {

    fn drop(&mut self) {

        log!(&mut *self.unphased.inner.logger, "Connector dropping. Goodbye!");

    }

        cu.inner.native_ports.insert(i);

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

        cu.inner.port_info.polarities.insert(o, Putter);

        cu.inner.port_info.polarities.insert(i, Getter);

        cu.inner.port_info.peers.insert(o, i);

        cu.inner.port_info.peers.insert(i, o);

        cu.inner.port_info.routes.insert(o, route);

        cu.inner.port_info.routes.insert(i, route);

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

        [o, i]

    }

    pub fn add_component(

            }

            if expected_polarity != *cu.inner.port_info.polarities.get(port).unwrap() {

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

            }

        }

        // 3. remove ports from old component & update port->route

        for port in ports.iter() {

        }

        cu.inner.native_ports.retain(|port| !ports.contains(port));

        // 4. add new component

        cu.proto_components.insert(

            ProtoComponent {

                state: cu.proto_description.new_main_component(identifier, ports),

                ports: ports.iter().copied().collect(),

            },

        );

        Ok(())

impl UdpInBuffer {

    const CAPACITY: usize = u16::MAX as usize;

    fn as_mut_slice(&mut self) -> &mut [u8] {

        self.byte_vec.as_mut_slice()

    }

}

    pub fn new(

        mut logger: Box<dyn Logger>,

        proto_description: Arc<ProtocolDescription>,

        connector_id: ConnectorId,

    ) -> Self {

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

        Self {

            unphased: ConnectorUnphased {

                proto_description,

                proto_components: Default::default(),

                inner: ConnectorUnphasedInner {

                    logger,

                    native_ports: Default::default(),

                    port_info: Default::default(),

                },

            },

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

                net_endpoint_setups: Default::default(),

                cu.inner.port_info.polarities.insert(uin, Getter);

                cu.inner.port_info.polarities.insert(uout, Putter);

                cu.inner.port_info.peers.insert(nin, uout);

                cu.inner.port_info.peers.insert(nout, uin);

                cu.inner.port_info.peers.insert(uin, nout);

                cu.inner.port_info.peers.insert(uout, nin);

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

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

                setup.udp_endpoint_setups.push(UdpEndpointSetup {

                    local_addr,

                    peer_addr,

                    getter_for_incoming: nin,

                cu.inner.native_ports.insert(local_port);

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

                cu.inner.port_info.polarities.insert(local_port, polarity);

                cu.inner

                    .port_info

                    .routes

                log!(

                    cu.inner.logger,

                    "Added net port {:?} with polarity {:?} addr {:?} endpoint_polarity {:?}",

                    local_port,

                    polarity,

                    &sock_addr,

    c.sync(SEC1).unwrap();

    c.put(p0, Payload::from(b"HEY" as &[_])).unwrap();

    c.put(p1, Payload::from(b"HELLO" as &[_])).unwrap();

    c.sync(SEC1).unwrap_err();

}