pub enum EvalErr {

    ComponentExitWhileBranching,

}

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

pub enum MessengerRecvErr {

    PollingFailed,

}

impl From<MainComponentErr> for ConfigErr {

    fn from(e: MainComponentErr) -> Self {

        use ConfigErr as C;

        use MainComponentErr as M;

        match e {

use super::vec_storage::VecStorage;

use crate::common::*;

use crate::runtime::endpoint::EndpointExt;

use crate::runtime::endpoint::EndpointInfo;

use crate::runtime::endpoint::{Endpoint, Msg, SetupMsg};

use crate::runtime::MessengerState;

use crate::runtime::Messengerlike;

use crate::runtime::ReceivedMsg;

use std::net::SocketAddr;

use std::sync::Arc;

pub enum Coupling {

    Active,

    Passive,

}

struct Family {

    parent: Option<Port>,

    children: HashSet<Port>,

}

pub struct Binding {

impl From<OutPort> for Port {

    fn from(x: OutPort) -> Self {

        x.0

    }

}

struct ChannelIndexStream {

    next: u32,

}

impl ChannelIndexStream {

    fn next(&mut self) -> u32 {

        self.next += 1;

impl Binds<OutPort> for Connecting {

    fn bind(&mut self, coupling: Coupling, addr: SocketAddr) -> OutPort {

        self.bindings.push(Binding { coupling, polarity: Polarity::Putter, addr });

        OutPort(Port(self.bindings.len() - 1))

    }

}

impl Connecting {

    fn random_controller_id() -> ControllerId {

        type Bytes8 = [u8; std::mem::size_of::<ControllerId>()];

        let mut bytes = Bytes8::default();

        getrandom::getrandom(&mut bytes).unwrap();

        unsafe {

    }

    fn new_connected(

        &self,

        controller_id: ControllerId,

        protocol: &Arc<Protocol>,

        timeout: Option<Duration>,

        // 1. bindings correspond with ports 0..bindings.len(). For each:

        //    - reserve a slot in endpoint_exts.

        //    - store the port in `native_ports' set.

        let mut endpoint_exts = VecStorage::<EndpointExt>::with_reserved_range(self.bindings.len());

        let native_ports = (0..self.bindings.len()).map(Port).collect();

        // 2. create MessengerState structure for polling channels

        let edge = PollOpt::edge();

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

        // 3. create one TODO task per (port,binding) as a vector with indices in lockstep.

        //    we will drain it gradually so we store elements of type Option<Todo> where all are initially Some(_)

        enum Todo {

            PassiveAccepting { listener: TcpListener, channel_id: ChannelId },

            ActiveConnecting { stream: TcpStream },

            .enumerate()

            .map(|(index, binding)| {

                Ok(Some(match binding.coupling {

                    Coupling::Passive => {

                        let channel_index = channel_index_stream.next();

                        let channel_id = ChannelId { controller_id, channel_index };

                        ms.poll.register(&listener, Token(index), ready_r, edge).unwrap(); // registration unique

                        Todo::PassiveAccepting { listener, channel_id }

                    }

                    Coupling::Active => {

                        ms.poll.register(&stream, Token(index), ready_w, edge).unwrap(); // registration unique

                        Todo::ActiveConnecting { stream }

                    }

                }))

            })

        let mut num_todos_remaining = todos.len();

        // 4. handle incoming events until all TODOs are completed OR we timeout

        let deadline = timeout.map(|t| Instant::now() + t);

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

        let mut backoff_millis = 10;

        while num_todos_remaining > 0 {

            for event in ms.events.iter() {

                let token = event.token();

                let index = token.0;

                let binding = &self.bindings[index];

                match todos[index].take() {

                    None => {

                        polled_undrained_later.insert(index);

                    }

                    Some(Todo::PassiveAccepting { listener, channel_id }) => {

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

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

                        todos[index] = Some(Todo::PassiveConnecting { stream, channel_id });

                    }

                    Some(Todo::ActiveConnecting { mut stream }) => {

                        let todo = if Self::test_stream_connectivity(&mut stream) {

                            Todo::ActiveConnecting { stream }

                        };

                        todos[index] = Some(todo);

                    }

                    Some(Todo::PassiveConnecting { mut stream, channel_id }) => {

                        if !Self::test_stream_connectivity(&mut stream) {

                        }

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

                        let polarity = binding.polarity;

                        let info = EndpointInfo { polarity, channel_id };

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

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

                        let endpoint_ext = EndpointExt { endpoint, info };

                        endpoint_exts.occupy_reserved(index, endpoint_ext);

                        num_todos_remaining -= 1;

                    }

                    Some(Todo::ActiveRecving { mut endpoint }) => {

                        let ekey = Key::from_raw(index);

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

                                if info.polarity == binding.polarity {

                                }

                                let channel_id = info.channel_id;

                                let info = EndpointInfo { polarity: binding.polarity, channel_id };

                                ms.polled_undrained.insert(ekey);

                                let endpoint_ext = EndpointExt { endpoint, info };

                                endpoint_exts.occupy_reserved(index, endpoint_ext);

            }

        }

        assert_eq!(None, endpoint_exts.iter_reserved().next());

        drop(todos);

        // 1. construct `family', i.e. perform the sink tree setup procedure

        use {Msg::SetupMsg as S, SetupMsg::*};

        let mut messenger = (&mut ms, &mut endpoint_exts);

        impl Messengerlike for (&mut MessengerState, &mut VecStorage<EndpointExt>) {

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

                self.0

            }

        // 1. broadcast my ID as the first echo. await reply from all in net_keylist

        let neighbors = (0..self.bindings.len()).map(Port);

        let echo = S(LeaderEcho { maybe_leader: controller_id });

        let mut awaiting = IndexSet::<Port>::with_capacity(neighbors.len());

        for n in neighbors.clone() {

            awaiting.insert(n);

        }

        // 2. Receive incoming replies. whenever a higher-id echo arrives,

        //    adopt it as leader, sender as parent, and reset the await set.

        let mut parent: Option<Port> = None;

        let mut my_leader = controller_id;

        messenger.undelay_all();

        'echo_loop: while !awaiting.is_empty() || parent.is_some() {

            match msg {

                S(LeaderAnnounce { leader }) => {

                    // someone else completed the echo and became leader first!

                    // the sender is my parent

                    parent = Some(recipient);

                    my_leader = leader;

                        Equal => {

                            awaiting.remove(&recipient);

                            if awaiting.is_empty() {

                                if let Some(p) = parent {

                                    // return the echo to my parent

                                    messenger

                                } else {

                                    // DECIDE!

                                    break 'echo_loop;

                                }

                            }

                        }

                            parent = Some(recipient);

                            my_leader = maybe_leader;

                            let echo = S(LeaderEcho { maybe_leader: my_leader });

                            awaiting.clear();

                            if neighbors.len() == 1 {

                                // immediately reply to parent

                            } else {

                                for n in neighbors.clone() {

                                    if n != recipient {

                                        awaiting.insert(n);

                                    }

                                }

                            }

                        }

                    }

                }

            }

        }

        match parent {

            None => assert_eq!(

                my_leader, controller_id,

                "I've got no parent, but I consider {:?} the leader?",

        // 3. broadcast leader announcement (except to parent: confirm they are your parent)

        //    in this loop, every node sends 1 message to each neighbor

        let msg_for_non_parents = S(LeaderAnnounce { leader: my_leader });

        for n in neighbors.clone() {

            let msg =

                if Some(n) == parent { S(YouAreMyParent) } else { msg_for_non_parents.clone() };

        }

        // await 1 message from all non-parents

        for n in neighbors.clone() {

            if Some(n) != parent {

                awaiting.insert(n);

            }

        }

        let mut children = HashSet::default();

        messenger.undelay_all();

        while !awaiting.is_empty() {

            match msg {

                S(YouAreMyParent) => {

                    assert!(awaiting.remove(&recipient));

                    children.insert(recipient);

                }

                S(SetupMsg::LeaderAnnounce { leader }) => {

                    assert!(awaiting.remove(&recipient));

                    assert!(leader == my_leader);

                    assert!(Some(recipient) != parent);

                    // they wouldn't send me this if they considered me their parent

                }

            }

        }

        let family = Family { parent, children };

        // 1. done! return

        Ok(Connected {

    /////////

    pub fn connect_using_id(

        &mut self,

        controller_id: ControllerId,

        protocol: &Arc<Protocol>,

        timeout: Option<Duration>,

        // 1. try and create a connection from these bindings with self immutable.

        let connected = self.new_connected(controller_id, protocol, timeout)?;

        // 2. success! drain self and return

        self.bindings.clear();

        Ok(connected)

    }

    pub fn connect(

        &mut self,

        protocol: &Arc<Protocol>,

        timeout: Option<Duration>,

        self.connect_using_id(Self::random_controller_id(), protocol, timeout)

    }

}

pub struct Protocol;

impl Protocol {

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

        Ok(Protocol)

    }

}

pub struct Connected {

    native_ports: HashSet<Port>,

    controller_id: ControllerId,

    channel_index_stream: ChannelIndexStream,

    endpoint_exts: VecStorage<EndpointExt>,

    protocol: Arc<Protocol>,

    family: Family,

}

impl Connected {

    pub fn new_channel(&mut self) -> (OutPort, InPort) {

        assert!(self.endpoint_exts.len() <= std::u32::MAX as usize - 2);

        let channel_id = ChannelId {

            controller_id: self.controller_id,

unsafe fn as_mut_slice<'a, T>(len: usize, ptr: *mut T) -> &'a mut [T] {

    std::slice::from_raw_parts_mut(ptr, len)

}

unsafe fn as_const_slice<'a, T>(len: usize, ptr: *const T) -> &'a [T] {

    std::slice::from_raw_parts(ptr, len)

}

            return Ok(Some(x));

        }

        loop {

            // polled_undrained may not be empty

            while let Some(eekey) = self.get_state_mut().polled_undrained.pop() {

                    // this endpoint MAY still have messages! check again in future

                    self.get_state_mut().polled_undrained.insert(eekey);

                    return Ok(Some(ReceivedMsg { recipient: eekey, msg }));

                }

            }

            return Ok(Some(x));

        }

        loop {

            // polled_undrained may not be empty

            while let Some(eekey) = self.get_state_mut().polled_undrained.pop() {

                    // this endpoint MAY still have messages! check again in future

                    self.get_state_mut().polled_undrained.insert(eekey);

                    return Ok(Some(ReceivedMsg { recipient: eekey, msg }));

                }

            }

}

impl From<MessengerRecvErr> for ConnectErr {

    fn from(e: MessengerRecvErr) -> ConnectErr {

        ConnectErr::MessengerRecvErr(e)

    }

}

impl<T> Default for Arena<T> {

    fn default() -> Self {

        Self { storage: vec![] }

    }

}

impl<T> Arena<T> {

        self.next_channel_index += 1;

        ChannelId { controller_id: self.controller_id, channel_index: self.next_channel_index - 1 }

    }

}

impl MessengerState {

    // does NOT guarantee that events is non-empty

    fn poll_events(&mut self, deadline: Instant) -> Result<(), PollDeadlineErr> {

        use PollDeadlineErr::*;

        self.events.clear();

        let poll_timeout = deadline.checked_duration_since(Instant::now()).ok_or(Timeout)?;

        self.poll.poll(&mut self.events, Some(poll_timeout)).map_err(|_| PollingFailed)?;