pub trait MonoContext {

    type D: ProtocolDescription;

    type S: ComponentState<D = Self::D>;

    fn new_component(&mut self, moved_keys: HashSet<Key>, init_state: Self::S);

    fn new_channel(&mut self) -> [Key; 2];

}

pub trait PolyContext {

    type D: ProtocolDescription;

}

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

impl Key {

    pub fn from_raw(raw: u64) -> Self {

        Self(raw)

macro_rules! log {

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

        use std::fmt::Write;

        writeln!($logger, $($arg)*).unwrap();

    }};

}

            match blocker {

                Sb::Inconsistent => {} // DROP

                Sb::CouldntReadMsg(ekey) => {

                    assert!(self.ekeys.contains(&ekey));

                    let channel_id =

                        r_ctx.m_ctx.inner.endpoint_exts.get(ekey).unwrap().info.channel_id;

                        channel_id,

                        &branch.inbox,

                    );

                    if predicate.replace_assignment(channel_id, true) != Some(false) {

                        // don't rerun now. Rerun at next `sync_run`

        protocol_description: &ProtocolD,

        ekey: Key,

        payload_predicate: Predicate,

        payload: Payload,

    ) -> Result<SyncRunResult, EndpointErr> {

        // try exact match

        let to_run = if self.complete.contains_key(&payload_predicate) {

            // exact match with stopped machine

            );

            vec![]

        } else if let Some(mut branch) = self.incomplete.remove(&payload_predicate) {

            // exact match with running machine

                &payload_predicate

            );

            branch.inbox.insert(ekey, payload);

            vec![(payload_predicate, branch)]

        } else {

            );

            let mut incomplete2 = HashMap::<_, _>::default();

            let to_run = self

                .incomplete

                .drain()

                .filter_map(|(old_predicate, mut branch)| {

                    use CommonSatResult as Csr;

                    match old_predicate.common_satisfier(&payload_predicate) {

                        Csr::FormerNotLatter | Csr::Equivalent => {

                                &old_predicate

                            );

                            // old_predicate COVERS the assumptions of payload_predicate

                            let was = branch.inbox.insert(ekey, payload.clone());

                            assert!(was.is_none()); // INBOX MUST BE EMPTY!

                            Some((old_predicate, branch))

                        }

                        Csr::New(new) => {

                                &payload_predicate,

                                &old_predicate,

                                &new,

                            );

                            // payload_predicate has new assumptions. FORK!

                            let mut payload_branch = branch.clone();

                            // put the original back untouched

                            incomplete2.insert(old_predicate, branch);

                            Some((new, payload_branch))

                        }

                        Csr::LatterNotFormer => {

                                &old_predicate,

                                &payload_predicate,

                            );

                            // payload_predicate has new assumptions. FORK!

                            let mut payload_branch = branch.clone();

                            let was = payload_branch.inbox.insert(ekey, payload.clone());

                            // put the original back untouched

                            incomplete2.insert(old_predicate, branch);

                            Some((payload_predicate.clone(), payload_branch))

                        }

                        Csr::Nonexistant => {

                                &old_predicate,

                                &payload_predicate,

                            );

                            // predicates contradict

                            incomplete2.insert(old_predicate, branch);

                            None

}

impl MonoContext for MonoPContext<'_> {

    type D = ProtocolD;

    type S = ProtocolS;

    fn new_component(&mut self, moved_ekeys: HashSet<Key>, init_state: Self::S) {

            &moved_ekeys,

        );

        if moved_ekeys.is_subset(self.ekeys) {

            self.ekeys.retain(|x| !moved_ekeys.contains(x));

            self.inner.mono_ps.push(MonoP { state: init_state, ekeys: moved_ekeys });

        } else {

            endpoint: a,

        });

        let kg = self.inner.endpoint_exts.alloc(EndpointExt {

            info: EndpointInfo { polarity: Putter, channel_id },

            endpoint: b,

        });

            [kp, kg],

        );

        [kp, kg]

    }

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

        let mut bytes = Bytes8::default();

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

        let val = unsafe { std::mem::transmute::<Bytes8, _>(bytes) };

            val,

        );

        val

    }

}

        }

    }

}

impl PolyContext for BranchPContext<'_, '_> {

    type D = ProtocolD;

        assert!(self.ekeys.contains(&ekey));

        let channel_id = self.m_ctx.inner.endpoint_exts.get(ekey).unwrap().info.channel_id;

        let val = self.predicate.query(channel_id);

            self.m_ctx.my_subtree_id,

