CouldntReadMsg(PortId),

    CouldntCheckFiring(PortId),

    PutMsg(PortId, Payload),

}

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

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

mod eval;

pub(crate) mod inputsource;

mod lexer;

mod parser;

use crate::common::*;

impl std::fmt::Debug for ProtocolDescription {

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

    }

}

impl ProtocolDescription {

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

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

use crate::common::*;

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

struct BranchingNative {

    branches: HashMap<Predicate, NativeBranch>,

}

struct ProtoComponentBranch {

    inbox: HashMap<PortId, Payload>,

    state: ComponentState,

    ended: bool,

}

struct CyclicDrainer<'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;

}

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

        timeout: Option<Duration>,

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

        use SyncError as Se;

        log!(

            cu.logger,

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

            comm.round_index

        );

        // 1. run all proto components to Nonsync blockers

        let mut branching_proto_components =

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

            branching_proto_components.len(),

        );

        };

        // 2. kick off the native

        log!(

            "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

            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!(

                    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,

            &mut branching_native,

            &mut branching_proto_components,

        )?;

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

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

            BranchingProtoComponent::drain_branches_to_blocked(

                cd,

                cu,

                proto_component_id,

                ports,

            )?;

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

                    }

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

                        cu,

                        getter,

                        &send_payload_msg,

                    ),

                            let proto_component_id = *proto_component_id;

                            branching_component.feed_msg(

                                cu,

                                proto_component_id,

                                getter,

                                &send_payload_msg,

                            )?;

            // 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, "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 {

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

                                return Ok(Decision::Failure);

                            }

                        }

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

                    }

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

                            getter,

                        );

                    }

                    CommMsgContents::Suggest { suggestion } => {

                        // only accept this control msg through a child endpoint

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

    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,

            );

            use SyncBlocker as B;

            match blocker {

                B::Inconsistent => {

                    // branch is inconsistent. throw it away

                    drop((predicate, branch));

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

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

                    }

                }

    fn feed_msg(

        &mut self,

        cu: &mut ConnectorUnphased,

        proto_component_id: ProtoComponentId,

        getter: PortId,

        send_payload_msg: &SendPayloadMsg,

    ) -> Result<(), SyncError> {

        BranchingProtoComponent::drain_branches_to_blocked(

            cd,

            cu,

            proto_component_id,

            ports,

        )?;

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

    }

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

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

    }

}

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

        }

    }

}

        &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 }))

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

pub enum SyncError {

    NotConnected,

    InconsistentProtoComponent(ProtoComponentId),

    RoundFailure,

    PollFailed,

    BrokenEndpoint(usize),

}

pub(crate) struct SyncProtoContext<'a> {

    logger: &'a mut dyn Logger,

    predicate: &'a Predicate,

    port_info: &'a PortInfo,

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

};

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

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

fn next_test_addr() -> SocketAddr {

    use std::{

        net::{Ipv4Addr, SocketAddrV4},

fn trivial_connect() {

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

    let mut c = file_logged_connector(0, test_log_path);

}

#[test]

    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]

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

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

}

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

}

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

}

fn trivial_sync() {

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

    let mut c = file_logged_connector(0, test_log_path);

}

#[test]

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

}

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

}

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

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

}

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

    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]

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

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

    let mut c = file_logged_connector(0, test_log_path);

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

    c.get(i).unwrap();

    // getting 0 batch

    c.next_batch().unwrap();

    // silent 1 batch

    c.gotten(i).unwrap_err();

}

        s.spawn(|_| {

            let mut c = file_logged_connector(0, test_log_path);

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

            c.next_batch().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();

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

    let mut c = file_logged_connector(0, test_log_path);

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

    c.get(g).unwrap();

}

#[test]

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

    let mut c = file_logged_connector(0, test_log_path);

    let [p, g] = c.new_port_pair();

    c.get(g).unwrap();

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

    c.get(g).unwrap();

}

#[test]

    let mut c = file_logged_connector(0, test_log_path);

    let [p, g] = c.new_port_pair();

    c.add_component(b"sync", &[g, p]).unwrap();

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

    c.get(g).unwrap_err();

}

    let [p0, g0] = c.new_port_pair();

    let [p1, g1] = c.new_port_pair();

    c.add_component(b"sync", &[g0, p1]).unwrap();

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

    c.get(g1).unwrap();

    c.gotten(g1).unwrap();

}

                c.new_net_port(Putter, sock_addrs[0], Active).unwrap(),

                c.new_net_port(Getter, sock_addrs[1], Active).unwrap(),

            ];

        });

        s.spawn(|_| {

            let mut c = file_logged_connector(1, test_log_path);

                c.new_net_port(Getter, sock_addrs[0], Passive).unwrap(),

                c.new_net_port(Putter, sock_addrs[1], Passive).unwrap(),

            ];

        });

    })

    .unwrap();

                c.new_net_port(Getter, sock_addrs[1], Active).unwrap(),

            ];

            c.add_component(b"sync", &[g, p]).unwrap();

        });

        s.spawn(|_| {

            /*

                c.new_net_port(Getter, sock_addrs[0], Passive).unwrap(),

                c.new_net_port(Putter, sock_addrs[1], Passive).unwrap(),

            ];

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

            c.get(g).unwrap();

            c.gotten(g).unwrap();

        });

    })

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

    let mut c = file_logged_connector(0, test_log_path);

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

    c.get(g).unwrap();

        Err(SyncError::RoundFailure) => {}

        res => panic!("expeted timeout. but got {:?}", res),

    }

            // parent; times out

            let mut c = file_logged_connector(999, test_log_path);

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

        });

        s.spawn(|_| {

            // child

            let mut c = file_logged_connector(000, test_log_path);

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

            c.get(g).unwrap(); // not matched by put

            c.sync(None).unwrap_err(); // no timeout

        });

            // child; times out

            let mut c = file_logged_connector(000, test_log_path);

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

        });

        s.spawn(|_| {

            // parent

            let mut c = file_logged_connector(999, test_log_path);

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

            c.get(g).unwrap(); // not matched by put

            c.sync(None).unwrap_err(); // no timeout

        });

        s.spawn(|_| {

            let mut c = file_logged_connector(0, test_log_path);

            c.new_net_port(Putter, sock_addrs[0], Passive).unwrap();

        });

        s.spawn(|_| {

            let mut c = file_logged_connector(10, test_log_path);

            c.new_net_port(Getter, sock_addrs[0], Active).unwrap();

            c.new_net_port(Putter, sock_addrs[1], Passive).unwrap();

        });

        s.spawn(|_| {

            // LEADER

            let mut c = file_logged_connector(7, test_log_path);

            c.new_net_port(Getter, sock_addrs[1], Active).unwrap();

            c.new_net_port(Putter, sock_addrs[2], Passive).unwrap();

        });

        s.spawn(|_| {

            let mut c = file_logged_connector(4, test_log_path);

            c.new_net_port(Getter, sock_addrs[2], Active).unwrap();

            c.new_net_port(Putter, sock_addrs[3], Passive).unwrap();

        });

        s.spawn(|_| {

            let mut c = file_logged_connector(1, test_log_path);

            c.new_net_port(Getter, sock_addrs[3], Active).unwrap();

        });

    })

    .unwrap();

    let mut c = file_logged_connector(0, test_log_path);

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

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

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

    c.get(g).unwrap();

}

#[test]

            let p3 = c.new_net_port(Putter, sock_addrs[1], Active).unwrap();

            let [p4, p5] = c.new_port_pair();

            c.add_component(b"together", &[p1, p2, p3, p4]).unwrap();

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