pub(crate) enum SyncBlocker {

    Inconsistent,

    SyncBlockEnd,

    CouldntReadMsg(PortId),

    CouldntCheckFiring(PortId),

    PutMsg(PortId, Payload),

}

pub(crate) struct DenseDebugHex<'a>(pub &'a [u8]);

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

impl U32Stream {

    pub(crate) fn next(&mut self) -> u32 {

use crate::protocol::ast::*;

use crate::protocol::inputsource::*;

pub fn get_declarations(h: &mut Heap, i: ImportId) -> Result<Vec<DeclarationId>, ParseError> {

    if h[i].value == b"std.reo" {

        let mut vec = Vec::new();

        vec.push(cd(h, i, b"sync", &[Type::INPUT, Type::OUTPUT]));

        vec.push(cd(h, i, b"syncdrain", &[Type::INPUT, Type::INPUT]));

        vec.push(cd(h, i, b"syncspout", &[Type::OUTPUT, Type::OUTPUT]));

mod arena;

mod ast;

mod eval;

pub(crate) mod inputsource;

mod lexer;

mod parser;

use crate::common::*;

use crate::protocol::ast::*;

use crate::protocol::eval::*;

use crate::protocol::inputsource::*;

    // None,

}

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

impl std::fmt::Debug for ProtocolDescription {

    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {

    }

}

impl ProtocolDescription {

    pub fn parse(buffer: &[u8]) -> Result<Self, String> {

        let mut heap = Heap::new();

        let mut source = InputSource::from_buffer(buffer).unwrap();

use crate::protocol::ast::*;

use crate::protocol::inputsource::*;

use crate::protocol::lexer::*;

// The following indirection is needed due to a bug in the cbindgen tool.

type Unit = ();

type VisitorResult = Result<Unit, ParseError>;

trait Visitor: Sized {

        recursive_protocol_description(self, h, pd)?;

        // Move all collected declarations to the protocol description

        h[pd].declarations.append(&mut self.declarations);

        Ok(())

    }

    fn visit_import(&mut self, h: &mut Heap, import: ImportId) -> VisitorResult {

    }

    fn visit_symbol_definition(&mut self, h: &mut Heap, definition: DefinitionId) -> VisitorResult {

        let signature = Signature::from_definition(h, definition);

        let decl = h

            .alloc_defined_declaration(|this| DefinedDeclaration { this, definition, signature })

            .upcast();

use super::*;

use crate::common::*;

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

struct BranchingNative {

    branches: HashMap<Predicate, NativeBranch>,

}

#[derive(Clone, Debug)]

struct NativeBranch {

    index: usize,

    branches: HashMap<Predicate, ProtoComponentBranch>,

}

#[derive(Debug, Clone)]

struct ProtoComponentBranch {

    inbox: HashMap<PortId, Payload>,

    state: ComponentState,

    ended: bool,

}

struct CyclicDrainer<'a, K: Eq + Hash, V> {

    input: &'a mut HashMap<K, V>,

    inner: CyclicDrainInner<'a, K, V>,

}

struct CyclicDrainInner<'a, K: Eq + Hash, V> {

    swap: &'a mut HashMap<K, V>,

    output: &'a mut HashMap<K, V>,

}

trait ReplaceBoolTrue {

    fn replace_with_true(&mut self) -> bool;

}

impl ReplaceBoolTrue for bool {

    fn replace_with_true(&mut self) -> bool {

        let was = *self;

    pub fn sync(&mut self, timeout: Option<Duration>) -> Result<usize, SyncError> {

        let Self { unphased, phased } = self;

        match phased {

            ConnectorPhased::Setup { .. } => Err(SyncError::NotConnected),

            ConnectorPhased::Communication(comm) => {

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

                match &comm.round_result {

                    Ok(None) => unreachable!(),

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

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

                }

            }

    fn connected_sync(

        cu: &mut ConnectorUnphased,

        comm: &mut ConnectorCommunication,

        timeout: Option<Duration>,

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

        use SyncError as Se;

        log!(

            cu.logger,

            "~~~ SYNC called with timeout {:?}; starting round {}",

            &timeout,

            comm.round_index

        );

        // 1. run all proto components to Nonsync blockers

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

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

        log!(

            cu.logger,

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

            branching_proto_components.len(),

        );

        };

        // 2. kick off the native

        log!(

            cu.logger,

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

            comm.native_batches.len()

        );

        log!(cu.logger, "Native branch spec var is {:?}", native_branch_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;

            log!(cu.logger, "Native branch index={:?} has consistent {:?}", index, &predicate);

            // put all messages

            for (putter, payload) in to_put {

                let msg = SendPayloadMsg { predicate: predicate.clone(), payload };

                log!(cu.logger, "Native branch {} sending msg {:?}", index, &msg);

            }

            if to_get.is_empty() {

                log!(

                    cu.logger,

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

                    index,

                    &predicate

                );

                    &mut *cu.logger,

                    Route::LocalComponent(ComponentId::Native),

                    predicate.clone(),

                );

            }

            let branch = NativeBranch { index, gotten: Default::default(), to_get };

                unreachable!()

            }

        }

        // restore the invariant

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

        let decision = Self::sync_reach_decision(

            cu,

            comm,

            &mut branching_native,

            &mut branching_proto_components,

        )?;

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

        // propagate the decision to children

        let msg = Msg::CommMsg(CommMsg {

            round_index: comm.round_index,

    fn sync_reach_decision(

        cu: &mut ConnectorUnphased,

        comm: &mut ConnectorCommunication,

        branching_native: &mut BranchingNative,

        branching_proto_components: &mut HashMap<ProtoComponentId, BranchingProtoComponent>,

    ) -> Result<Decision, SyncError> {

        let mut already_requested_failure = false;

        if branching_native.branches.is_empty() {

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

            match comm.neighborhood.parent {

                Some(parent) => {

            let mut blocked = HashMap::default();

            // drain from branches --> blocked

            let cd = CyclicDrainer::new(branches, &mut swap, &mut blocked);

            BranchingProtoComponent::drain_branches_to_blocked(

                cd,

                cu,

                proto_component_id,

                ports,

            )?;

            // swap the blocked branches back

            std::mem::swap(&mut blocked, branches);

            if branches.is_empty() {

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

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

        comm.endpoint_manager.undelay_all();

        'undecided: loop {

            // drain payloads_to_get, sending them through endpoints / feeding them to components

                assert!(cu.port_info.polarities.get(&getter).copied() == Some(Getter));

                let route = cu.port_info.routes.get(&getter);

                log!(cu.logger, "Routing msg {:?} to {:?}", &send_payload_msg, &route);

                match route {

                    None => {

                        log!(

                            contents: CommMsgContents::SendPayload(send_payload_msg),

                        });

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

                    }

                    Some(Route::LocalComponent(ComponentId::Native)) => branching_native.feed_msg(

                        cu,

                        getter,

                        &send_payload_msg,

                    ),

                    Some(Route::LocalComponent(ComponentId::Proto(proto_component_id))) => {

                        if let Some(branching_component) =

                            branching_proto_components.get_mut(proto_component_id)

                        {

                            let proto_component_id = *proto_component_id;

                            branching_component.feed_msg(

                                cu,

                                proto_component_id,

                                getter,

                                &send_payload_msg,

                            )?;

                            if branching_component.branches.is_empty() {

                                log!(

                                    cu.logger,

                    }

                }

            }

            // check if we have a solution yet

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

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

                match comm.neighborhood.parent {

                    Some(parent) => {

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

                        let suggestion = Decision::Success(solution);

                        let msg = Msg::CommMsg(CommMsg {

            // 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...");

            {

                let (endpoint_index, msg) = loop {

                        None => {

                            log!(cu.logger, "Reached user-defined deadling without decision...");

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

                            }

                        }

                        Some((endpoint_index, msg)) => break (endpoint_index, msg),

                    }

                };

                log!(cu.logger, "Received from endpoint {} msg {:?}", endpoint_index, &msg);

                let comm_msg_contents = match msg {

                        assert!(cu.port_info.polarities.get(&getter) == Some(&Getter));

                        log!(

                            cu.logger,

                            "Msg routed to getter port {:?}. Buffer for recv loop",

                            getter,

                        );

                    }

                    CommMsgContents::Suggest { suggestion } => {

                        // only accept this control msg through a child endpoint

                        if comm.neighborhood.children.contains(&endpoint_index) {

                            match suggestion {

                                Decision::Success(predicate) => {

                                    // child solution contributes to local solution

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

                                    let route = Route::Endpoint { index: endpoint_index };

                                        &mut *cu.logger,

                                        route,

                                        predicate,

                                    );

                                }

                                Decision::Failure => {

//     payloads_to_get.push((getter, m));

// },

impl BranchingProtoComponent {

    fn drain_branches_to_blocked(

        cd: CyclicDrainer<Predicate, ProtoComponentBranch>,

        cu: &mut ConnectorUnphased,

        proto_component_id: ProtoComponentId,

        ports: &HashSet<PortId>,

    ) -> Result<(), SyncError> {

        cd.cylic_drain(|mut predicate, mut branch, mut drainer| {

            let mut ctx = SyncProtoContext {

                logger: &mut *cu.logger,

                predicate: &predicate,

                port_info: &cu.port_info,

                inbox: &branch.inbox,

            };

            let blocker = branch.state.sync_run(&mut ctx, &cu.proto_description);

                proto_component_id,

                &predicate,

                &blocker,

            );

            use SyncBlocker as B;

            match blocker {

                B::Inconsistent => {

                    // branch is inconsistent. throw it away

                    drop((predicate, branch));

                }

                B::SyncBlockEnd => {

                    // make concrete all variables

                    for &port in ports.iter() {

                        let var = cu.port_info.spec_var_for(port);

                        predicate.assigned.entry(var).or_insert(SpecVal::SILENT);

                    }

                    // submit solution for this component

                        &mut *cu.logger,

                        Route::LocalComponent(ComponentId::Proto(proto_component_id)),

                        predicate.clone(),

                    );

                    branch.ended = true;

                    // move to "blocked"

                        // discard forever

                        drop((predicate, branch));

                    } else {

                        // keep in "unblocked"

                        log!(cu.logger, "Proto component {:?} putting payload {:?} on port {:?} (using var {:?})", proto_component_id, &payload, putter, var);

                        let msg = SendPayloadMsg { predicate: predicate.clone(), payload };

                        drainer.add_input(predicate, branch);

                    }

                }

            }

            Ok(())

        })

    }

    fn feed_msg(

        &mut self,

        cu: &mut ConnectorUnphased,

        proto_component_id: ProtoComponentId,

        getter: PortId,

        send_payload_msg: &SendPayloadMsg,

    ) -> Result<(), SyncError> {

        let logger = &mut *cu.logger;

        log!(

            logger,

        // drain from unblocked --> blocked

        let mut swap = HashMap::default();

        let cd = CyclicDrainer::new(&mut unblocked, &mut swap, &mut blocked);

        BranchingProtoComponent::drain_branches_to_blocked(

            cd,

            cu,

            proto_component_id,

            ports,

        )?;

        // swap the blocked branches back

        std::mem::swap(&mut blocked, branches);

        log!(cu.logger, "component settles down with branches: {:?}", branches.keys());

                return ProtoComponent { state, ports };

            }

        }

        panic!("ProtoComponent had no branches matching pred {:?}", solution_predicate);

    }

    fn initial(ProtoComponent { state, ports }: ProtoComponent) -> Self {

        Self { ports, branches: hashmap! { Predicate::default() => branch  } }

    }

}

impl SolutionStorage {

    fn new(routes: impl Iterator<Item = Route>) -> Self {

        let mut subtree_id_to_index: HashMap<Route, usize> = Default::default();

        self.new_local.clear();

        for key in subtree_ids {

            self.subtree_id_to_index.insert(key, self.subtree_solutions.len());

            self.subtree_solutions.push(Default::default())

        }

    }

    pub(crate) fn iter_new_local_make_old(&mut self) -> impl Iterator<Item = Predicate> + '_ {

        let Self { old_local, new_local, .. } = self;

        new_local.drain().map(move |local| {

            old_local.insert(local.clone());

            local

        })

                log!(logger, "storing NEW LOCAL SOLUTION {:?}", &partial);

                new_local.insert(partial);

            }

        }

    }

}

        &mut self,

        cu: &mut ConnectorUnphased,

        putter: PortId,

        msg: SendPayloadMsg,

    ) -> Result<(), SyncError> {

        if let Some(&getter) = cu.port_info.peers.get(&putter) {

            Ok(())

        } else {

            Err(SyncError::MalformedStateError(MalformedStateError::GetterUnknownFor { putter }))

        }

    }

}

        let var = self.port_info.spec_var_for(port);

        self.predicate.query(var).map(SpecVal::is_firing)

    }

    pub(crate) fn read_msg(&mut self, port: PortId) -> Option<&Payload> {

        self.inbox.get(&port)

    }

}

impl<'a, K: Eq + Hash, V> CyclicDrainInner<'a, K, V> {

    fn add_input(&mut self, k: K, v: V) {

        self.swap.insert(k, v);

    }

    fn add_output(&mut self, k: K, v: V) {

}

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

#[derive(Debug, Clone)]

pub enum SyncError {

    NotConnected,

    InconsistentProtoComponent(ProtoComponentId),

    RoundFailure,

    PollFailed,

    BrokenEndpoint(usize),

    MalformedStateError(MalformedStateError),

}

#[derive(Debug, Clone)]

    id_manager: &'a mut IdManager,

    proto_component_ports: &'a mut HashSet<PortId>,

    unrun_components: &'a mut Vec<(ProtoComponentId, ProtoComponent)>,

}

pub(crate) struct SyncProtoContext<'a> {

    logger: &'a mut dyn Logger,

    predicate: &'a Predicate,

    port_info: &'a PortInfo,

    inbox: &'a HashMap<PortId, Payload>,

}

#[derive(

    EndpointPolarity::{Active, Passive},

    Polarity::{Getter, Putter},

    *,

};

use std::{fs::File, net::SocketAddr, path::Path, sync::Arc, time::Duration};

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

fn next_test_addr() -> SocketAddr {

    use std::{

        net::{Ipv4Addr, SocketAddrV4},

        sync::atomic::{AtomicU16, Ordering::SeqCst},

    };

    static TEST_PORT: AtomicU16 = AtomicU16::new(5_000);

}

#[test]

fn trivial_connect() {

    let test_log_path = Path::new("./logs/trivial_connect");

    let mut c = file_logged_connector(0, test_log_path);

}

#[test]

fn single_node_connect() {

    let sock_addr = next_test_addr();

    let test_log_path = Path::new("./logs/single_node_connect");

    let mut c = file_logged_connector(0, test_log_path);

    let _ = c.new_net_port(Getter, sock_addr, Passive).unwrap();

    let _ = c.new_net_port(Putter, sock_addr, Active).unwrap();

}

#[test]

fn minimal_net_connect() {

    let sock_addr = next_test_addr();

    let test_log_path = Path::new("./logs/minimal_net_connect");

    scope(|s| {

        s.spawn(|_| {

            let mut c = file_logged_connector(0, test_log_path);

            let _ = c.new_net_port(Getter, sock_addr, Active).unwrap();

        });

        s.spawn(|_| {

            let mut c = file_logged_connector(1, test_log_path);

            let _ = c.new_net_port(Putter, sock_addr, Passive).unwrap();

        });

    })

    .unwrap();

}

#[test]

fn put_no_sync() {

    let test_log_path = Path::new("./logs/put_no_sync");

    let mut c = file_logged_connector(0, test_log_path);

    let [o, _] = c.new_port_pair();

    c.put(o, TEST_MSG.clone()).unwrap();

}

#[test]

fn wrong_polarity_bad() {

    let test_log_path = Path::new("./logs/wrong_polarity_bad");

    let mut c = file_logged_connector(0, test_log_path);

    let [_, i] = c.new_port_pair();

    c.put(i, TEST_MSG.clone()).unwrap_err();

}

#[test]

fn dup_put_bad() {

    let test_log_path = Path::new("./logs/dup_put_bad");

    let mut c = file_logged_connector(0, test_log_path);

    let [o, _] = c.new_port_pair();

    c.put(o, TEST_MSG.clone()).unwrap();

    c.put(o, TEST_MSG.clone()).unwrap_err();

}

#[test]

fn trivial_sync() {

    let test_log_path = Path::new("./logs/trivial_sync");

    let mut c = file_logged_connector(0, test_log_path);

}

#[test]

fn unconnected_gotten_err() {

    let test_log_path = Path::new("./logs/unconnected_gotten_err");

    let mut c = file_logged_connector(0, test_log_path);

#[test]

fn connected_gotten_err_no_round() {

    let test_log_path = Path::new("./logs/connected_gotten_err_no_round");

    let mut c = file_logged_connector(0, test_log_path);

    let [_, i] = c.new_port_pair();

    assert_eq!(reowolf::error::GottenError::NoPreviousRound, c.gotten(i).unwrap_err());

}

#[test]

fn connected_gotten_err_ungotten() {

    let test_log_path = Path::new("./logs/connected_gotten_err_ungotten");

    let mut c = file_logged_connector(0, test_log_path);

    let [_, i] = c.new_port_pair();

    assert_eq!(reowolf::error::GottenError::PortDidntGet, c.gotten(i).unwrap_err());

}

#[test]

fn native_polarity_checks() {

    let test_log_path = Path::new("./logs/native_polarity_checks");

    let mut c = file_logged_connector(0, test_log_path);

    let [o, i] = c.new_port_pair();

    // fail...

    c.get(o).unwrap_err();

    c.put(i, TEST_MSG.clone()).unwrap_err();

    // succeed..

    c.get(i).unwrap();

    c.put(o, TEST_MSG.clone()).unwrap();

#[test]

fn native_multiple_gets() {

    let test_log_path = Path::new("./logs/native_multiple_gets");

    let mut c = file_logged_connector(0, test_log_path);

    let [_, i] = c.new_port_pair();

    c.get(i).unwrap();

    c.get(i).unwrap_err();

}

#[test]

fn next_batch() {

    let test_log_path = Path::new("./logs/next_batch");

    let mut c = file_logged_connector(0, test_log_path);

    c.next_batch().unwrap_err();

    c.next_batch().unwrap();

    c.next_batch().unwrap();

    c.next_batch().unwrap();

}

#[test]

fn native_self_msg() {

    let test_log_path = Path::new("./logs/native_self_msg");

    let mut c = file_logged_connector(0, test_log_path);

    let [o, i] = c.new_port_pair();

    c.get(i).unwrap();

    c.put(o, TEST_MSG.clone()).unwrap();

}

#[test]

fn two_natives_msg() {

    let test_log_path = Path::new("./logs/two_natives_msg");

    let sock_addr = next_test_addr();

    scope(|s| {

        s.spawn(|_| {

            let mut c = file_logged_connector(0, test_log_path);

            let g = c.new_net_port(Getter, sock_addr, Active).unwrap();

            c.get(g).unwrap();

            c.gotten(g).unwrap();

        });

        s.spawn(|_| {

            let mut c = file_logged_connector(1, test_log_path);

            let p = c.new_net_port(Putter, sock_addr, Passive).unwrap();

            c.put(p, TEST_MSG.clone()).unwrap();

        });

    })

    .unwrap();