/*

Change the definition of these macros to control the logging level statically

*/

macro_rules! log {

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

        // if let Some(w) = $logger.line_writer() {

        //     let _ = writeln!(w, $($arg)*);

        // }

    }};

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

        // if let Some(w) = $logger.line_writer() {

        //     let _ = writeln!(w, $($arg)*);

        // }

    }};

}

                comm.round_result = Self::connected_sync(cu, comm, timeout);

                comm.round_index += 1;

                match &comm.round_result {

                    Ok(None) => unreachable!(),

                    Ok(Some(ok_result)) => Ok(ok_result.batch_index),

                    Err(sync_error) => Err(sync_error.clone()),

                }

            }

        }

    }

    // Attempts to complete the synchronous round for the given

    // communication-phased connector structure.

    // Modifies components and ports in `cu` IFF the round succeeds.

    #[inline]

    fn connected_sync(

        cu: &mut ConnectorUnphased,

        comm: &mut ConnectorCommunication,

        timeout: Option<Duration>,

    ) -> Result<Option<RoundEndedNative>, SyncError> {

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

        use SyncError as Se;

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

        // Create separate storages for ports and components stored in `cu`,

        // while kicking off the branching of components until the set of

        // components entering their synchronous block is finalized in `branching_proto_components`.

        // This is the last time cu's components and ports are accessed until the round is decided.

        let mut ips = cu.ips.clone();

        let mut branching_proto_components =

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

        let mut unrun_components: Vec<(ComponentId, ComponentState)> = cu

            .proto_components

            .iter()

            .map(|(&proto_id, proto)| (proto_id, proto.clone()))

            .collect();

        log!(cu.logger(), "Nonsync running {} proto components...", unrun_components.len());

        // initially, the set of components to run is the set of components stored by `cu`,

        // but they are eventually drained into `branching_proto_components`.

        // Some components exit first, and others are created and put into `unrun_components`.

        while let Some((proto_component_id, mut component)) = unrun_components.pop() {

            log!(

                cu.logger(),

                "Nonsync running proto component with ID {:?}. {} to go after this",

                proto_component_id,

                unrun_components.len()

            );

            let (logger, proto_description) = cu.logger_and_protocol_description();

                proto_component_id,

                unrun_components: &mut unrun_components,

            };

            let blocker = component.nonsync_run(&mut ctx, proto_description);

            log!(

                cu.logger(),

                "proto component {:?} ran to nonsync blocker {:?}",

                proto_component_id,

                &blocker

            );

            use NonsyncBlocker as B;

            match blocker {

                B::ComponentExit => drop(component),

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

                B::SyncBlockStart => assert!(branching_proto_components

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

                    .is_none()), // Some(_) returned IFF some component identifier key is overwritten (BAD!)

            }

        }

        log!(

            cu.logger(),

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

            branching_proto_components.len(),

        );

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

        let mut rctx = RoundCtx {

            ips, // already used previously, now moved into RoundCtx

            solution_storage: {

                let subtree_id_iter = {

                    // Create an iterator over the identifiers of this

                    // connector's childen in the _solution tree_.

                    // Namely, the native, all locally-managed components,

                    // and all this connector's children in the _consensus tree_ (other connectors).

                    let n = std::iter::once(SubtreeId::LocalComponent(cu.native_component_id));

                    let c = branching_proto_components

                        .keys()

                        .map(|&cid| SubtreeId::LocalComponent(cid));

                    let e = comm

                        .neighborhood

                        .children

                        .iter()

                        .map(|&index| SubtreeId::NetEndpoint { index });

                    n.chain(c).chain(e)

                };

                log!(

                    cu.logger,

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

                    DebuggableIter(subtree_id_iter.clone())

                );

                SolutionStorage::new(subtree_id_iter)

            },

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

            payload_inbox: Default::default(), // buffer for in-memory payloads to be handled

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

        };

        log!(cu.logger(), "Round context structure initialized");

        // Prepare the branching native component, involving the conversion

        // of its synchronous batches (user provided) into speculative branches eagerly.

        // As a side effect, send all PUTs with the appropriate predicates.

        // Afterwards, each native component's speculative branch finds a local

        // solution the moment it's received all the messages it's awaiting.

        log!(

            cu.logger(),

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

            comm.native_batches.len()

        );

        // Allocate a single speculative variable to distinguish each native branch.

        // This enables native components to have distinct branches with identical

        // FIRING variables.

        let native_spec_var = rctx.spec_var_stream.next();

        log!(cu.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())

        {

            let NativeBatch { to_get, to_put } = native_branch;

            // compute the solution predicate to associate with this branch.

            let predicate = {

                let mut predicate = Predicate::default();

            if branch.is_ended() {

                // empty to_get set => already corresponds with a component solution

                log!(

                    cu.logger(),

                    "Native submitting solution for batch {} with {:?}",

                    index,

                    &predicate

                );

                rctx.solution_storage.submit_and_digest_subtree_solution(

                    cu,

                    SubtreeId::LocalComponent(cu.native_component_id),

                    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!

        log!(cu.logger(), "Searching for decision...");

        let decision = Self::sync_reach_decision(

            cu,

            comm,

            &mut branching_native,

            &mut branching_proto_components,

            &mut rctx,

        )?;

        log!(cu.logger(), "Committing to decision {:?}!", &decision);

        comm.endpoint_manager.udp_endpoints_round_end(&mut *cu.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.logger(),

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

            &msg,

            &comm.neighborhood.children

        );

        for &child in comm.neighborhood.children.iter() {

            comm.endpoint_manager.send_to_comms(child, &msg)?;

        }

        let ret = match decision {

            Decision::Failure => {

                // untouched port/component fields of `cu` are NOT overwritten.

                // the result is a rollback.

                Err(Se::RoundFailure)

            }

            Decision::Success(predicate) => {

                // commit changes to component states

                cu.proto_components.clear();

                cu.proto_components.extend(

                    // "flatten" branching components, committing the speculation

                    // consistent with the predicate decided upon.

                    branching_proto_components

                        .into_iter()

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

                );

                // commit changes to ports and id_manager

                cu.ips = rctx.ips;

                log!(

                    cu.logger,

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

                    cu.proto_components.keys()

                );

                // consume native

                let round_ok = branching_native.collapse_with(&mut *cu.logger(), &predicate);

                Ok(Some(round_ok))

            }

        };

        log!(cu.logger(), "Sync round ending! Cleaning up");

        ret

    }

    // Once the synchronous round has been started, this procedure

    // routs and handles payloads, receives control messages from neighboring connectors,

    // checks for timeout, and aggregates solutions until a distributed decision is reached.

    // The decision is either a solution (success case), or a distributed timeout rollback (failure case)

    // The final possible outcome is an unrecoverable error, which results from some fundamental misbehavior,

    // a network channel breaking, etc.

    fn sync_reach_decision(

        cu: &mut impl CuUndecided,

        comm: &mut ConnectorCommunication,

        branching_native: &mut BranchingNative,

        branching_proto_components: &mut HashMap<ComponentId, BranchingProtoComponent>,

        rctx: &mut RoundCtx,

    ) -> Result<Decision, UnrecoverableSyncError> {

        // The round is in progress, and now its just a matter of arriving at a decision.

        let mut already_requested_failure = false;

        if branching_native.branches.is_empty() {

            // An unsatisfiable native is the easiest way to detect failure

            log!(cu.logger(), "Native starts with no branches! Failure!");

            match comm.neighborhood.parent {

                Some(parent) => {

                    if already_requested_failure.replace_with_true() {

                        Self::request_failure(cu, comm, parent)?

                    } else {

                        log!(cu.logger(), "Already requested failure");

                    }

                }

                None => {

                    log!(cu.logger(), "No parent. Deciding on failure");

                    return Ok(Decision::Failure);

                }

            }

        }

        // Create a small set of "workspace" hashmaps, to be passed by-reference into various calls.

        // This is an optimization, avoiding repeated allocation.

        let mut pcb_temps_owner = <[HashMap<Predicate, ProtoComponentBranch>; 3]>::default();

        let mut pcb_temps = MapTempsGuard(&mut pcb_temps_owner);

        let mut bn_temp_owner = <HashMap<Predicate, NativeBranch>>::default();

            std::mem::swap(blocked.0, branches);

            if branches.is_empty() {

                log!(cu.logger(), "{:?} has become inconsistent!", proto_component_id);

                if let Some(parent) = comm.neighborhood.parent {

                    if already_requested_failure.replace_with_true() {

                        Self::request_failure(cu, comm, parent)?

                    } else {

                        log!(cu.logger(), "Already requested failure");

                    }

                } else {

                    log!(cu.logger(), "As the leader, deciding on timeout");

                    return Ok(Decision::Failure);

                }

            }

        }

        log!(cu.logger(), "All proto components are blocked");

        // ...invariant established!

        log!(cu.logger(), "Entering decision loop...");

        comm.endpoint_manager.undelay_all();

        'undecided: loop {

            // handle all buffered messages, sending them through endpoints / feeding them to components

            log!(cu.logger(), "Decision loop! have {} messages to recv", rctx.payload_inbox.len());

            while let Some((getter, send_payload_msg)) = rctx.getter_pop() {

                let getter_info = rctx.ips.port_info.map.get(&getter).unwrap();

                let cid = getter_info.owner; // the id of the component owning `getter` port

                assert_eq!(Getter, getter_info.polarity); // sanity check

                log!(

                    cu.logger(),

                    "Routing msg {:?} to {:?} via {:?}",

                    &send_payload_msg,

                    getter,

                    &getter_info.route

                );

                match getter_info.route {

                    Route::UdpEndpoint { index } => {

                        // this is a message sent over the network through a UDP endpoint

                        let udp_endpoint_ext =

                            &mut comm.endpoint_manager.udp_endpoint_store.endpoint_exts[index];

                        let SendPayloadMsg { predicate, payload } = send_payload_msg;

                        log!(cu.logger(), "Delivering to udp endpoint index={}", index);

                        // UDP mediator messages are buffered until the end of the round,

                        // because they are still speculative

                        udp_endpoint_ext.outgoing_payloads.insert(predicate, payload);

                    }

                    Route::NetEndpoint { index } => {

                        // this is a message sent over the network as a control message

                        let msg = Msg::CommMsg(CommMsg {

                            round_index: comm.round_index,

                            contents: CommMsgContents::SendPayload(send_payload_msg),

                        });

                        // actually send the message now

                        comm.endpoint_manager.send_to_comms(index, &msg)?;

                    }

                    Route::LocalComponent if cid == cu.native_component_id() => branching_native

                        .feed_msg(

                            cu,

                            rctx,

                            getter,

                            &send_payload_msg,

                            MapTempGuard::new(&mut bn_temp_owner),

                        ),

                    Route::LocalComponent => {

                        // some other component_id routed locally. must be a protocol component!

                        if let Some(branching_component) = branching_proto_components.get_mut(&cid)

                        {

                            // The recipient component is still running!

                            // Feed it this message AND run it again until all branches are blocked

                            branching_component.feed_msg(

                                cu,

                                rctx,

                                    return Ok(Decision::Failure);

                                }

                            }

                        } else {

                            // This case occurs when the component owning `getter` has exited,

                            // but the putter is still running (and sent this message).

                            // we drop the message on the floor, because it cannot be involved

                            // in a solution (requires sending a message over a dead channel!).

                            log!(

                                cu.logger(),

                                "Delivery to getter {:?} msg {:?} failed because {:?} isn't here",

                                getter,

                                &send_payload_msg,

                                cid

                            );

                        }

                    }

                }

            }

            // payload buffer is empty.

            // check if we have a solution yet

            log!(cu.logger(), "Check if we have any local decisions...");

            for solution in rctx.solution_storage.iter_new_local_make_old() {

                log!(cu.logger(), "New local decision with solution {:?}...", &solution);

                match comm.neighborhood.parent {

                    Some(parent) => {

                        // Always forward connector-local solutions to my parent

                        // AS they are moved from new->old in solution storage.

                        log!(cu.logger(), "Forwarding to my parent {:?}", parent);

                        let suggestion = Decision::Success(solution);

                        let msg = Msg::CommMsg(CommMsg {

                            round_index: comm.round_index,

                            contents: CommMsgContents::CommCtrl(CommCtrlMsg::Suggest {

                                suggestion,

                            }),

                        });

                        comm.endpoint_manager.send_to_comms(parent, &msg)?;

                    }

                    None => {

                        log!(cu.logger(), "No parent. Deciding on solution {:?}", &solution);

                        return Ok(Decision::Success(solution));

                    }

                }

            }

            // stuck! make progress by receiving a msg

            // try recv ONE message arriving through an endpoint

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