// P is in O's owned set

        for (port, info) in self.map.iter() {

            match self.owned.get(&info.owner) {

                Some(set) if set.contains(port) => {}

                _ => {

                    println!("{:#?}\n WITH port {:?}", self, port);

                    return false;

                }

            }

        }

        // for every port P owned by every owner O,

        // P's owner is O

        for (&owner, set) in self.owned.iter() {

            for port in set {

                match self.map.get(port) {

                    Some(info) if info.owner == owner => {}

                    _ => {

                        println!("{:#?}\n WITH owner {:?} port {:?}", self, owner, port);

                        return false;

                    }

                }

            }

        }

        true

    }

}

impl SpecVarStream {

    fn next(&mut self) -> SpecVar {

        let phantom_port: PortId =

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

                .into();

        SpecVar(phantom_port)

    }

}

impl IdManager {

    fn new(connector_id: ConnectorId) -> Self {

        Self {

            connector_id,

            port_suffix_stream: Default::default(),

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

        // E.g. spot the spec var: { v0_0, v1_2, v1_103 }

        const SKIP_N: u32 = 100;

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

    }

    fn new_component_id(&mut self) -> ComponentId {

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

            .into()

    }

}

impl Drop for Connector {

    fn drop(&mut self) {

        log!(self.unphased.logger(), "Connector dropping. Goodbye!");

    }

}

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

        match vec.binary_search(port) {

            Err(index) => vec.insert(index, *port),

            Ok(_) => return Some(*port),

        }

    }

    None

}

impl Connector {

    /// Generate a random connector identifier from the system's source of randomness.

    pub fn random_id() -> ConnectorId {

        type Bytes8 = [u8; std::mem::size_of::<ConnectorId>()];

        unsafe {

            let mut bytes = std::mem::MaybeUninit::<Bytes8>::uninit();

            // getrandom is the canonical crate for a small, secure rng

            getrandom::getrandom(&mut *bytes.as_mut_ptr()).unwrap();

            // safe! representations of all valid Byte8 values are valid ConnectorId values

            std::mem::transmute::<_, _>(bytes.assume_init())

        }

    }

    /// Returns true iff the connector is in connected state, i.e., it's setup phase is complete,

    /// and it is ready to participate in synchronous rounds of communication.

    pub fn is_connected(&self) -> bool {

        // If designed for Rust usage, connectors would be exposed as an enum type from the start.

        // consequently, this "phased" business would also include connector variants and this would

        // get a lot closer to the connector impl. itself.

        // Instead, the C-oriented implementation doesn't distinguish connector states as types,

        // and distinguish them as enum variants instead

        match self.phased {

            ConnectorPhased::Setup(..) => false,

            ConnectorPhased::Communication(..) => true,

        }

    }

    /// Enables the connector's current logger to be swapped out for another

    pub fn swap_logger(&mut self, mut new_logger: Box<dyn Logger>) -> Box<dyn Logger> {

        std::mem::swap(&mut self.unphased.logger, &mut new_logger);

        new_logger

    }

    /// Access the connector's current logger

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

        &mut *self.unphased.logger

    }

    /// Create a new synchronous channel, returning its ends as a pair of ports,

    /// with polarity output, input respectively. Available during either setup/communication phase.

    /// # Panics

    /// This function panics if the connector's (large) port id space is exhausted.

    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

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

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

        cu.ips.port_info.map.insert(

            o,

            PortInfo {

                route: Route::LocalComponent,

                peer: Some(i),

                owner: cu.native_component_id,

                polarity: Putter,

            },

        );

        cu.ips.port_info.map.insert(

            i,

            PortInfo {

                route: Route::LocalComponent,

                peer: Some(o),

                owner: cu.native_component_id,

                polarity: Getter,

            },

        );

        cu.ips

            .port_info

            .owned

            .entry(cu.native_component_id)

            .or_default()

            .extend([o, i].iter().copied());

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

        [o, i]

    }

    /// Instantiates a new component for the connector runtime to manage, and passing

    /// the given set of ports from the interface of the native component, to that of the

    /// newly created component (passing their ownership).

    /// # Errors

    /// Error is returned if the moved ports are not owned by the native component,

    /// if the given component name is not defined in the connector's protocol,

    /// the given sequence of ports contains a duplicate port,

    /// or if the component is unfit for instantiation with the given port sequence.

    /// # Panics

    /// This function panics if the connector's (large) component id space is exhausted.

    pub fn add_component(

        &mut self,

        identifier: &[u8],

        ports: &[PortId],

    ) -> Result<(), AddComponentError> {

        // Check for error cases first before modifying `cu`

        use AddComponentError as Ace;

        let cu = &self.unphased;

        if let Some(port) = duplicate_port(ports) {

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

        }

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

            let info = cu.ips.port_info.map.get(&port).ok_or(Ace::UnknownPort(port))?;

            if info.owner != cu.native_component_id {

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

            }

            if info.polarity != expected_polarity {

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

            }

        }

        // No errors! Time to modify `cu`

        // create a new component and identifier

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

        cu.proto_components.insert(new_cid, cu.proto_description.new_component(identifier, ports));

        // update the ownership of moved ports

        for port in ports.iter() {

            match cu.ips.port_info.map.get_mut(port) {

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

                None => unreachable!(),

            }

        }

        if let Some(set) = cu.ips.port_info.owned.get_mut(&cu.native_component_id) {

            set.retain(|x| !ports.contains(x));

        }

        Ok(())

    }

}

impl Predicate {

    #[inline]

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

