// 3. define the mapping from ekey -> actor

        //    this is needed during the event loop to determine which actor

        //    should receive the incoming message.

        //    TODO: store and update this mapping rather than rebuilding it each round.

        let ekey_to_holder: HashMap<Key, PolyId> = {

            use PolyId::*;

            let n = self.inner.mono_n.ekeys.iter().map(move |&e| (e, N));

            let p = self

                .ephemeral

                .poly_ps

                .iter()

                .enumerate()

                .flat_map(|(index, m)| m.ekeys.iter().map(move |&e| (e, P { index })));

            n.chain(p).collect()

        };

        log!(

            &mut self.inner.logger,

            "SET OF PolyPs and MonoPs final! ekey lookup map is {:?}",

            &ekey_to_holder

        );

        // 4. Create the solution storage. it tracks the solutions of "subtrees"

        //    of the controller in the overlay tree.

        self.ephemeral.solution_storage.reset({

            let n = std::iter::once(SubtreeId::PolyN);

            let m = (0..self.ephemeral.poly_ps.len()).map(|index| SubtreeId::PolyP { index });

            let c = self

                .inner

                .family

                .children_ekeys

                .iter()

                .map(|&ekey| SubtreeId::ChildController { ekey });

            let subtree_id_iter = n.chain(m).chain(c);

            log!(

                &mut self.inner.logger,

                "Solution Storage has subtree Ids: {:?}",

                &subtree_id_iter.clone().collect::<Vec<_>>()

            );

            subtree_id_iter

        });

        // 5. kick off the synchronous round of the native actor if it exists

        log!(&mut self.inner.logger, "Kicking off native's synchronous round...");

        self.ephemeral.poly_n = if let Some(sync_batches) = sync_batches {

            // using if let because of nested ? operator

            // TODO check that there are 1+ branches or NO SOLUTION

            let poly_n = self.kick_off_native(sync_batches)?;

            log!(

                &mut self.inner.logger,

                "PolyN kicked off, and has branches with predicates... {:?}",

                poly_n.branches.keys().collect::<Vec<_>>()

            );

            Some(poly_n)

        } else {

            log!(&mut self.inner.logger, "NO NATIVE COMPONENT");

            None

        };

        // 6. Kick off the synchronous round of each protocol actor

        //    If just one actor becomes inconsistent now, there can be no solution!

        //    TODO distinguish between completed and not completed poly_p's?

        log!(&mut self.inner.logger, "Kicking off {} PolyP's.", self.ephemeral.poly_ps.len());

        for (index, poly_p) in self.ephemeral.poly_ps.iter_mut().enumerate() {

            let my_subtree_id = SubtreeId::PolyP { index };

            let m_ctx = PolyPContext {

                my_subtree_id,

                inner: &mut self.inner,

                solution_storage: &mut self.ephemeral.solution_storage,

            };

            use SyncRunResult as Srr;

            let blocker = poly_p.poly_run(m_ctx, &self.protocol_description)?;

            log!(&mut self.inner.logger, "... PolyP's poly_run got blocker {:?}", &blocker);

            match blocker {

                Srr::NoBranches => return Err(SyncErr::Inconsistent),

                Srr::AllBranchesComplete | Srr::BlockingForRecv => (),

            }

        }

        log!(&mut self.inner.logger, "All Poly machines have been kicked off!");

        // 7. `solution_storage` may have new solutions for this controller

        //    handle their discovery. LEADER => announce, otherwise => send to parent

        {

            let peeked = self.ephemeral.solution_storage.peek_new_locals().collect::<Vec<_>>();

            log!(

                &mut self.inner.logger,

                "Got {} controller-local solutions before a single RECV: {:?}",

                peeked.len(),

                peeked

            );

        }

        if self.handle_locals_maybe_decide()? {

            return Ok(());

        }

        log!(&mut self.inner.logger, "`No decision yet`. Time to recv messages");

        self.undelay_all();

        'recv_loop: loop {

            log!(&mut self.inner.logger, "`POLLING` with deadline {:?}...", deadline);

            let received = match deadline {

                Some(d) => match self.recv(d)? {

                    Some(received) => received,

                    None => {

                        deadline = None;

                        match self.inner.family.parent_ekey {

                            Some(parent_ekey) => {

                                let announcement = Msg::CommMsg(CommMsg {

                                    round_index: self.inner.round_index,

                                    contents: CommMsgContents::Failure,

                                });

                                log!(

                                    &mut self.inner.logger,

                                    "Forwarding {:?} to parent with ekey {:?}",

                                    &announcement,

                                    parent_ekey

                                );

                                self.inner

                                    .endpoint_exts

                                    .get_mut(parent_ekey)

                                    .expect("ss")

                                    .endpoint

                                    .send(announcement.clone())?;

                            }

                        }

                    }

                },

            };

            log!(&mut self.inner.logger, "::: message {:?}...", &received);

            let current_content = match received.msg {

                Msg::SetupMsg(s) => {

                    // This occurs in the event the connector was malformed during connect()

                    println!("WASNT EXPECTING {:?}", s);

                    return Err(SyncErr::UnexpectedSetupMsg);

                }

                Msg::CommMsg(CommMsg { round_index, .. })

                    if round_index < self.inner.round_index =>

                {

                    // Old message! Can safely discard

                    log!(&mut self.inner.logger, "...and its OLD! :(");

                    drop(received);

                    continue 'recv_loop;

                }

                Msg::CommMsg(CommMsg { round_index, .. })

                    if round_index > self.inner.round_index =>

                {

                    // Message from a next round. Keep for later!

                    log!(&mut self.inner.logger, "... DELAY! :(");

                    self.delay(received);

                    continue 'recv_loop;

                }

                Msg::CommMsg(CommMsg { contents, round_index }) => {

                    log!(

                        &mut self.inner.logger,

                        "... its a round-appropriate CommMsg with key {:?}",

                        received.recipient

                    );

                    assert_eq!(round_index, self.inner.round_index);

                    contents

                }

            };

            match current_content {

                CommMsgContents::Failure => match self.inner.family.parent_ekey {

                    Some(parent_ekey) => {

                        let announcement = Msg::CommMsg(CommMsg {

                            round_index: self.inner.round_index,

                            contents: CommMsgContents::Failure,

                        });

                        log!(

                            &mut self.inner.logger,

                            "Forwarding {:?} to parent with ekey {:?}",

                            &announcement,

                            parent_ekey

                        );

                        self.inner

                            .endpoint_exts

                            .get_mut(parent_ekey)

                            .expect("ss")

                            .endpoint

                            .send(announcement.clone())?;

                    }

                    None => return self.end_round_with_decision(Decision::Failure),

                },

                CommMsgContents::Elaborate { partial_oracle } => {

                    // Child controller submitted a subtree solution.

                    if !self.inner.family.children_ekeys.contains(&received.recipient) {

                        return Err(SyncErr::ElaborateFromNonChild);

                    }

                    let subtree_id = SubtreeId::ChildController { ekey: received.recipient };

                    log!(

                        &mut self.inner.logger,

                        "Received elaboration from child for subtree {:?}: {:?}",

                        subtree_id,

                        &partial_oracle

                    );

                    self.ephemeral.solution_storage.submit_and_digest_subtree_solution(

                        &mut self.inner.logger,

                        subtree_id,

                        partial_oracle,

                    );

                    if self.handle_locals_maybe_decide()? {

                        return Ok(());

                    }

                }

                CommMsgContents::Announce { decision } => {

                    if self.inner.family.parent_ekey != Some(received.recipient) {

                        return Err(SyncErr::AnnounceFromNonParent);

                    }

                    log!(

                        &mut self.inner.logger,

                        "Received ANNOUNCEMENT from from parent {:?}: {:?}",

                        received.recipient,

                        &decision

                    );

                    return self.end_round_with_decision(decision);

                }

                CommMsgContents::SendPayload { payload_predicate, payload } => {

                    assert_eq!(

                        Getter,

                        self.inner.endpoint_exts.get(received.recipient).unwrap().info.polarity

                    );

                    // message for some actor. Feed it to the appropriate actor

    }

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

        self.new_local.iter()

    }

    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

        })

    }

    pub(crate) fn submit_and_digest_subtree_solution(

        &mut self,

        logger: &mut String,

        subtree_id: SubtreeId,

        predicate: Predicate,

    ) {

        log!(logger, "NEW COMPONENT SOLUTION {:?} {:?}", subtree_id, &predicate);

        let index = self.subtree_id_to_index[&subtree_id];

        let left = 0..index;

        let right = (index + 1)..self.subtree_solutions.len();

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

        let was_new = subtree_solutions[index].insert(predicate.clone());

        if was_new {

            let set_visitor = left.chain(right).map(|index| &subtree_solutions[index]);

            Self::elaborate_into_new_local_rec(

                logger,

                predicate,

                set_visitor,

                old_local,

                new_local,

            );

        }

    }

    fn elaborate_into_new_local_rec<'a, 'b>(

        logger: &mut String,

        partial: Predicate,

        mut set_visitor: impl Iterator<Item = &'b HashSet<Predicate>> + Clone,

        old_local: &'b HashSet<Predicate>,

        new_local: &'a mut HashSet<Predicate>,

    ) {

        if let Some(set) = set_visitor.next() {

            // incomplete solution. keep traversing

            for pred in set.iter() {

                if let Some(elaborated) = pred.union_with(&partial) {

                    Self::elaborate_into_new_local_rec(

                        logger,

                        elaborated,

                        set_visitor.clone(),

                        old_local,

                        new_local,

                    )

                }

            }

        } else {

            // recursive stop condition. `partial` is a local subtree solution

            if !old_local.contains(&partial) {

                // ... and it hasn't been found before

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

                new_local.insert(partial);

            }

        }

    }

}

