() => (print!("\n"));

    ($($arg:tt)*) => ({

        use std::io::Write;

        std::writeln!(std::io::stdout().lock(), $($arg)*).expect("LPRINTLN");

    })

}

macro_rules! assert_let {

    ($pat:pat = $expr:expr => $work:expr) => {

        if let $pat = $expr {

            $work

        } else {

            panic!("assert_let failed");

        mut m_ctx: PolyPContext,

        protocol_description: &ProtocolD,

        mut to_run: Vec<(Predicate, BranchP)>,

    ) -> Result<SyncRunResult, EndpointErr> {

        use SyncRunResult as Srr;

        let cid = m_ctx.inner.channel_id_stream.controller_id;

        while let Some((mut predicate, mut branch)) = to_run.pop() {

            let mut r_ctx = BranchPContext {

                m_ctx: m_ctx.reborrow(),

                ekeys: &self.ekeys,

                predicate: &predicate,

                inbox: &branch.inbox,

            };

            use PolyBlocker as Sb;

            let blocker = branch.state.sync_run(&mut r_ctx, protocol_description);

                r_ctx.m_ctx.my_subtree_id,

                &predicate,

                &blocker

            );

            match blocker {

                Sb::Inconsistent => {} // DROP

                        channel_id,

                        &branch.inbox,

                    );

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

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

                        self.incomplete.insert(predicate, branch);

                    } else {

                    }

                    // ELSE DROP

                }

                Sb::CouldntCheckFiring(ekey) => {

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

                    let channel_id =

                    }

                    assert!(predicate.replace_assignment(channel_id, true).is_none());

                    to_run.push((predicate, branch));

                    to_run.push((predicate_f, branch_f));

                }

                Sb::SyncBlockEnd => {

                    // come up with the predicate for this local solution

                    let lookup =

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

                    let ekeys_channel_id_iter = self.ekeys.iter().map(lookup);

                    predicate.batch_assign_nones(ekeys_channel_id_iter, false);

                    // OK now check we really received all the messages we expected to

                    let num_fired = predicate.iter_matching(true).count();

                    let num_msgs =

                        branch.inbox.keys().chain(branch.outbox.keys()).map(lookup).count();

                    match num_fired.cmp(&num_msgs) {

                        Ordering::Less => unreachable!(),

                            m_ctx.my_subtree_id,

                            &predicate,

                        ),

                        Ordering::Equal => {

                                m_ctx.my_subtree_id,

                                &predicate,

                            );

                            m_ctx.solution_storage.submit_and_digest_subtree_solution(

                                m_ctx.my_subtree_id,

                                predicate.clone(),

                    }

                })

                .collect();

            std::mem::swap(&mut self.incomplete, &mut incomplete2);

            to_run

        };

        self.poly_run_these_branches(m_ctx, protocol_description, to_run)

    }

    pub(crate) fn become_mono(

        mut self,

        decision: &Predicate,

use crate::common::*;

use crate::runtime::{actors::*, endpoint::*, errors::*, *};

impl Controller {

    fn end_round_with_decision(&mut self, decision: Predicate) -> Result<(), SyncErr> {

        let mut table_row = HashMap::default();

        self.inner.mono_n = self

            .ephemeral

            .poly_n

            .take()

            .map(|poly_n| poly_n.become_mono(&decision, &mut table_row));

        self.inner.mono_ps.extend(

            self.ephemeral.poly_ps.drain(..).map(|m| m.become_mono(&decision, &mut table_row)),

        );

        for (ekey, payload) in table_row {

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

        }

        for channel_id in decision.iter_matching(false) {

        }

        let announcement =

            CommMsgContents::Announce { oracle: decision }.into_msg(self.inner.round_index);

        for &child_ekey in self.inner.family.children_ekeys.iter() {

                &announcement,

                child_ekey

            );

            self.inner

                .endpoint_exts

                .get_mut(child_ekey)

        self.ephemeral.clear();

        Ok(())

    }

    // Drain self.ephemeral.solution_storage and handle the new locals. Return decision if one is found

    fn handle_locals_maybe_decide(&mut self) -> Result<bool, SyncErr> {

        if let Some(parent_ekey) = self.inner.family.parent_ekey {

            // I have a parent -> I'm not the leader

            let parent_endpoint =

                &mut self.inner.endpoint_exts.get_mut(parent_ekey).expect("huu").endpoint;

            for partial_oracle in self.ephemeral.solution_storage.iter_new_local_make_old() {

                let msg =

                    CommMsgContents::Elaborate { partial_oracle }.into_msg(self.inner.round_index);

                parent_endpoint.send(msg)?;

            }

            Ok(false)

        } else {

            // I have no parent -> I'm the leader

            assert!(self.inner.family.parent_ekey.is_none());

            let maybe_decision = self.ephemeral.solution_storage.iter_new_local_make_old().next();

            Ok(if let Some(decision) = maybe_decision {

                self.end_round_with_decision(decision)?;

                true

            } else {

                false

            })

        }

        sync_batches: Option<impl Iterator<Item = SyncBatch>>,

    ) -> Result<(), SyncErr> {

        // TODO! fuse handle_locals_return_decision and end_round_return_decision

        assert!(self.ephemeral.is_clear());

        // 1. Run the Mono for each Mono actor (stored in `self.mono_ps`).

        //    Some actors are dropped. some new actors are created.

        //    Ultimately, we have 0 Mono actors and a list of unnamed sync_actors

        self.ephemeral.poly_ps.clear();

        // let mut poly_ps: Vec<PolyP> = vec![];

        while let Some(mut mono_p) = self.inner.mono_ps.pop() {

            let mut m_ctx = MonoPContext {

                ekeys: &mut mono_p.ekeys,

                inner: &mut self.inner,

                // endpoint_exts: &mut self.endpoint_exts,

                // mono_ps: &mut self.mono_ps,

                // channel_id_stream: &mut self.channel_id_stream,

            };

            // cross boundary into crate::protocol

            let blocker = mono_p.state.pre_sync_run(&mut m_ctx, &self.protocol_description);

            match blocker {

                MonoBlocker::Inconsistent => return Err(SyncErr::Inconsistent),

                MonoBlocker::ComponentExit => drop(mono_p),

                MonoBlocker::SyncBlockStart => self.ephemeral.poly_ps.push(mono_p.into()),

            }

        }

            self.ephemeral.poly_ps.len()

        );

        // 3. define the mapping from ekey -> actor

        //    this is needed during the event loop to determine which actor

        //    should receive the incoming message.

                .poly_ps

                .iter()

                .enumerate()

                .flat_map(|(index, m)| m.ekeys.iter().map(move |&e| (e, P { index })));

            n.chain(p).collect()

        };

            &ekey_to_holder

        );

        // 4. Create the solution storage. it tracks the solutions of "subtrees"

        //    of the controller in the overlay tree.

        self.ephemeral.solution_storage.reset({

                .inner

                .family

                .children_ekeys

                .iter()

                .map(|&ekey| SubtreeId::ChildController { ekey });

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

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

            );

            subtree_id_iter

        });

        // 5. kick off the synchronous round of the native actor if it exists

        assert_eq!(sync_batches.is_some(), self.inner.mono_n.is_some()); // TODO better err

        self.ephemeral.poly_n = if let Some(sync_batches) = sync_batches {

            // using if let because of nested ? operator

            // TODO check that there are 1+ branches or NO SOLUTION

            let poly_n = self.kick_off_native(sync_batches)?;

                poly_n.branches.keys().collect::<Vec<_>>()

            );

            Some(poly_n)

        } else {

            None

        };

        // 6. Kick off the synchronous round of each protocol actor

        //    If just one actor becomes inconsistent now, there can be no solution!

        //    TODO distinguish between completed and not completed poly_p's?

        for (index, poly_p) in self.ephemeral.poly_ps.iter_mut().enumerate() {

            let my_subtree_id = SubtreeId::PolyP { index };

            let m_ctx = PolyPContext {

                my_subtree_id,

                inner: &mut self.inner,

                solution_storage: &mut self.ephemeral.solution_storage,

            };

            use SyncRunResult as Srr;

            let blocker = poly_p.poly_run(m_ctx, &self.protocol_description)?;

            match blocker {

                Srr::NoBranches => return Err(SyncErr::Inconsistent),

                Srr::AllBranchesComplete | Srr::BlockingForRecv => (),

            }

        }

        // 7. `solution_storage` may have new solutions for this controller

        //    handle their discovery. LEADER => announce, otherwise => send to parent

        {

            let peeked = self.ephemeral.solution_storage.peek_new_locals().collect::<Vec<_>>();

                peeked.len(),

                peeked

            );

        }

        if self.handle_locals_maybe_decide()? {

            return Ok(());

        }

        // 4. Receive incoming messages until the DECISION is made

        self.undelay_all();

        'recv_loop: loop {

            let received = self.recv(deadline)?.ok_or(SyncErr::Timeout)?;

            let current_content = match received.msg {

                Msg::SetupMsg(_) => {

                    // This occurs in the event the connector was malformed during connect()

                    return Err(SyncErr::UnexpectedSetupMsg);

                }

                Msg::CommMsg(CommMsg { round_index, .. })

                    if round_index < self.inner.round_index =>

                {

                    // Old message! Can safely discard

                    drop(received);

                    continue 'recv_loop;

                }

                Msg::CommMsg(CommMsg { round_index, .. })

                    if round_index > self.inner.round_index =>

                {

                    // Message from a next round. Keep for later!

                        &received

                    );

                    self.delay(received);

                    continue 'recv_loop;

                }

                Msg::CommMsg(CommMsg { contents, round_index }) => {

                    assert_eq!(round_index, self.inner.round_index);

                    contents

                }

            };

            match current_content {

                CommMsgContents::Elaborate { partial_oracle } => {

                    // Child controller submitted a subtree solution.

                    if !self.inner.family.children_ekeys.contains(&received.recipient) {

                        return Err(SyncErr::ElaborateFromNonChild);

                    }

                    let subtree_id = SubtreeId::ChildController { ekey: received.recipient };

                        subtree_id,

                        &partial_oracle

                    );

                    self.ephemeral

                        .solution_storage

                        .submit_and_digest_subtree_solution(subtree_id, partial_oracle);

                    }

                }

                CommMsgContents::Announce { oracle } => {

                    if self.inner.family.parent_ekey != Some(received.recipient) {

                        return Err(SyncErr::AnnounceFromNonParent);

                    }

                        received.recipient,

                        &oracle

                    );

                    return self.end_round_with_decision(oracle);

                }

                CommMsgContents::SendPayload { payload_predicate, payload } => {

                    // message for some actor. Feed it to the appropriate actor

                    // and then give them another chance to run.

                    let subtree_id = ekey_to_holder.get(&received.recipient);

                    );

                    match subtree_id {

                        None => {

                            // this happens when a message is sent to a component that has exited.

                            // It's safe to drop this message;

                            // The sender branch will certainly not be part of the solution

                                m_ctx,

                                &self.protocol_description,

                                received.recipient,

                                payload_predicate,

                                payload,

                            )?;

                                subtree_id,

                                blocker

                            );

                            match blocker {

                                Srr::NoBranches => return Err(SyncErr::Inconsistent),

                                Srr::BlockingForRecv | Srr::AllBranchesComplete => {

                                }

                            }

                        }

                    };

                }

            }

        }

    }

}