            val,

        );

        val

    }

        assert!(self.ekeys.contains(&ekey));

        let val = self.inbox.get(&ekey);

            self.m_ctx.my_subtree_id,

            val,

        );

        val

    }

}

        protocol_description: Arc<ProtocolD>,

        bound_proto_interface: &[(PortBinding, Polarity)],

        deadline: Instant,

    ) -> Result<(Self, Vec<(Key, Polarity)>), ConnectErr> {

        use ConnectErr::*;

        let mut channel_id_stream = ChannelIdStream::new(major);

        let mut endpoint_ext_todos = Arena::default();

        let mut ekeys_native = vec![];

        let mut ekeys_proto = vec![];

        let mut ekeys_network = vec![];

                    });

                    ekeys_network.push(ekey_proto);

                    ekeys_proto.push(ekey_proto);

                }

            }

        }

        // 2. convert the arena to Arena<EndpointExt>  and return the

        let (mut messenger_state, mut endpoint_exts) =

        let n_mono = Some(MonoN { ekeys: ekeys_native.into_iter().collect(), result: None });

        let p_monos = vec![MonoP {

            state: protocol_description.new_main_component(&ekeys_proto),

            ekeys: ekeys_proto.into_iter().collect(),

        }];

        // 6. Become a node in a sink tree, computing {PARENT, CHILDREN} from {NEIGHBORS}

        let family = Self::setup_sink_tree_family(

            major,

            &mut endpoint_exts,

            &mut messenger_state,

            ekeys_network,

            deadline,

        )?;

        let inner = ControllerInner {

            family,

            messenger_state,

            channel_id_stream,

            endpoint_exts,

            mono_ps: p_monos,

            mono_n: n_mono,

            round_index: 0,

        };

        let controller = Self { protocol_description, inner, ephemeral: Default::default() };

        Ok((controller, native_interface))

    }

    fn test_stream_connectivity(stream: &mut TcpStream) -> bool {

        stream.write(&[]).is_ok()

    }

    // inserts

    fn finish_endpoint_ext_todos(

        major: ControllerId,

        mut endpoint_ext_todos: Arena<EndpointExtTodo>,

        deadline: Instant,

    ) -> Result<(MessengerState, Arena<EndpointExt>), ConnectErr> {

        use {ConnectErr::*, EndpointExtTodo::*};

        // 1. define and setup a poller and event loop

            polled_undrained: Default::default(),

        };

        // 2. Register all EndpointExtTodos with ms.poll. each has one of {Endpoint, TcpStream, TcpListener}

        // 3. store the keyset of EndpointExtTodos which are not Finished in `to_finish`.

        let mut to_finish = HashSet::<_>::default();

        for (key, t) in endpoint_ext_todos.iter() {

            let token = key.to_token();

            match t {

                ActiveRecving { .. } | PassiveConnecting { .. } => unreachable!(),

                Finished(EndpointExt { endpoint, .. }) => {

                    ms.poll.register(endpoint, token, ready_r, edge)

                let entry = endpoint_ext_todos.get_mut(ekey).unwrap();

                match entry {

                    Finished(_) => {

                        polled_undrained_later.insert(ekey);

                    }

                    PassiveAccepting { addr, listener, .. } => {

                        assert!(event.readiness().is_readable());

                        let (stream, _peer_addr) =

                            listener.accept().map_err(|_| AcceptFailed(*addr))?;

                        ms.poll.deregister(listener).expect("wer");

                        ms.poll.register(&stream, token, ready_w, edge).expect("3y5");

                        take_mut::take(entry, |e| {

                            assert_let![PassiveAccepting { addr, info, .. } = e => {

                                PassiveConnecting { addr, info, stream }

                            }]

                        });

                    }

                    PassiveConnecting { addr, stream, .. } => {

                        assert!(event.readiness().is_writable());

                        if !Self::test_stream_connectivity(stream) {

                            return Err(PassiveConnectFailed(*addr));

                        }

                        ms.poll.reregister(stream, token, ready_r, edge).expect("52");

                        let mut res = Ok(());

                                let msg = Msg::SetupMsg(SetupMsg::ChannelSetup { info });

                                res = endpoint.send(msg);

                                Finished(EndpointExt { info, endpoint })

                            }]

                        });

                        res?;

                        assert!(to_finish.remove(&ekey));

                    }

                    ActiveConnecting { addr, stream, .. } => {

                        assert!(event.readiness().is_writable());

                        if Self::test_stream_connectivity(stream) {

                            // connect successful

                            ms.poll.reregister(stream, token, ready_r, edge).expect("52");

                            take_mut::take(entry, |e| {

                                assert_let![ActiveConnecting { stream, polarity, addr } = e => {

                                    let endpoint = Endpoint::from_fresh_stream(stream);

                                    ActiveRecving { endpoint, polarity, addr }

                                }]

                            });

                        } else {

                            // connect failure. retry!

                            ms.poll.deregister(stream).expect("wt");

                            std::thread::sleep(Duration::from_millis(backoff_millis));

                            backoff_millis = ((backoff_millis as f32) * 1.2) as u64 + 3;

                            let mut new_stream = TcpStream::connect(addr).unwrap();

                            ms.poll.register(&new_stream, token, ready_w, edge).expect("PAC 3");

                            std::mem::swap(stream, &mut new_stream);

                        }

                    }

                    ActiveRecving { addr, polarity, endpoint } => {

                        assert!(event.readiness().is_readable());

                        'recv_loop: while let Some(msg) = endpoint.recv()? {

                            if let Msg::SetupMsg(SetupMsg::ChannelSetup { info }) = msg {

                                if info.polarity == *polarity {

                                    return Err(PolarityMatched(*addr));

                                }

                                assert!(to_finish.remove(&ekey));

                                break 'recv_loop;

                            } else {

                                ms.delayed.push(ReceivedMsg { recipient: ekey, msg });

                            }

                        }

                    }

                }

            }

        }

        for ekey in polled_undrained_later {

            ms.polled_undrained.insert(ekey);

        });

        Ok((ms, endpoint_exts))

    }

    fn setup_sink_tree_family(

        major: ControllerId,

        endpoint_exts: &mut Arena<EndpointExt>,

        messenger_state: &mut MessengerState,

        neighbors: Vec<Key>,

        deadline: Instant,

    ) -> Result<ControllerFamily, ConnectErr> {

        use {ConnectErr::*, Msg::SetupMsg as S, SetupMsg::*};

        let mut messenger = (messenger_state, endpoint_exts);

        impl Messengerlike for (&mut MessengerState, &mut Arena<EndpointExt>) {

            fn get_state_mut(&mut self) -> &mut MessengerState {

                self.0

            }

        }

        // 1. broadcast my ID as the first echo. await reply from all in net_keylist

        let echo = S(LeaderEcho { maybe_leader: major });

        let mut awaiting = IndexSet::with_capacity(neighbors.len());

        for &n in neighbors.iter() {

            messenger.send(n, echo.clone())?;

            awaiting.insert(n);

        }

        // 2. Receive incoming replies. whenever a higher-id echo arrives,

        //    adopt it as leader, sender as parent, and reset the await set.

        let mut parent: Option<Key> = None;

        let mut my_leader = major;

        messenger.undelay_all();

        'echo_loop: while !awaiting.is_empty() || parent.is_some() {

            let ReceivedMsg { recipient, msg } = messenger.recv(deadline)?.ok_or(Timeout)?;

            match msg {

                S(LeaderAnnounce { leader }) => {

                    // someone else completed the echo and became leader first!

                    // the sender is my parent

                    parent = Some(recipient);

                    my_leader = leader;

                                    break 'echo_loop;

                                }

                            }

                        }

                        Greater => {

                            // join new echo

                            parent = Some(recipient);

                            my_leader = maybe_leader;

                            let echo = S(LeaderEcho { maybe_leader: my_leader });

                            awaiting.clear();

                            if neighbors.len() == 1 {

                                // immediately reply to parent

                                    "{:?} replying echo to parent {:?} immediately",

                                );

                                messenger.send(recipient, echo.clone())?;

                            } else {

                                for &n in neighbors.iter() {

                                    if n != recipient {

                                            "{:?} repeating echo {:?} to {:?}",

                                        );

                                        messenger.send(n, echo.clone())?;

                                        awaiting.insert(n);

                                    }

                                }

                            }

            Some(parent) => assert_ne!(

                my_leader, major,

                "I have {:?} as parent, but I consider myself ({:?}) the leader?",

                parent, major

            ),

        }

        // 3. broadcast leader announcement (except to parent: confirm they are your parent)

        //    in this loop, every node sends 1 message to each neighbor

        let msg_for_non_parents = S(LeaderAnnounce { leader: my_leader });

        for &k in neighbors.iter() {

            let msg =

                if Some(k) == parent { S(YouAreMyParent) } else { msg_for_non_parents.clone() };

            messenger.send(k, msg)?;

        }

        // await 1 message from all non-parents

        for &n in neighbors.iter() {

            if Some(n) != parent {