impl PolyContext for BranchPContext<'_, '_> {

    type D = ProtocolD;

    fn is_firing(&mut self, ekey: Key) -> Option<bool> {

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

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

        let val = self.predicate.query(channel_id);

        log!(

            &mut self.m_ctx.inner.logger,

            "!! PolyContext callback to is_firing by {:?}! returning {:?}",

            self.m_ctx.my_subtree_id,

            val,

        );

        val

    }

    fn read_msg(&mut self, ekey: Key) -> Option<&Payload> {

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

        let val = self.inbox.get(&ekey);

        log!(

            &mut self.m_ctx.inner.logger,

            "!! PolyContext callback to read_msg by {:?}! returning {:?}",

            self.m_ctx.my_subtree_id,

            val,

        );

        val

    }

}

pub mod ffi;

mod actors;

pub(crate) mod communication;

pub(crate) mod connector;

pub(crate) mod endpoint;

pub mod errors;

// pub mod experimental;

mod serde;

pub(crate) mod setup;

pub(crate) type ProtocolD = crate::protocol::ProtocolDescriptionImpl;

pub(crate) type ProtocolS = crate::protocol::ComponentStateImpl;

use crate::common::*;

use actors::*;

use endpoint::*;

use errors::*;

#[derive(Debug, PartialEq)]