impl ControllerEphemeral {

            native_interface,

            sync_batches: vec![Default::default()],

            controller,

        });

        Ok(())

    }

    pub fn put(&mut self, native_port_index: usize, payload: Payload) -> Result<(), PortOpErr> {

        use PortOpErr::*;

        let connected = match self {

            Connector::Connected(connected) => connected,

            _ => return Err(NotConnected),

    channel_id_stream: ChannelIdStream,

    endpoint_exts: Arena<EndpointExt>,

    messenger_state: MessengerState,

    mono_n: Option<MonoN>,

    mono_ps: Vec<MonoP>,

    family: ControllerFamily,

}

/// This structure has its state entirely reset between synchronous rounds

#[derive(Debug, Default)]

struct ControllerEphemeral {

    solution_storage: SolutionStorage,

            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 {

#[test]

fn incremental() {

    let timeout = Duration::from_millis(1_500);

    let addrs = ["127.0.0.1:7010".parse().unwrap(), "127.0.0.1:7011".parse().unwrap()];

    let a = thread::spawn(move || {

        x.configure(

            b"primitive main(out a, out b) {

            synchronous {

                msg m = create(0);

                put(a, m);

            }

        )

        .unwrap();

        x.bind_port(0, PortBinding::Passive(addrs[0])).unwrap();

        x.bind_port(1, PortBinding::Passive(addrs[1])).unwrap();

        x.connect(timeout).unwrap();

        assert_eq!(0, x.sync(timeout).unwrap());

    });

    let b = thread::spawn(move || {

        x.configure(

            b"primitive main(in a, in b) {

            synchronous {

                get(a);

            }

        }",

        )

        .unwrap();

        x.bind_port(0, PortBinding::Active(addrs[0])).unwrap();

        x.bind_port(1, PortBinding::Active(addrs[1])).unwrap();

        x.connect(timeout).unwrap();

        assert_eq!(0, x.sync(timeout).unwrap());

    });

    handle(a.join());

    handle(b.join());

}

#[test]

fn duo_positive() {

    let timeout = Duration::from_millis(1_500);

    let addrs = ["127.0.0.1:7012".parse().unwrap(), "127.0.0.1:7013".parse().unwrap()];

    let a = thread::spawn(move || {

        let mut x = Connector::Unconfigured(Unconfigured { controller_id: 0 });

        primitive main(out a, out b) {

            synchronous {}

            synchronous {}

            synchronous {

                msg m = create(0);

                put(a, m);

            }

            synchronous {

                msg m = create(0);

                put(b, m);

            }

        x.bind_port(0, PortBinding::Passive(addrs[0])).unwrap();

        x.bind_port(1, PortBinding::Passive(addrs[1])).unwrap();

        x.connect(timeout).unwrap();

        assert_eq!(0, x.sync(timeout).unwrap());

        assert_eq!(0, x.sync(timeout).unwrap());

        assert_eq!(0, x.sync(timeout).unwrap());

        assert_eq!(0, x.sync(timeout).unwrap());

    });

    let b = thread::spawn(move || {

        let mut x = Connector::Unconfigured(Unconfigured { controller_id: 1 });

        primitive main(in a, in b) {

            while (true) {

                synchronous {

                    if (fires(a)) {

                        get(a);

                    }

                synchronous {

                    if (fires(b)) {

                        get(b);

                    }

                }

            }

        x.bind_port(0, PortBinding::Active(addrs[0])).unwrap();

        x.bind_port(1, PortBinding::Active(addrs[1])).unwrap();

        x.connect(timeout).unwrap();

        assert_eq!(0, x.sync(timeout).unwrap());

        assert_eq!(0, x.sync(timeout).unwrap());

        assert_eq!(0, x.sync(timeout).unwrap());