}

            // stuck! make progress by receiving a msg

            // try recv messages arriving through endpoints

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

            {

                            } else {

                            }

                        }

                log!(

                    cu.logger,

                    "Received from endpoint {} ctrl msg  {:?}",

                    &comm_ctrl_msg

                );

                match comm_ctrl_msg {

                    CommCtrlMsg::Suggest { suggestion } => {

                        // only accept this control msg through a child endpoint

                            match suggestion {

                                Decision::Success(predicate) => {

                                    // child solution contributes to local solution

                                    log!(cu.logger, "Child provided solution {:?}", &predicate);

                                    rctx.solution_storage.submit_and_digest_subtree_solution(

                                        &mut *cu.logger,

                                        subtree_id,

                                        predicate,

                                    );

                                }

                            }

                        } else {

                            log!(

                                cu.logger,

                                "Discarding suggestion {:?} from non-child endpoint idx {:?}",

                                &suggestion,

                            );

                        }

                    }

                    CommCtrlMsg::Announce { decision } => {

                            // adopt this decision

                            return Ok(decision);

                        } else {

                            log!(

                                cu.logger,

                                "Discarding announcement {:?} from non-parent endpoint idx {:?}",

                                &decision,

                            );

                        }

                    }

                }

            }

            log!(cu.logger, "Endpoint msg recv done");

                    "skipping branch with {:?} that doesn't want the message (fastpath)",

                    &predicate

                );

                finished.insert(predicate, branch);

                continue;

            }

                    // this branch does not receive the message

                    log!(

                        cu.logger,

                        "skipping branch with {:?} that doesn't want the message (slowpath)",

                        &predicate

                    );

                    finished.insert(predicate, branch);

                }

                    // retain the existing predicate, but add this payload

                    feed_branch(&mut branch, &predicate);

                    log!(cu.logger, "branch pred covers it! Accept the msg");

                    finished.insert(predicate, branch);

                }

                    // fork branch, give fork the message and payload predicate. original branch untouched

                    let mut branch2 = branch.clone();

                    let predicate2 = send_payload_msg.predicate.clone();

                    feed_branch(&mut branch2, &predicate2);

                    log!(

                        cu.logger,

                        &predicate2,

                        &predicate

                    );

                    finished.insert(predicate, branch);

                    finished.insert(predicate2, branch2);

                }

                    // fork branch, give fork the message and the new predicate. original branch untouched

                    let mut branch2 = branch.clone();

                    feed_branch(&mut branch2, &predicate2);

                    log!(

                        cu.logger,

                        "new subsuming pred created {:?}. forking and feeding",

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

            if branch.ended {

                log!(logger, "Skipping ended branch with {:?}", &predicate);

                blocked.insert(predicate, branch);

                continue;

            }

            log!(logger, "visiting branch with pred {:?}", &predicate);

                    // this branch does not receive the message

                    log!(logger, "skipping branch");

                    blocked.insert(predicate, branch);

                }

                    // retain the existing predicate, but add this payload

                    log!(logger, "feeding this branch without altering its predicate");

                    branch.feed_msg(getter, send_payload_msg.payload.clone());

                    unblocked.insert(predicate, branch);

                }

                    // fork branch, give fork the message and payload predicate. original branch untouched

                    log!(logger, "Forking this branch, giving it the predicate of the msg");

                    let mut branch2 = branch.clone();

                    let predicate2 = send_payload_msg.predicate.clone();

                    branch2.feed_msg(getter, send_payload_msg.payload.clone());

                    blocked.insert(predicate, branch);

                    unblocked.insert(predicate2, branch2);

                }

                    // fork branch, give fork the message and the new predicate. original branch untouched

                    log!(logger, "Forking this branch with new predicate {:?}", &predicate2);

                    let mut branch2 = branch.clone();

                    branch2.feed_msg(getter, send_payload_msg.payload.clone());

                    blocked.insert(predicate, branch);

                    unblocked.insert(predicate2, branch2);

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

        impl EndpointManager {

            fn handle_msg(

                &mut self,

                logger: &mut dyn Logger,

                round_ctx: &mut impl RoundCtxTrait,

                msg: Msg,

                round_index: usize,

                some_message_enqueued: &mut bool,

            ) -> Option<(usize, CommCtrlMsg)> {

                let comm_msg_contents = match msg {

                    Msg::SetupMsg(..) => return None,

                            log!(

                                logger,

                                "We are in round {}, but msg is for round {}. Buffer",

                                comm_msg.round_index,

                                round_index,

                            );

                            return None;

                        }

                    },

                };

                match comm_msg_contents {

                    CommMsgContents::SendPayload(send_payload_msg) => {

                        round_ctx.getter_add(getter, send_payload_msg);

                        *some_message_enqueued = true;

                        None

                    }

                }

            }

        }

        use {PollAndPopulateError as Pape, UnrecoverableSyncError as Use};

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

        let mut some_message_enqueued = false;

        // try yield undelayed net message

                logger,

                round_ctx,

                msg,

                round_index,

                &mut some_message_enqueued,

            ) {

            }

        }

        loop {

            // try receive a net message

                    logger,

                    round_ctx,

                    msg,

                    round_index,

                    &mut some_message_enqueued,

                ) {

                }

            }

            // try receive a udp message

            let recv_buffer = self.udp_in_buffer.as_mut_slice();

            while let Some(index) = self.udp_endpoint_store.polled_undrained.pop() {

                let ee = &mut self.udp_endpoint_store.endpoint_exts[index];

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

struct SendPayloadMsg {

    predicate: Predicate,

    payload: Payload,

}

#[derive(Debug, PartialEq)]

    FormerNotLatter,

    LatterNotFormer,

    Equivalent,

    New(Predicate),

    Nonexistant,

}