pub(crate) enum CommonSatResult {

    FormerNotLatter,

    LatterNotFormer,

    Equivalent,

    New(Predicate),

    Nonexistant,

}

#[derive(Clone, Eq, PartialEq, Hash)]

pub(crate) struct Predicate {

    pub assigned: BTreeMap<ChannelId, bool>,

}

#[derive(Debug, Default)]

struct SyncBatch {

    puts: HashMap<Key, Payload>,

    gets: HashSet<Key>,

}

#[derive(Debug)]

pub enum Connector {

    Unconfigured(Unconfigured),

    Configured(Configured),

    Connected(Connected), // TODO consider boxing. currently takes up a lot of stack real estate

}

#[derive(Debug)]

pub struct Unconfigured {

    pub controller_id: ControllerId,

}

#[derive(Debug)]

pub struct Configured {

    controller_id: ControllerId,

    polarities: Vec<Polarity>,

    bindings: HashMap<usize, PortBinding>,

    protocol_description: Arc<ProtocolD>,

    main_component: Vec<u8>,

}

#[derive(Debug)]

pub struct Connected {

    native_interface: Vec<(Key, Polarity)>,

    sync_batches: Vec<SyncBatch>,

    controller: Controller,

}

#[derive(Debug, Copy, Clone)]

pub enum PortBinding {

