[package]

name = "reowolf_rs"

version = "0.1.0"

edition = "2018"

[dependencies]

getrandom = "0.1.14" # tiny crate. used to guess controller-id

take_mut = "0.2.2"

maplit = "1.0.2" # convenience macros

# network stuff

integer-encoding = "1.0.7"

byteorder = "1.3.2"

mio = "0.6.21" # migrate to mio 0.7.0 when it stabilizes. It's much better.

mio-extras = "2.0.6"

# protocol stuff

id-arena = "2.2.1"

backtrace = "0.3"

[dev-dependencies]

test-generator = "0.3.0"

[lib]

crate-type = ["cdylib"]

[features]

default = ["ffi"]

macro_rules! assert_let {

    ($pat:pat = $expr:expr => $work:expr) => {

        if let $pat = $expr {

            $work

        } else {

            panic!("assert_let failed");

        }

    };

}

#[test]

fn assert_let() {

    let x = Some(5);

    let z = assert_let![Some(y) = x => {

        println!("{:?}", y);

        3

    }];

    println!("{:?}", z);

}

#[test]

#[should_panic]

fn must_let_panic() {

    let x: Option<u32> = None;

    assert_let![Some(y) = x => {

        println!("{:?}", y);

    }];

}

    }

    pub(crate) fn poly_run_these_branches(

        &mut self,

        mut m_ctx: PolyPContext,

        protocol_description: &ProtocolD,

        mut to_run: Vec<(Predicate, BranchP)>,

    ) -> Result<SyncRunResult, EndpointErr> {

        use SyncRunResult as Srr;

        while let Some((mut predicate, mut branch)) = to_run.pop() {

            let mut r_ctx = BranchPContext {

                m_ctx: m_ctx.reborrow(),

                ekeys: &self.ekeys,

                predicate: &predicate,

                inbox: &branch.inbox,

            };

            use PolyBlocker as Sb;

            let blocker = branch.state.sync_run(&mut r_ctx, protocol_description);

            match blocker {

                Sb::Inconsistent => {} // DROP

                Sb::CouldntReadMsg(ekey) => {

                    assert!(self.ekeys.contains(&ekey));

                    let channel_id =

                        r_ctx.m_ctx.inner.endpoint_exts.get(ekey).unwrap().info.channel_id;

                    if predicate.replace_assignment(channel_id, true) != Some(false) {

                        // don't rerun now. Rerun at next `sync_run`

                        self.incomplete.insert(predicate, branch);

                    }

                    // ELSE DROP

                }

                Sb::CouldntCheckFiring(ekey) => {

                    assert!(self.ekeys.contains(&ekey));

                    let channel_id =

                        r_ctx.m_ctx.inner.endpoint_exts.get(ekey).unwrap().info.channel_id;

                    // split the branch!

                    let branch_f = branch.clone();

                    let mut predicate_f = predicate.clone();

                    if predicate_f.replace_assignment(channel_id, false).is_some() {

                        panic!("OI HANS QUERY FIRST!");

                    }

                    assert!(predicate.replace_assignment(channel_id, true).is_none());

                    to_run.push((predicate, branch));

                    to_run.push((predicate_f, branch_f));

                }

                Sb::SyncBlockEnd => {

                    // come up with the predicate for this local solution

                    let ekeys_channel_id_iter = self

                        .ekeys

                        .iter()

                        .map(|&ekey| m_ctx.inner.endpoint_exts.get(ekey).unwrap().info.channel_id);

                    predicate.batch_assign_nones(ekeys_channel_id_iter, false);

                    // report the local solution

                    m_ctx

                        .solution_storage

                        .submit_and_digest_subtree_solution(m_ctx.my_subtree_id, predicate.clone());

                    // store the solution for recovering later

                    self.complete.insert(predicate, branch);

                }

                Sb::PutMsg(ekey, payload) => {

                    assert!(self.ekeys.contains(&ekey));

                    let EndpointExt { info, endpoint } =

                        m_ctx.inner.endpoint_exts.get_mut(ekey).unwrap();

                    if predicate.replace_assignment(info.channel_id, true) != Some(false) {

                        let msg = CommMsgContents::SendPayload {

                            payload_predicate: predicate.clone(),

                            payload,

                        }

                        .into_msg(m_ctx.inner.round_index);

                        endpoint.send(msg)?;

                        to_run.push((predicate, branch));

                    }

                    // ELSE DROP

                }

            }

        }

        // all in self.incomplete most recently returned Blocker::CouldntReadMsg

        Ok(if self.incomplete.is_empty() {

            if self.complete.is_empty() {

                Srr::NoBranches

            } else {

                Srr::AllBranchesComplete

            }

        } else {

            Srr::BlockingForRecv

        })

    }

    pub(crate) fn poly_recv_run(

        &mut self,

        m_ctx: PolyPContext,

        protocol_description: &ProtocolD,

        ekey: Key,

        payload_predicate: Predicate,

        payload: Payload,

    ) -> Result<SyncRunResult, EndpointErr> {

        // try exact match

        let to_run = if self.complete.contains_key(&payload_predicate) {

            // exact match with stopped machine

            vec![]

        } else if let Some(mut branch) = self.incomplete.remove(&payload_predicate) {

            // exact match with running machine

            branch.inbox.insert(ekey, payload);

            vec![(payload_predicate, branch)]

        } else {

            let mut incomplete2 = HashMap::<_, _>::default();

            let to_run = self

                .incomplete

                .drain()

                .filter_map(|(old_predicate, mut branch)| {

                    use CommonSatResult as Csr;

                    match old_predicate.common_satisfier(&payload_predicate) {

                        Csr::FormerNotLatter | Csr::Equivalent => {

                            // old_predicate COVERS the assumptions of payload_predicate

                            let was = branch.inbox.insert(ekey, payload.clone());

                            assert!(was.is_none()); // INBOX MUST BE EMPTY!

                            Some((old_predicate, branch))

                        }

                            // payload_predicate has new assumptions. FORK!

                            let mut payload_branch = branch.clone();

                            let was = payload_branch.inbox.insert(ekey, payload.clone());

                            assert!(was.is_none()); // INBOX MUST BE EMPTY!

                            // put the original back untouched

                            incomplete2.insert(old_predicate, branch);

                        }

                        Csr::LatterNotFormer => {

                            // payload_predicate has new assumptions. FORK!

                            let mut payload_branch = branch.clone();

                            let was = payload_branch.inbox.insert(ekey, payload.clone());

                            assert!(was.is_none()); // INBOX MUST BE EMPTY!

                            // put the original back untouched

                            incomplete2.insert(old_predicate, branch);

                            Some((payload_predicate.clone(), payload_branch))

                        }

                        Csr::Nonexistant => {

                            // predicates contradict

                            incomplete2.insert(old_predicate, branch);

                            None

                        }

                    }

                })

                .collect();

            std::mem::swap(&mut self.incomplete, &mut incomplete2);

            to_run

        };

        self.poly_run_these_branches(m_ctx, protocol_description, to_run)

    }

    pub(crate) fn become_mono(

        mut self,

        decision: &Predicate,

        all_inboxes: &mut HashMap<Key, Payload>,

    ) -> MonoP {

        if let Some((_, branch)) = self.complete.drain().find(|(p, _)| decision.satisfies(p)) {

            let BranchP { inbox, state } = branch;

            for (key, payload) in inbox {

                assert!(all_inboxes.insert(key, payload).is_none());

            }

            self.incomplete.clear();

            MonoP { state, ekeys: self.ekeys }

        } else {

            panic!("No such solution!")

        }

    }

}