        Self::default().inserted(k, v)

    }

    #[inline]

    pub fn inserted(mut self, k: SpecVar, v: SpecVal) -> Self {

        self.assigned.insert(k, v);

        self

    }

    // Return true whether `self` is a subset of `maybe_superset`

    pub fn assigns_subset(&self, maybe_superset: &Self) -> bool {

        for (var, val) in self.assigned.iter() {

            match maybe_superset.assigned.get(var) {

                Some(val2) if val2 == val => {}

                _ => return false, // var unmapped, or mapped differently

            }

        }

        // `maybe_superset` mirrored all my assignments!

        true

    }

    /// Given the two predicates {self, other}, return that whose

    /// assignments are the union of those of both.

    fn assignment_union(&self, other: &Self) -> AssignmentUnionResult {

        use AssignmentUnionResult as Aur;

        // iterators over assignments of both predicates. Rely on SORTED ordering of BTreeMap's keys.

        let [mut s_it, mut o_it] = [self.assigned.iter(), other.assigned.iter()];

        let [mut s, mut o] = [s_it.next(), o_it.next()];

        // populate lists of assignments in self but not other and vice versa.

        // do this by incrementally unfolding the iterators, keeping an eye

        // on the ordering between the head elements [s, o].

        // whenever s<o, other is certainly missing element 's', etc.

        let [mut s_not_o, mut o_not_s] = [vec![], vec![]];

        loop {

            match [s, o] {

                [None, None] => break, // both iterators are empty

                [None, Some(x)] => {

                    // self's iterator is empty.

                    // all remaning elements are in other but not self

                    o_not_s.push(x);

                    o_not_s.extend(o_it);

                    break;

                }

                [Some(x), None] => {

                    // other's iterator is empty.

                    // all remaning elements are in self but not other

                    s_not_o.push(x);

                    s_not_o.extend(s_it);

                    break;

                }

                [Some((sid, sb)), Some((oid, ob))] => {

                    if sid < oid {

                        // o is missing this element

                        s_not_o.push((sid, sb));

                        s = s_it.next();

                    } else if sid > oid {

                        // s is missing this element

                        o_not_s.push((oid, ob));

                        o = o_it.next();

                    } else if sb != ob {

                        assert_eq!(sid, oid);

                        // both predicates assign the variable but differ on the value

                        // No predicate exists which satisfies both!

                        return Aur::Nonexistant;

                    } else {

                        // both predicates assign the variable to the same value

                        s = s_it.next();

                        o = o_it.next();

                    }

                }

            }

        }

        // Observed zero inconsistencies. A unified predicate exists...

        match [s_not_o.is_empty(), o_not_s.is_empty()] {

            [true, true] => Aur::Equivalent,       // ... equivalent to both.

            [false, true] => Aur::FormerNotLatter, // ... equivalent to self.

            [true, false] => Aur::LatterNotFormer, // ... equivalent to other.

            [false, false] => {

                // ... which is the union of the predicates' assignments but

                //     is equivalent to neither self nor other.

                let mut new = self.clone();

                for (&id, &b) in o_not_s {

                    new.assigned.insert(id, b);

                }

                Aur::New(new)

            }

        }

    }

    // Compute the union of the assignments of the two given predicates, if it exists.

    // It doesn't exist if there is some value which the predicates assign to different values.

    pub(crate) fn union_with(&self, other: &Self) -> Option<Self> {

        let mut res = self.clone();

        for (&channel_id, &assignment_1) in other.assigned.iter() {

            match res.assigned.insert(channel_id, assignment_1) {

                Some(assignment_2) if assignment_1 != assignment_2 => return None,

                _ => {}

            }

        }

        Some(res)

    }

    pub(crate) fn query(&self, var: SpecVar) -> Option<SpecVal> {

        self.assigned.get(&var).copied()

    }

}

impl RoundCtx {

    // remove an arbitrary buffered message, along with the ID of the getter who receives it

    fn getter_pop(&mut self) -> Option<(PortId, SendPayloadMsg)> {

        self.payload_inbox.pop()

    }

    // buffer a message along with the ID of the getter who receives it

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

        if let Some(getter) = self.ips.port_info.map.get(&putter).unwrap().peer {

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

        }

    }

}

impl<T: Debug + std::cmp::Ord> Debug for VecSet<T> {

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

        f.debug_set().entries(self.vec.iter()).finish()

    }

}

impl Debug for Predicate {

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

        struct Assignment<'a>((&'a SpecVar, &'a SpecVal));

        impl Debug for Assignment<'_> {

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

                write!(f, "{:?}={:?}", (self.0).0, (self.0).1)

            }

        }

        f.debug_set().entries(self.assigned.iter().map(Assignment)).finish()

    }

}

impl serde::Serialize for SerdeProtocolDescription {

    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>

    where

        S: serde::Serializer,

    {

        let inner: &ProtocolDescription = &self.0;

        inner.serialize(serializer)

    }

}

impl<'de> serde::Deserialize<'de> for SerdeProtocolDescription {

    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>

    where

        D: serde::Deserializer<'de>,

    {

        let inner: ProtocolDescription = ProtocolDescription::deserialize(deserializer)?;

        Ok(Self(Arc::new(inner)))

    }

}

impl IdParts for SpecVar {

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

        self.0.id_parts()

    }

}

impl Debug for SpecVar {

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

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

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

    }

}

impl SpecVal {

    const FIRING: Self = SpecVal(1);

    const SILENT: Self = SpecVal(0);

    fn is_firing(self) -> bool {

        self == Self::FIRING

        // all else treated as SILENT

    }

    fn iter_domain() -> impl Iterator<Item = Self> {

        (0..).map(SpecVal)

    }

}

impl Debug for SpecVal {

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