    Native,

    Active(SocketAddr),

    Passive(SocketAddr),

}

#[derive(Debug)]

struct Arena<T> {

    storage: Vec<T>,

}

#[derive(Debug)]

struct ReceivedMsg {

    recipient: Key,

    msg: Msg,

}

#[derive(Debug)]

struct MessengerState {

    poll: Poll,

    events: Events,

    delayed: Vec<ReceivedMsg>,

    undelayed: Vec<ReceivedMsg>,

    polled_undrained: IndexSet<Key>,

}

#[derive(Debug)]

struct ChannelIdStream {

    controller_id: ControllerId,

    next_channel_index: ChannelIndex,

}

#[derive(Debug)]

struct Controller {

    protocol_description: Arc<ProtocolD>,

    inner: ControllerInner,

    ephemeral: ControllerEphemeral,

}

#[derive(Debug)]

struct ControllerInner {

    round_index: usize,

    channel_id_stream: ChannelIdStream,

    endpoint_exts: Arena<EndpointExt>,

    messenger_state: MessengerState,

    mono_n: MonoN,       // state at next round start

    mono_ps: Vec<MonoP>, // state at next round start

    family: ControllerFamily,

    logger: String,

}

/// This structure has its state entirely reset between synchronous rounds

#[derive(Debug, Default)]

struct ControllerEphemeral {

    solution_storage: SolutionStorage,

    poly_n: Option<PolyN>,

    poly_ps: Vec<PolyP>,

    mono_ps: Vec<MonoP>,

    ekey_to_holder: HashMap<Key, PolyId>,

}

#[derive(Debug)]

struct ControllerFamily {

    parent_ekey: Option<Key>,

    children_ekeys: Vec<Key>,

}

#[derive(Debug)]

pub(crate) enum SyncRunResult {

    BlockingForRecv,

    AllBranchesComplete,

    NoBranches,

}

// Used to identify poly actors

#[derive(Debug, Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]

enum PolyId {

    N,

    P { index: usize },

}

#[derive(Debug, Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]

pub(crate) enum SubtreeId {

    PolyN,

    PolyP { index: usize },

    ChildController { ekey: Key },

}

pub(crate) struct MonoPContext<'a> {

    inner: &'a mut ControllerInner,

    ekeys: &'a mut HashSet<Key>,

    mono_ps: &'a mut Vec<MonoP>,

}

pub(crate) struct PolyPContext<'a> {

    my_subtree_id: SubtreeId,

    inner: &'a mut ControllerInner,

    solution_storage: &'a mut SolutionStorage,

}

impl PolyPContext<'_> {

    #[inline(always)]

    fn reborrow<'a>(&'a mut self) -> PolyPContext<'a> {

        let Self { solution_storage, my_subtree_id, inner } = self;

        PolyPContext { solution_storage, my_subtree_id: *my_subtree_id, inner }

    }

}

struct BranchPContext<'m, 'r> {

    m_ctx: PolyPContext<'m>,

    ekeys: &'r HashSet<Key>,

    predicate: &'r Predicate,

    inbox: &'r HashMap<Key, Payload>,

}

#[derive(Default)]

pub(crate) struct SolutionStorage {

    old_local: HashSet<Predicate>,

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

}

trait Messengerlike {

    fn get_state_mut(&mut self) -> &mut MessengerState;

    fn get_endpoint_mut(&mut self, eekey: Key) -> &mut Endpoint;

