macro_rules! log {

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

    }};

}

    }

    // private function. mutates state but returns with round

    // result ASAP (allows for convenient error return with ?)

    fn connected_sync(

        cu: &mut ConnectorUnphased,

        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

        );

        // 1. run all proto components to Nonsync blockers

        // NOTE: original components are immutable until Decision::Success

        let mut branching_proto_components =

            HashMap::<ProtoComponentId, BranchingProtoComponent>::default();

        let mut unrun_components: Vec<(ProtoComponentId, ProtoComponent)> =

            cu.proto_components.iter().map(|(&k, v)| (k, v.clone())).collect();

            );

            use NonsyncBlocker as B;

            match blocker {

                B::ComponentExit => drop(component),

                B::Inconsistent => return Err(Se::InconsistentProtoComponent(proto_component_id)),

                B::SyncBlockStart => {

                    branching_proto_components

                        .insert(proto_component_id, BranchingProtoComponent::initial(component));

                }

            }

        }

        log!(

            cu.inner.logger,

            "All {} proto components are now done with Nonsync phase",

            branching_proto_components.len(),

        );

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

        let mut rctx = RoundCtx {

            solution_storage: {

                let n = std::iter::once(SubtreeId::LocalComponent(ComponentId::Native));

                let c = cu

                    .proto_components

                    .keys()

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

                log!(

                    cu.inner.logger,

                    "Children in subtree are: {:?}",

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

                );

                SolutionStorage::new(subtree_id_iter)

            },

            spec_var_stream: cu.inner.id_manager.new_spec_var_stream(),

            getter_buffer: Default::default(),

            deadline: timeout.map(|to| Instant::now() + to),

        };

        // Explore all native branches eagerly. Find solutions, buffer messages, etc.

        log!(

            cu.inner.logger,

            "Translating {} native batches into branches...",

            comm.native_batches.len()

        );

        let native_spec_var = rctx.spec_var_stream.next();

        log!(cu.inner.logger, "Native branch spec var is {:?}", native_spec_var);

        let mut branching_native = BranchingNative { branches: Default::default() };

        'native_branches: for ((native_branch, index), branch_spec_val) in

            comm.native_batches.drain(..).zip(0..).zip(SpecVal::iter_domain())

                    SubtreeId::LocalComponent(ComponentId::Native),

                    predicate.clone(),

                );

            }

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

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

                unreachable!()

            }

        }

        // restore the invariant: !native_batches.is_empty()

        comm.native_batches.push(Default::default());

        // Call to another big method; keep running this round until a distributed decision is reached

        let decision = Self::sync_reach_decision(

            cu,

            comm,

            &mut branching_native,

            &mut branching_proto_components,

            &mut rctx,

        )?;

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

        // propagate the decision to children

        let msg = Msg::CommMsg(CommMsg {

            round_index: comm.round_index,

            contents: CommMsgContents::CommCtrl(CommCtrlMsg::Announce {

                decision: decision.clone(),

            }),

        });

        log!(

            cu.inner.logger,

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

                        .into_iter()

                        .map(|(id, bpc)| (id, bpc.collapse_with(&predicate))),

                );

                log!(

                    cu.inner.logger,

                    "End round with (updated) component states {:?}",

                    cu.proto_components.keys()

                );

                // consume native

                Ok(Some(branching_native.collapse_with(&mut *cu.inner.logger, &predicate)))

            }

        };

        ret

    }

    fn sync_reach_decision(

        cu: &mut ConnectorUnphased,

        comm: &mut ConnectorCommunication,

        branching_native: &mut BranchingNative,

        branching_proto_components: &mut HashMap<ProtoComponentId, BranchingProtoComponent>,

        rctx: &mut RoundCtx,

    ) -> Result<Decision, UnrecoverableSyncError> {

        let mut already_requested_failure = false;

        if branching_native.branches.is_empty() {

        use NetEndpointError as Nee;

        // populate inbox as much as possible

        let before_len = self.inbox.len();

        'read_loop: loop {

            let res = self.stream.read_to_end(&mut self.inbox);

            match res {

                Err(e) if would_block(&e) => break 'read_loop,

                Ok(0) => break 'read_loop,

                Ok(_) => (),

                Err(_e) => return Err(Nee::BrokenNetEndpoint),

            }

        }

            logger,

            "Inbox bytes [{:x?}| {:x?}]",

            DenseDebugHex(&self.inbox[..before_len]),

            DenseDebugHex(&self.inbox[before_len..]),

        );

        let mut monitored = MonitoredReader::from(&self.inbox[..]);

        use bincode::config::Options;

        match Self::bincode_opts().deserialize_from(&mut monitored) {

            Ok(msg) => {

                let msg_size = monitored.bytes_read();

                self.inbox.drain(0..(msg_size.try_into().unwrap()));

                    logger,

                    "Yielding msg. Inbox len {}-{}=={}: [{:?}]",

                    self.inbox.len() + msg_size,

                    msg_size,

                    self.inbox.len(),

                    DenseDebugHex(&self.inbox[..]),

                );

                Ok(Some(msg))

            }

            Err(e) => match *e {

                bincode::ErrorKind::Io(k) if k.kind() == std::io::ErrorKind::UnexpectedEof => {

                    Ok(None)

            ConnectError::NetEndpointSetupError(

                net_endpoint_store.endpoint_exts[trane.index]

                    .net_endpoint

                    .stream

                    .local_addr()

                    .unwrap(), // stream must already be connected

                trane.error,

            )

        }

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

        // try yield undelayed net message

        if let Some(tup) = self.undelayed_messages.pop() {

            return Ok(tup);

        }

        loop {

            // try recv from some polled undrained NET endpoint

            if let Some(tup) = self

                .try_recv_undrained_net(logger)

                .map_err(|trane| map_trane(trane, &self.net_endpoint_store))?

            {

                return Ok(tup);

            }

            // poll if time remains

            self.poll_and_populate(logger, deadline)?;

        }

    }

    fn try_recv_undrained_net(

        &mut self,

        logger: &mut dyn Logger,

    ) -> Result<Option<(usize, Msg)>, TryRecvAnyNetError> {

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

            let net_endpoint = &mut self.net_endpoint_store.endpoint_exts[index].net_endpoint;

            if let Some(msg) = net_endpoint

                .try_recv(logger)

                .map_err(|error| TryRecvAnyNetError { error, index })?

            {

                if !net_endpoint.inbox.is_empty() {

                    // there may be another message waiting!

                    self.net_endpoint_store.polled_undrained.insert(index);

                }

                return Ok(Some((index, msg)));

            }

        }

        Ok(None)

    }

    fn poll_and_populate(

        &mut self,

        logger: &mut dyn Logger,

        } else {

            None

        };

        // Yes we do! Poll with remaining time as poll deadline

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

        for event in self.events.iter() {

            match TokenTarget::from(event.token()) {

                TokenTarget::Waker => {

                    // Can ignore. Residual event from endpoint manager setup procedure

                }

                TokenTarget::NetEndpoint { index } => {

                    self.net_endpoint_store.polled_undrained.insert(index);

                        logger,

                        "RECV poll event {:?} for NET endpoint index {:?}. undrained: {:?}",

                        &event,

                        index,

                        self.net_endpoint_store.polled_undrained.iter()

                    );

                }

                TokenTarget::UdpEndpoint { index } => {

                    self.udp_endpoint_store.polled_undrained.insert(index);

                        logger,

                        "RECV poll event {:?} for UDP endpoint index {:?}. undrained: {:?}",

                        &event,

                        index,

                        self.udp_endpoint_store.polled_undrained.iter()

                    );

                }

            }

        }

        self.events.clear();

        Ok(())

    }