impl PolyN {

    pub fn sync_recv(

        &mut self,

        ekey: Key,

        payload: Payload,

        solution_storage: &mut SolutionStorage,

    ) {

        for (predicate, branch) in self.branches.iter_mut() {

            if branch.to_get.remove(&ekey) {

                branch.gotten.insert(ekey, payload.clone());

                if branch.to_get.is_empty() {

                    solution_storage

                        .submit_and_digest_subtree_solution(SubtreeId::PolyN, predicate.clone());

                }

            }

        }

    }

    pub fn become_mono(

        mut self,

        decision: &Predicate,

        all_inboxes: &mut HashMap<Key, Payload>,

    ) -> MonoN {

        if let Some((_, branch)) = self.branches.drain().find(|(p, _)| decision.satisfies(p)) {

            let BranchN { gotten, sync_batch_index, .. } = branch;

            for (&key, payload) in gotten.iter() {

                assert!(all_inboxes.insert(key, payload.clone()).is_none());

            }

            MonoN { ekeys: self.ekeys, result: Some((sync_batch_index, gotten)) }

        } else {

            panic!("No such solution!")

        }

    }

}

use crate::common::*;

use crate::runtime::{actors::*, endpoint::*, errors::*, *};

impl Controller {

    fn end_round_with_decision(&mut self, decision: Predicate) -> Result<(), SyncErr> {

        let mut all_inboxes = HashMap::default();

        self.inner.mono_n = self

            .ephemeral

            .poly_n

            .take()

            .map(|poly_n| poly_n.become_mono(&decision, &mut all_inboxes));

        self.inner.mono_ps.extend(

            self.ephemeral.poly_ps.drain(..).map(|m| m.become_mono(&decision, &mut all_inboxes)),

        );

        let valuations: HashMap<_, _> = all_inboxes

            .drain()

            .map(|(ekey, payload)| {

                let channel_id = self.inner.endpoint_exts.get(ekey).unwrap().info.channel_id;

                (channel_id, Some(payload))

            })

            .collect();

        for (channel_id, value) in decision.assigned.iter() {

            if !value {

                println!("VALUE {:?} => *", channel_id);

            } else if let Some(payload) = valuations.get(channel_id) {

                println!("VALUE {:?} => Message({:?})", channel_id, payload);

            } else {

                println!("VALUE {:?} => Message(?)", channel_id);

            }

        }

        let announcement =

            CommMsgContents::Announce { oracle: decision }.into_msg(self.inner.round_index);

        for &child_ekey in self.inner.family.children_ekeys.iter() {

            self.inner

                .endpoint_exts

                .get_mut(child_ekey)

                .expect("eefef")

                .endpoint

                .send(announcement.clone())?;

        }

        self.inner.round_index += 1;

        self.ephemeral.clear();

        Ok(())

    }