#[derive(Debug, Default)]

struct NativeBatch {

    // invariant: putters' and getters' polarities respected

    to_put: HashMap<PortId, Payload>,

    to_get: HashSet<PortId>,

}

enum TokenTarget {

    NetEndpoint { index: usize },

    UdpEndpoint { index: usize },

    Waker,

}

trait RoundCtxTrait {

    fn get_deadline(&self) -> &Option<Instant>;

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

}

enum CommRecvOk {

    TimeoutWithoutNew,

    NewPayloadMsgs,

}

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

fn would_block(err: &std::io::Error) -> bool {

    err.kind() == std::io::ErrorKind::WouldBlock

}

impl TokenTarget {

    }

    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 => {}

            }

        }

        true

    }

    // returns true IFF self.unify would return Equivalent OR FormerNotLatter

        // 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()];

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

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

        loop {

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

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

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

                }

            }

        }

    }

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

        let mut res = self.clone();

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

                log!(cu.logger, "~~~ CONNECT called timeout {:?}", timeout);

                let deadline = timeout.map(|to| Instant::now() + to);

                // connect all endpoints in parallel; send and receive peer ids through ports

                let mut endpoint_manager = new_endpoint_manager(

                    &mut *cu.logger,

                    &setup.net_endpoint_setups,

                    &mut cu.port_info,

                    &deadline,

                )?;

                log!(

                    cu.logger,

                    "Successfully connected {} endpoints",

            }

        }

    }

}