    fn delay(&mut self, received: ReceivedMsg) {

        self.get_state_mut().delayed.push(received);

    }

    fn undelay_all(&mut self) {

        let MessengerState { delayed, undelayed, .. } = self.get_state_mut();

        undelayed.extend(delayed.drain(..))

    }

    fn send(&mut self, to: Key, msg: Msg) -> Result<(), EndpointErr> {

        let mut ekeys_network = vec![];

        let mut native_interface = vec![];

        /*

        1.  - allocate an EndpointExtTodo for every native and interface port

            - store all the resulting keys in two keylists for the interfaces of the native and proto components

                native: [a, c,    f]

                         |  |     |

                         |  |     |

                proto:  [b, d, e, g]

                               ^todo

                arena: <A,B,C,D,E,F,G>

        */

        for &(binding, polarity) in bound_proto_interface.iter() {

            match binding {

                PortBinding::Native => {

                    let channel_id = channel_id_stream.next();

                    let ([ekey_native, ekey_proto], native_polarity) = {

                        let [p, g] = Endpoint::new_memory_pair();

                        let mut endpoint_to_key = |endpoint, polarity| {

                            endpoint_ext_todos.alloc(EndpointExtTodo::Finished(EndpointExt {

                                endpoint,

                                info: EndpointInfo { polarity, channel_id },

                            }))

                        };

                        let pkey = endpoint_to_key(p, Putter);

                        let gkey = endpoint_to_key(g, Getter);

                        let key_pair = match polarity {

                            Putter => [gkey, pkey],

                            Getter => [pkey, gkey],

                        };

                        (key_pair, !polarity)

                    };

                    native_interface.push((ekey_native, native_polarity));

                    ekeys_native.push(ekey_native);

                    ekeys_proto.push(ekey_proto);

                }

                PortBinding::Passive(addr) => {

                    let channel_id = channel_id_stream.next();

                    let ekey_proto = endpoint_ext_todos.alloc(EndpointExtTodo::PassiveAccepting {

                        addr,

                        info: EndpointInfo { polarity, channel_id },

                        listener: TcpListener::bind(&addr).map_err(|_| BindFailed(addr))?,

                    });

                    ekeys_network.push(ekey_proto);

                    ekeys_proto.push(ekey_proto);

                }

                PortBinding::Active(addr) => {

                    let ekey_proto = endpoint_ext_todos.alloc(EndpointExtTodo::ActiveConnecting {

                        addr,

                        polarity,

                        stream: TcpStream::connect(&addr).unwrap(),

                    });

                    ekeys_network.push(ekey_proto);

                    ekeys_proto.push(ekey_proto);

                }

            }

        }

        log!(&mut logger, "{:03?} setup todos...", major);

        // 2. convert the arena to Arena<EndpointExt>  and return the

        let (mut messenger_state, mut endpoint_exts) =

            Self::finish_endpoint_ext_todos(major, &mut logger, endpoint_ext_todos, deadline)?;

        let n_mono = MonoN { ekeys: ekeys_native.into_iter().collect(), result: None };

        let p_monos = vec![MonoP {

            state: protocol_description.new_main_component(main_component, &ekeys_proto),

            ekeys: ekeys_proto.into_iter().collect(),

        }];

        // 6. Become a node in a sink tree, computing {PARENT, CHILDREN} from {NEIGHBORS}

        let family = Self::setup_sink_tree_family(

            major,

            &mut logger,

            &mut endpoint_exts,

            &mut messenger_state,

            ekeys_network,

            deadline,

        )?;

        log!(&mut logger, "CONNECT PHASE END! ~");

        let inner = ControllerInner {

            family,

            messenger_state,

            channel_id_stream,

            endpoint_exts,

            mono_ps: p_monos,

            mono_n: n_mono,

            round_index: 0,

            logger,

        };

        let controller = Self {

            protocol_description,

            inner,

            ephemeral: Default::default(),

        };

        Ok((controller, native_interface))

    }