    // Drain self.ephemeral.solution_storage and handle the new locals. Return decision if one is found

    fn handle_locals_maybe_decide(&mut self) -> Result<bool, SyncErr> {

        if let Some(parent_ekey) = self.inner.family.parent_ekey {

            // I have a parent -> I'm not the leader

            let parent_endpoint =

                &mut self.inner.endpoint_exts.get_mut(parent_ekey).expect("huu").endpoint;

            for partial_oracle in self.ephemeral.solution_storage.iter_new_local_make_old() {

                let msg =

                    CommMsgContents::Elaborate { partial_oracle }.into_msg(self.inner.round_index);

                parent_endpoint.send(msg)?;

            }

            Ok(false)

        } else {

            // I have no parent -> I'm the leader

            assert!(self.inner.family.parent_ekey.is_none());

            let maybe_decision = self.ephemeral.solution_storage.iter_new_local_make_old().next();

            Ok(if let Some(decision) = maybe_decision {

                self.end_round_with_decision(decision)?;

                true

            } else {

                false

            })

        }

    }

    fn kick_off_native(

        &mut self,

        sync_batches: impl Iterator<Item = SyncBatch>,

    ) -> Result<PolyN, EndpointErr> {

        let MonoN { ekeys, .. } = self.inner.mono_n.take().unwrap();

        let Self { inner: ControllerInner { endpoint_exts, round_index, .. }, .. } = self;

        let mut branches = HashMap::<_, _>::default();

        for (sync_batch_index, SyncBatch { puts, gets }) in sync_batches.enumerate() {

            let ekey_to_channel_id = |ekey| endpoint_exts.get(ekey).unwrap().info.channel_id;

            let all_ekeys = ekeys.iter().copied();

            let all_channel_ids = all_ekeys.map(ekey_to_channel_id);

            let mut predicate = Predicate::new_trivial();

            // assign TRUE for puts and gets

            let true_ekeys = puts.keys().chain(gets.iter()).copied();

            let true_channel_ids = true_ekeys.clone().map(ekey_to_channel_id);

            predicate.batch_assign_nones(true_channel_ids, true);

            // assign FALSE for all in interface not assigned true

            predicate.batch_assign_nones(all_channel_ids.clone(), false);

            if branches.contains_key(&predicate) {

                // TODO what do I do with redundant predicates?

                unimplemented!(

                    "Having multiple batches with the same

                    predicate requires the support of oracle boolean variables"

                )

            }

    /// Given self and other, two predicates, return the most general Predicate possible, N

    /// such that n.satisfies(self) && n.satisfies(other).

    /// If none exists Nonexistant is returned.

    /// If the resulting predicate is equivlanet to self, other, or both,

    /// FormerNotLatter, LatterNotFormer and Equivalent are returned respectively.

    /// otherwise New(N) is returned.

    pub fn common_satisfier(&self, other: &Self) -> CommonSatResult {

        use CommonSatResult::*;

        // iterators over assignments of both predicates. Rely on SORTED ordering of BTreeMap's keys.

        let [mut s_it, mut o_it] = [self.assigned.iter(), other.assigned.iter()];

        let [mut s, mut o] = [s_it.next(), o_it.next()];

        // lists of assignments in self but not other and vice versa.

        let [mut s_not_o, mut o_not_s] = [vec![], vec![]];

        loop {

            match [s, o] {

                [None, None] => break,

                [None, Some(x)] => {

                    o_not_s.push(x);

                    o_not_s.extend(o_it);

                    break;

                }

                [Some(x), None] => {

                    s_not_o.push(x);

                    s_not_o.extend(s_it);

                    break;

                }

                [Some((sid, sb)), Some((oid, ob))] => {

                    if sid < oid {

                        // o is missing this element

                        s_not_o.push((sid, sb));

                        s = s_it.next();

                    } else if sid > oid {

                        // s is missing this element

                        o_not_s.push((sid, sb));

                        o = o_it.next();

                    } else if sb != ob {

                        assert_eq!(sid, oid);

                        // both predicates assign the variable but differ on the value

                        return Nonexistant;

                    } else {

                        // both predicates assign the variable to the same value

                        s = s_it.next();

                        o = o_it.next();

                    }

                }

            }

        }

        // Observed zero inconsistencies. A unified predicate exists...

        match [s_not_o.is_empty(), o_not_s.is_empty()] {

            [true, true] => Equivalent,       // ... equivalent to both.

            [false, true] => FormerNotLatter, // ... equivalent to self.

            [true, false] => LatterNotFormer, // ... equivalent to other.

            [false, false] => {

                // ... which is the union of the predicates' assignments but

                //     is equivalent to neither self nor other.

                let mut predicate = self.clone();

                for (&id, &b) in o_not_s {

                    predicate.assigned.insert(id, b);

                }

                New(predicate)

            }

        }

    }

    pub fn batch_assign_nones(

        &mut self,

        channel_ids: impl Iterator<Item = ChannelId>,

        value: bool,

    ) {

        for channel_id in channel_ids {

            self.assigned.entry(channel_id).or_insert(value);

        }

    }

    pub fn replace_assignment(&mut self, channel_id: ChannelId, value: bool) -> Option<bool> {

        self.assigned.insert(channel_id, value)

    }

    pub fn union_with(&self, other: &Self) -> Option<Self> {

        let mut res = self.clone();

        for (&channel_id, &assignment_1) in other.assigned.iter() {

            match res.assigned.insert(channel_id, assignment_1) {

                Some(assignment_2) if assignment_1 != assignment_2 => return None,

                _ => {}

            }

        }

        Some(res)

    }

    pub fn query(&self, x: ChannelId) -> Option<bool> {

        self.assigned.get(&x).copied()

    }

    pub fn new_trivial() -> Self {

        Self { assigned: Default::default() }

    }

}

impl Debug for Predicate {

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

        for (ChannelId { controller_id, channel_index }, &v) in self.assigned.iter() {

        }

        Ok(())

    }

}

#[test]

fn pred_sat() {

    use maplit::btreemap;

    let mut c = ChannelIdStream::new(0);

    let ch = std::iter::repeat_with(move || c.next()).take(5).collect::<Vec<_>>();

    let p = Predicate::new_trivial();

    let p_0t = Predicate { assigned: btreemap! { ch[0] => true } };

    let p_0f = Predicate { assigned: btreemap! { ch[0] => false } };

    let p_0f_3f = Predicate { assigned: btreemap! { ch[0] => false, ch[3] => false } };

    let p_0f_3t = Predicate { assigned: btreemap! { ch[0] => false, ch[3] => true } };

    assert!(p.satisfies(&p));

    assert!(p_0t.satisfies(&p_0t));

    assert!(p_0f.satisfies(&p_0f));

    assert!(p_0f_3f.satisfies(&p_0f_3f));

    assert!(p_0f_3t.satisfies(&p_0f_3t));

    assert!(p_0t.satisfies(&p));

    assert!(p_0f.satisfies(&p));

    assert!(p_0f_3f.satisfies(&p_0f));

    assert!(p_0f_3t.satisfies(&p_0f));

    assert!(!p.satisfies(&p_0t));

    assert!(!p.satisfies(&p_0f));

    assert!(!p_0f.satisfies(&p_0t));

    assert!(!p_0t.satisfies(&p_0f));

    assert!(!p_0f_3f.satisfies(&p_0f_3t));

    assert!(!p_0f_3t.satisfies(&p_0f_3f));

    assert!(!p_0t.satisfies(&p_0f_3f));

    assert!(!p_0f.satisfies(&p_0f_3f));

    assert!(!p_0t.satisfies(&p_0f_3t));

    assert!(!p_0f.satisfies(&p_0f_3t));

}

#[test]

fn pred_common_sat() {

    use maplit::btreemap;

    use CommonSatResult::*;

    let mut c = ChannelIdStream::new(0);

    let ch = std::iter::repeat_with(move || c.next()).take(5).collect::<Vec<_>>();

    let p = Predicate::new_trivial();

    let p_0t = Predicate { assigned: btreemap! { ch[0] => true } };

    let p_0f = Predicate { assigned: btreemap! { ch[0] => false } };

    let p_3f = Predicate { assigned: btreemap! { ch[3] => false } };

    let p_0f_3f = Predicate { assigned: btreemap! { ch[0] => false, ch[3] => false } };

    let p_0f_3t = Predicate { assigned: btreemap! { ch[0] => false, ch[3] => true } };

    assert_eq![p.common_satisfier(&p), Equivalent];

    assert_eq![p_0t.common_satisfier(&p_0t), Equivalent];

    assert_eq![p.common_satisfier(&p_0t), LatterNotFormer];

    assert_eq![p_0t.common_satisfier(&p), FormerNotLatter];

    assert_eq![p_0t.common_satisfier(&p_0f), Nonexistant];

    assert_eq![p_0f_3t.common_satisfier(&p_0f_3f), Nonexistant];

    assert_eq![p_0f_3t.common_satisfier(&p_3f), Nonexistant];

    assert_eq![p_3f.common_satisfier(&p_0f_3t), Nonexistant];

    assert_eq![p_0f.common_satisfier(&p_3f), New(p_0f_3f)];

}