atomic_refcell = { version = "0.1.6", optional = true }

[dev-dependencies]

# test-generator = "0.3.0"

crossbeam-utils = "0.7.2"

lazy_static = "1.4.0"

[lib]

# compile target: dynamically linked library using C ABI

crate-type = ["cdylib"]

[features]

ffi = [] # see src/ffi.rs

ffi_socket_api = ["ffi", "lazy_static", "atomic_refcell"]

endpoint_logging = [] # see src/macros.rs

session_optimization = [] # see src/runtime/setup.rs

primitive pres_2(in i, out o) {

  synchronous {

  }

}

primitive together(in ia, in ib, out oa, out ob){

  while(true) synchronous {

    if(fires(ia)) {

      put(oa, get(ia));

      put(ob, get(ib));

    }

  }	

}

    offset: usize,

}

static STD_LIB_PDL: &'static [u8] = b"

primitive forward(in i, out o) {

    while(true) synchronous() put(o, get(i));

}

primitive sync(in i, out o) {

    while(true) synchronous() if(fires(i)) put(o, get(i));

}

primitive alternator_2(in i, out l, out r) {

    while(true) {

    }

}

primitive replicator_2(in i, out l, out r) {

    while(true) synchronous() if(fires(i)) {

        msg m = get(i);

        put(l, m);

        put(r, m);

    }

}

primitive merger_2(in l, in r, out o) {

    while(true) synchronous {

        if(fires(l))      put(o, get(l));

    }

}";

impl InputSource {

    // Constructors

    pub fn new<R: io::Read, S: ToString>(filename: S, reader: &mut R) -> io::Result<InputSource> {

        let mut vec = STD_LIB_PDL.to_vec();

        reader.read_to_end(&mut vec)?;

        Ok(InputSource {

            filename: filename.to_string(),

            input: vec,

            line: 1,

    new_local: HashSet<Predicate>,

    // this pair acts as SubtreeId -> HashSet<Predicate> which is friendlier to iteration

    subtree_solutions: Vec<HashSet<Predicate>>,

    subtree_id_to_index: HashMap<SubtreeId, usize>,

}

#[derive(Debug)]

struct BranchingProtoComponent {

    ports: HashSet<PortId>,

    branches: HashMap<Predicate, ProtoComponentBranch>,

}

#[derive(Debug, Clone)]

struct ProtoComponentBranch {

    inbox: HashMap<PortId, Payload>,

    state: ComponentState,

    untaken_choice: Option<u16>,

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

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

        // drains unrun_components, and populates branching_proto_components.

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

            // TODO coalesce fields

            log!(

                cu.logger,

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

                proto_component_id,

                        &predicate,

                        &branch.to_get

                    );

                }

            };

            if predicate.query(var) != Some(SpecVal::FIRING) {

                // optimization. Don't bother trying this branch

                log!(

                    cu.logger,

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

                    &predicate

                );

                continue;

            }

            use AssignmentUnionResult as Aur;

            match predicate.assignment_union(&send_payload_msg.predicate) {

                Aur::Nonexistant => {

                    // this branch does not receive the message

                    log!(

                        cu.logger,

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

                        &predicate

                    );

                }

                Aur::Equivalent | Aur::FormerNotLatter => {

                    // retain the existing predicate, but add this payload

                    feed_branch(&mut branch, &predicate);

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

                }

                Aur::LatterNotFormer => {

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

                        "payload pred {:?} covers branch pred {:?}",

                        &predicate2,

                        &predicate

                    );

                }

                Aur::New(predicate2) => {

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

                        &predicate2

                    );

                }

            }

        }

    }

        branches: &mut HashMap<Predicate, NativeBranch>,

        predicate: Predicate,

        mut branch: NativeBranch,

    ) {

        let e = branches.entry(predicate);

        use std::collections::hash_map::Entry;

        match e {

            Entry::Vacant(ev) => {

                ev.insert(branch);

            }

            Entry::Occupied(mut eo) => {

                for (k, v) in branch.gotten.drain() {

                    }

                }

            }

        }

    }

    fn collapse_with(self, logger: &mut dyn Logger, solution_predicate: &Predicate) -> RoundOk {

        log!(

            logger,

            "Collapsing native with {} branch preds {:?}",

            self.branches.len(),

            self.branches.keys()

        );

        for (branch_predicate, branch) in self.branches {

            log!(

                logger,

                "Considering native branch {:?} with to_get {:?} gotten {:?}",

        }

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

    }

}

impl BranchingProtoComponent {