    fn test_stream_connectivity(stream: &mut TcpStream) -> bool {

        use std::io::Write;

        stream.write(&[]).is_ok()

    }

    // inserts

    fn finish_endpoint_ext_todos(

        major: ControllerId,

        logger: &mut String,

        mut endpoint_ext_todos: Arena<EndpointExtTodo>,

        deadline: Instant,

    ) -> Result<(MessengerState, Arena<EndpointExt>), ConnectErr> {

        use {ConnectErr::*, EndpointExtTodo::*};

        // 1. define and setup a poller and event loop

        let edge = PollOpt::edge();

        let [ready_r, ready_w] = [Ready::readable(), Ready::writable()];

        let mut ms = MessengerState {

            poll: Poll::new().map_err(|_| PollInitFailed)?,

            events: Events::with_capacity(endpoint_ext_todos.len()),

            delayed: vec![],

            undelayed: vec![],

            polled_undrained: Default::default(),

        };

        // 2. Register all EndpointExtTodos with ms.poll. each has one of {Endpoint, TcpStream, TcpListener}

        // 3. store the keyset of EndpointExtTodos which are not Finished in `to_finish`.

        let mut to_finish = HashSet::<_>::default();

        log!(logger, "endpoint_ext_todos len {:?}", endpoint_ext_todos.len());

        for (key, t) in endpoint_ext_todos.iter() {

            let token = key.to_token();

            match t {

                ActiveRecving { .. } | PassiveConnecting { .. } => unreachable!(),

                Finished(EndpointExt { endpoint, .. }) => {

                    ms.poll.register(endpoint, token, ready_r, edge)

                }

                ActiveConnecting { stream, .. } => {

                    to_finish.insert(key);

                    ms.poll.register(stream, token, ready_w, edge)

                }

                PassiveAccepting { listener, .. } => {

                    to_finish.insert(key);

                    ms.poll.register(listener, token, ready_r, edge)

                }

            }

            .expect("register first");

        }

        // invariant: every EndpointExtTodo has one thing registered with mio

        // 4. until all in endpoint_ext_todos are Finished variant, handle events

        let mut polled_undrained_later = IndexSet::<_>::default();

        let mut backoff_millis = 10;

        while !to_finish.is_empty() {

            ms.poll_events(deadline)?;

            for event in ms.events.iter() {

                let token = event.token();

                let ekey = Key::from_token(token);

                let entry = endpoint_ext_todos.get_mut(ekey).unwrap();

                match entry {

                    Finished(_) => {

                        polled_undrained_later.insert(ekey);

                    }

                    PassiveAccepting { addr, listener, .. } => {

                        log!(logger, "{:03?} start PassiveAccepting...", major);

                        assert!(event.readiness().is_readable());

                        let (stream, _peer_addr) =

                            listener.accept().map_err(|_| AcceptFailed(*addr))?;

                        ms.poll.deregister(listener).expect("wer");

                        ms.poll.register(&stream, token, ready_w, edge).expect("3y5");

                        take_mut::take(entry, |e| {

                            assert_let![PassiveAccepting { addr, info, .. } = e => {

                                PassiveConnecting { addr, info, stream }

                            }]

                        });

                        log!(logger, "{:03?} ... end PassiveAccepting", major);

                    }

                    PassiveConnecting { addr, stream, .. } => {

                        log!(logger, "{:03?} start PassiveConnecting...", major);

                        assert!(event.readiness().is_writable());

                        if !Self::test_stream_connectivity(stream) {

                            return Err(PassiveConnectFailed(*addr));

                        }

                        ms.poll.reregister(stream, token, ready_r, edge).expect("52");

                        let mut res = Ok(());

                        take_mut::take(entry, |e| {

                            assert_let![PassiveConnecting { info, stream, .. } = e => {

                                let mut endpoint = Endpoint::from_fresh_stream(stream);

                                let msg = Msg::SetupMsg(SetupMsg::ChannelSetup { info });

                                res = endpoint.send(msg);

                                Finished(EndpointExt { info, endpoint })

                            }]

                        });