fn new_endpoint_manager(

    logger: &mut dyn Logger,

    port_info: &mut PortInfo,

    deadline: &Option<Instant>,

) -> Result<EndpointManager, ConnectError> {

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

    use std::sync::atomic::AtomicBool;

    use ConnectError as Ce;

        todo_endpoint: TodoEndpoint,

        endpoint_setup: NetEndpointSetup,

        local_port: PortId,

        sent_local_port: bool,          // true <-> I've sent my local port

        recv_peer_port: Option<PortId>, // Some(..) <-> I've received my peer's port

    }

    enum TodoEndpoint {

        Accepting(TcpListener),

        NetEndpoint(NetEndpoint),

    }

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

    // 1. Start to construct EndpointManager

    const WAKER_PERIOD: Duration = Duration::from_millis(300);

    struct WakerState {

        continue_signal: AtomicBool,

        waker: mio::Waker,

    }

    let mut waker_state: Option<Arc<WakerState>> = None;

    let mut poll = Poll::new().map_err(|_| Ce::PollInitFailed)?;

    let mut net_polled_undrained = VecSet::default();

    let udp_polled_undrained = VecSet::default();

    let mut delayed_messages = vec![];

    // 2. create a registered (TcpListener/Endpoint) for passive / active respectively

        .iter()

        .enumerate()

        .map(|(index, (local_port, endpoint_setup))| {

        })

        .collect::<Result<Vec<Todo>, ConnectError>>()?;

    // 3. Using poll to drive progress:

    //    - accept an incoming connection for each TcpListener (turning them into endpoints too)

    //    - for each endpoint, send the local PortId

    //    - for each endpoint, recv the peer's PortId, and

    // all in connect_failed are NOT registered with Poll

    let mut connect_failed: HashSet<usize> = Default::default();

    while !setup_incomplete.is_empty() {

        let remaining = if let Some(deadline) = deadline {

            Some(deadline.checked_duration_since(Instant::now()).ok_or(Ce::Timeout)?)

        } else {

            None

        };

        poll.poll(&mut events, remaining).map_err(|_| Ce::PollFailed)?;

        for event in events.iter() {

            let token = event.token();

            let token_target = TokenTarget::from(token);

            match token_target {

                TokenTarget::Waker => {

                    log!(

                        logger,

                        "Notification from waker. connect_failed is {:?}",

                        connect_failed.iter()

                    );

                    assert!(waker_state.is_some());

                        log!(

                            logger,

                            "Restarting connection with endpoint {:?} {:?}",

                            todo.endpoint_setup.sock_addr

                        );

                        match &mut todo.todo_endpoint {

                            TodoEndpoint::NetEndpoint(endpoint) => {

                                let mut new_stream =

                                    TcpStream::connect(todo.endpoint_setup.sock_addr)

                                        .expect("mio::TcpStream connect should not fail!");

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

                                poll.registry()

                                    .register(&mut endpoint.stream, token, BOTH)

                                    .unwrap();

                            }

                            _ => unreachable!(),

                        }

                    }

                }

                TokenTarget::NetEndpoint { index } => {

                    let todo: &mut Todo = &mut todos[index];

                    if let TodoEndpoint::Accepting(listener) = &mut todo.todo_endpoint {

                        // FIRST try complete this connection

                        match listener.accept() {

                            Err(e) if would_block(&e) => {

                        }

                        // event wasn't ERROR

                        if connect_failed.contains(&index) {

                            // spurious wakeup

                            continue;

                        }

                        let local_polarity = *port_info.polarities.get(&todo.local_port).unwrap();

                        if event.is_writable() && !todo.sent_local_port {

                            // can write and didn't send setup msg yet? Do so!

                            let msg = Msg::SetupMsg(SetupMsg::MyPortInfo(MyPortInfo {

                                polarity: local_polarity,

                                port: todo.local_port,

                                }

                            }

                        }

                        // is the setup for this net_endpoint now complete?

                        if todo.sent_local_port && todo.recv_peer_port.is_some() {

                            // yes! connected, sent my info and received peer's info

                            log!(logger, "endpoint[{}] is finished!", index);

                        }

                    }

                }

            }

        }

    log!(logger, "Endpoint setup complete! Cleaning up and building structures");

    if let Some(arc) = waker_state {

        log!(logger, "Sending waker the stop signal");

        arc.continue_signal.store(false, std::sync::atomic::Ordering::SeqCst);

        // TODO leave the waker registered?

    }

    let net_endpoint_exts = todos

        .into_iter()

        .enumerate()

        .map(|(index, Todo { todo_endpoint, local_port, .. })| NetEndpointExt {

            net_endpoint: match todo_endpoint {

                }

                _ => unreachable!(),

            },

            getter_for_incoming: local_port,

        })

        .collect();

    Ok(EndpointManager {

        poll,

        events,

        undelayed_messages: delayed_messages, // no longer delayed

        delayed_messages: Default::default(),

        net_endpoint_store: EndpointStore {

            }

        });

    })

    .unwrap();

}