    fn drain_branches_to_blocked(

        cd: CyclicDrainer<Predicate, ProtoComponentBranch>,

        cu: &mut ConnectorUnphased,

        rctx: &mut RoundCtx,

        proto_component_id: ProtoComponentId,

        ports: &HashSet<PortId>,

    ) -> Result<(), UnrecoverableSyncError> {

            let mut ctx = SyncProtoContext {

                untaken_choice: &mut branch.untaken_choice,

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

            log!(

                cu.logger,

                "Proto component with id {:?} branch with pred {:?} hit blocker {:?}",

                proto_component_id,

            use SyncBlocker as B;

            match blocker {

                B::NondetChoice { n } => {

                    let var = rctx.spec_var_stream.next();

                    for val in SpecVal::iter_domain().take(n as usize) {

                        let pred = predicate.clone().inserted(var, val);

                        let mut branch_n = branch.clone();

                        branch_n.untaken_choice = Some(val.0);

                        drainer.add_input(pred, branch_n);

                    }

                }

                B::Inconsistent => {

                    drop((predicate, branch));

                }

                B::SyncBlockEnd => {

                    // make concrete all variables

                    }

                    // submit solution for this component

                    let subtree_id = SubtreeId::LocalComponent(ComponentId::Proto(proto_component_id));

                    rctx.solution_storage.submit_and_digest_subtree_solution(

                        &mut *cu.logger,

                        subtree_id,

                        predicate.clone(),

                    );

                    branch.ended = true;

                    // move to "blocked"

                    drainer.add_output(predicate, branch);

                }

                B::PutMsg(putter, payload) => {

                    // sanity check

                    assert_eq!(Some(&Putter), cu.port_info.polarities.get(&putter));

                    // overwrite assignment

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

                    let was = predicate.assigned.insert(var, SpecVal::FIRING);

                    if was == Some(SpecVal::SILENT) {

                        log!(cu.logger, "Proto component {:?} tried to PUT on port {:?} when pred said var {:?}==Some(false). inconsistent!", proto_component_id, putter, var);

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

                        rctx.getter_buffer.putter_add(cu, putter, msg);

                        drainer.add_input(predicate, branch);

                    }

                }

            }

            Ok(())

        })

    }

    fn feed_msg(

        &mut self,

        cu: &mut ConnectorUnphased,

        rctx: &mut RoundCtx,

        proto_component_id: ProtoComponentId,

        getter: PortId,

        send_payload_msg: &SendPayloadMsg,

    ) -> Result<(), UnrecoverableSyncError> {

        let logger = &mut *cu.logger;

        log!(

            logger,

            "feeding proto component {:?} getter {:?} {:?}",

        let BranchingProtoComponent { ports, branches } = self;

        for (branch_predicate, branch) in branches {

            if branch.ended && branch_predicate.assigns_subset(solution_predicate) {

                let ProtoComponentBranch { state, .. } = branch;

                return ProtoComponent { state, ports };

            }

        }

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

    }

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

        let branch = ProtoComponentBranch {

            inbox: Default::default(),

            state,

            ended: false,

            untaken_choice: None,

        };

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

    }

}

impl SolutionStorage {

    fn new(subtree_ids: impl Iterator<Item = SubtreeId>) -> Self {

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

        let mut subtree_solutions = vec![];

        for id in subtree_ids {

    }

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

        self.inbox.get(&port)

    }

    pub(crate) fn take_choice(&mut self) -> Option<u16> {

        self.untaken_choice.take()

    }

}

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

        self.output.insert(k, v);

    }

}

impl NonsyncProtoContext<'_> {

    pub fn new_component(&mut self, moved_ports: HashSet<PortId>, state: ComponentState) {

        // called by a PROTO COMPONENT. moves its own ports.

        // 1. sanity check: this component owns these ports

        log!(

            self.logger,

            "Component {:?} added new component with state {:?}, moving ports {:?}",

            self.proto_component_id,

        let was = self.inbox.insert(getter, payload);

        assert!(was.is_none())

    }

}

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

    fn new(

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

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

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

    ) -> Self {

        Self { input, inner: CyclicDrainInner { swap, output } }

    }

        self,

        mut func: impl FnMut(K, V, CyclicDrainInner<'_, K, V>) -> Result<(), E>,

    ) -> Result<(), E> {

        let Self { input, inner: CyclicDrainInner { swap, output } } = self;

        // assert!(swap.is_empty());

        while !input.is_empty() {

            for (k, v) in input.drain() {

                func(k, v, CyclicDrainInner { swap, output })?

            }

            std::mem::swap(input, swap);

        }

        Ok(())

const SEC5: Option<Duration> = Some(Duration::from_secs(5));

const SEC15: Option<Duration> = Some(Duration::from_secs(15));

fn next_test_addr() -> SocketAddr {

    use std::{

        net::{Ipv4Addr, SocketAddrV4},

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

    };

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

    let port = TEST_PORT.fetch_add(1, SeqCst);

    SocketAddrV4::new(Ipv4Addr::LOCALHOST, port).into()

}

fn file_logged_connector(connector_id: ConnectorId, dir_path: &Path) -> Connector {

    let _ = std::fs::create_dir(dir_path); // we will check failure soon

    let path = dir_path.join(format!("cid_{:?}.txt", connector_id));

    let file = File::create(path).unwrap();

    let file_logger = Box::new(FileLogger::new(connector_id, file));

}

static MINIMAL_PDL: &'static [u8] = b"

primitive together(in ia, in ib, out oa, out ob){

  while(true) synchronous() {

    if(fires(ia)) {

      put(oa, get(ia));

      put(ob, get(ib));

    }

  } 

}

";

lazy_static::lazy_static! {

            udp.connect(sock_addrs[0]).unwrap();

            let mut buf = new_u8_buffer(256);

            udp.send(TEST_MSG_BYTES).unwrap();

            let len = udp.recv(&mut buf).unwrap();

            assert_eq!(TEST_MSG_BYTES, &buf[0..len]);

            barrier.wait();

        });

    })

    .unwrap();

}

#[test]

    let sock_addrs = [next_test_addr(), next_test_addr()];

    scope(|s| {

        s.spawn(|_| {

            // "amy"

            let mut c = file_logged_connector(0, test_log_path);

            let p0 = c.new_net_port(Putter, sock_addrs[0], Active).unwrap();

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

            c.connect(SEC1).unwrap();

            // put {A} and FAIL

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

            c.sync(SEC1).unwrap_err();

            // put {B} and FAIL

                c.get(p0).unwrap();

                c.get(p1).unwrap();

                c.sync(SEC1).unwrap_err();

            }

            // get {A, B} and SUCCEED

            c.get(p0).unwrap();

            c.get(p1).unwrap();

            c.sync(SEC1).unwrap();

        });

    })

    .unwrap();

}