# network

# integer-encoding = "1.1.5"

# byteorder = "1.3.4"

# protocol

backtrace = "0.3"

            },

        }

    }

        // returns index of new batch

        let Self { phased, .. } = self;

        match phased {

            ConnectorPhased::Communication(comm) => {

                comm.native_batches.push(Default::default());

                Ok(comm.native_batches.len() - 1)

                };

                match comm_msg_contents {

                    CommMsgContents::SendPayload(send_payload_msg) => {

                        assert!(cu.port_info.polarities.get(&getter) == Some(&Getter));

                        log!(

                            cu.logger,

    EndpointError { error: EndpointError, index: usize },

}

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

    fn bincode_opts() -> impl bincode::config::Options {

        bincode::config::DefaultOptions::default()

    }

impl EndpointManager {

    pub(super) fn index_iter(&self) -> Range<usize> {

    }

    }

    pub(super) fn send_to_comms(

        &mut self,

        msg: &Msg,

    ) -> Result<(), UnrecoverableSyncError> {

        use UnrecoverableSyncError as Use;

    }

    pub(super) fn send_to_setup(&mut self, index: usize, msg: &Msg) -> Result<(), ConnectError> {

        })

    }

    pub(super) fn try_recv_any_comms(

            Trae::Timeout => Ce::Timeout,

            Trae::PollFailed => Ce::PollFailed,

            Trae::EndpointError { error, index } => Ce::EndpointSetupError(

                error,

            ),

        })

        loop {

            // 2. try read a message from an endpoint that raised an event with poll() but wasn't drained

            while let Some(index) = self.polled_undrained.pop() {

                    .try_recv(logger)

                    .map_err(|error| Trea::EndpointError { error, index })?

                {

                    endptlog!(logger, "RECV polled_undrained {:?}", &msg);

                        // there may be another message waiting!

                        self.polled_undrained.insert(index);

                    }

        self.undelayed_messages.extend(self.delayed_messages.drain(..));

    }

}

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

        f.debug_struct("Endpoint").field("inbox", &self.inbox).finish()

    }

    PreviousSyncFailed,

}

#[derive(Debug, Eq, PartialEq)]

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

impl From<UnrecoverableSyncError> for SyncError {

    fn from(e: UnrecoverableSyncError) -> Self {

use crate::common::*;

use error::*;

#[derive(Debug)]

pub struct Connector {

    New(Predicate),

    Nonexistant,

}

    inbox: Vec<u8>,

    stream: TcpStream,

}

    ports: HashSet<PortId>,

}

#[derive(Debug, Clone)]

    sock_addr: SocketAddr,

    endpoint_polarity: EndpointPolarity,

}

#[derive(Debug)]

    getter_for_incoming: PortId,

}

#[derive(Debug)]

    polled_undrained: VecSet<usize>,

    delayed_messages: Vec<(usize, Msg)>,

    undelayed_messages: Vec<(usize, Msg)>,

}

#[derive(Clone, Debug, Default, serde::Serialize, serde::Deserialize)]

struct PortInfo {

    port_info: PortInfo,

}

#[derive(Debug)]

enum ConnectorPhased {

    Communication(Box<ConnectorCommunication>),

}

#[derive(Default, Clone, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]

    to_put: HashMap<PortId, Payload>,

    to_get: HashSet<PortId>,

}

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

    err.kind() == std::io::ErrorKind::WouldBlock

}

impl<T: std::cmp::Ord> VecSet<T> {

    fn new(mut vec: Vec<T>) -> Self {

        vec.sort();

                native_ports: Default::default(),

                port_info: Default::default(),

            },

        }

    }

    pub fn new_net_port(

        polarity: Polarity,

        sock_addr: SocketAddr,

        endpoint_polarity: EndpointPolarity,

        let Self { unphased: up, phased } = self;

        match phased {

                let p = up.id_manager.new_port_id();

                up.native_ports.insert(p);

                // {polarity, route} known. {peer} unknown.

                    polarity,

                    &endpoint_setup

                );

                Ok(p)

            }

        }

    pub fn connect(&mut self, timeout: Option<Duration>) -> Result<(), ConnectError> {

        use ConnectError as Ce;

        let Self { unphased: cu, phased } = self;

            ConnectorPhased::Communication { .. } => {

                log!(cu.logger, "Call to connecting in connected state");

                Err(Ce::AlreadyConnected)

            }

                log!(cu.logger, "~~~ CONNECT called timeout {:?}", timeout);

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

                // connect all endpoints in parallel; send and receive peer ids through ports

                let mut endpoint_manager = new_endpoint_manager(

                    &mut *cu.logger,

                    &mut cu.port_info,

                    deadline,

                )?;

                log!(

                    cu.logger,

                    "Successfully connected {} endpoints",

                );

                // leader election and tree construction

                let neighborhood = init_neighborhood(

}

fn new_endpoint_manager(

    logger: &mut dyn Logger,

    port_info: &mut PortInfo,

    deadline: Option<Instant>,

) -> Result<EndpointManager, ConnectError> {

    const BOTH: Interest = Interest::READABLE.add(Interest::WRITABLE);

    struct Todo {

        todo_endpoint: TodoEndpoint,

        local_port: PortId,

        sent_local_port: bool,          // true <-> I've sent my local port

        recv_peer_port: Option<PortId>, // Some(..) <-> I've received my peer's port

    }

    enum TodoEndpoint {

        Accepting(TcpListener),

    }

    fn init_todo(

        token: Token,

        local_port: PortId,

        poll: &mut Poll,

    ) -> Result<Todo, ConnectError> {

        let todo_endpoint = if let EndpointPolarity::Active = endpoint_setup.endpoint_polarity {

            let mut stream = TcpStream::connect(endpoint_setup.sock_addr)

                .expect("mio::TcpStream connect should not fail!");

            poll.registry().register(&mut stream, token, BOTH).unwrap();

        } else {

            let mut listener = TcpListener::bind(endpoint_setup.sock_addr)

                .map_err(|_| Ce::BindFailed(endpoint_setup.sock_addr))?;

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

    // 1. Start to construct EndpointManager

    const WAKER_PERIOD: Duration = Duration::from_millis(300);

    let mut poll = Poll::new().map_err(|_| Ce::PollInitFailed)?;

    let mut events = Events::with_capacity(endpoint_setups.len() * 2 + 4);

    let mut polled_undrained = VecSet::default();

        .iter()

        .enumerate()

        .map(|(index, (local_port, endpoint_setup))| {

        })

        .collect::<Result<Vec<Todo>, ConnectError>>()?;

        poll.poll(&mut events, remaining).map_err(|_| Ce::PollFailed)?;

        for event in events.iter() {

            let token = event.token();

                    log!(

                        logger,

                        "Notification from waker. connect_failed is {:?}",

                        connect_failed.iter()

                    );

                        log!(

                            logger,

                            "Restarting connection with endpoint {:?} {:?}",

                            todo.endpoint_setup.sock_addr

                        );

                        match &mut todo.todo_endpoint {

                                        .expect("mio::TcpStream connect should not fail!");

                                std::mem::swap(&mut endpoint.stream, &mut new_stream);

                                poll.registry()

                                    .unwrap();

                            }

                            _ => unreachable!(),

                        }

                    }

                    let todo: &mut Todo = &mut todos[index];

                    if let TodoEndpoint::Accepting(listener) = &mut todo.todo_endpoint {

                        match listener.accept() {

                            Ok((mut stream, peer_addr)) => {

                                poll.registry().deregister(listener).unwrap();

                                poll.registry().register(&mut stream, token, BOTH).unwrap();

                                log!(

                                    logger,

                                    index,

                                    peer_addr

                                );

                            }

                        }

                    }

                        if event.is_error() {

                            if todo.endpoint_setup.endpoint_polarity == EndpointPolarity::Passive {

                            }

                            if connect_failed.insert(index) {

                                log!(

                                    logger,

                                    index,

                                    connect_failed.iter()

                                );

                            } else {

                                continue;

                            }

                                log!(logger, "First connect failure. Starting waker thread");

                                std::thread::spawn(move || {

                                        std::thread::sleep(WAKER_PERIOD);

                                    }

                                });

                            }

                            continue;

                        }

                        if connect_failed.contains(&index) {

                            // spurious wakeup

                            continue;

                        }

                        let local_polarity = *port_info.polarities.get(&todo.local_port).unwrap();

                        if event.is_writable() && !todo.sent_local_port {

                            let msg = Msg::SetupMsg(SetupMsg::MyPortInfo(MyPortInfo {

                                polarity: local_polarity,

                                port: todo.local_port,

                            }));

                                .send(&msg)

                                .map_err(|e| {

                                })

                                .unwrap();

                            log!(logger, "endpoint[{}] sent msg {:?}", index, &msg);

                            todo.sent_local_port = true;

                        }

                        if event.is_readable() && todo.recv_peer_port.is_none() {

                            })?;

                                polled_undrained.insert(index);

                            }

                            match maybe_msg {

                                None => {} // msg deserialization incomplete

                                Some(Msg::SetupMsg(SetupMsg::MyPortInfo(peer_info))) => {

                                    if peer_info.polarity == local_polarity {

                                        return Err(ConnectError::PortPeerPolarityMismatch(

                                            todo.local_port,

                                }

                            }

                        }

                        if todo.sent_local_port && todo.recv_peer_port.is_some() {

                            setup_incomplete.remove(&index);

                            log!(logger, "endpoint[{}] is finished!", index);

                        }

                    }

                }

            }

        events.clear();

    }

        .into_iter()

        .enumerate()

                    poll.registry()

                        .unwrap();

                }

                _ => unreachable!(),

            },

            getter_for_incoming: local_port,

        })

        .collect();

        log!(logger, "Sending waker the stop signal");

    }

    Ok(EndpointManager {

        poll,

        polled_undrained,

        undelayed_messages: delayed_messages, // no longer delayed

        delayed_messages: Default::default(),

    })

}

    NOTE the distinction between PARENT and LEADER

    */

    log!(logger, "beginning neighborhood construction");

        log!(logger, "Edge case of no neighbors! No parent an no children!");

        return Ok(Neighborhood { parent: None, children: VecSet::new(vec![]) });

    }

    // 1+ neighbors. Leader can only be learned by receiving messages

    // loop ends when I know my sink tree parent (implies leader was elected)

    let election_result: WaveState = {

        serde_proto_description: SerdeProtocolDescription(cu.proto_description.clone()),

        endpoint_incoming_to_getter: comm

            .endpoint_manager

            .iter()

            .map(|ee| ee.getter_for_incoming)

            .collect(),

    cu.proto_components = proto_components;

    cu.proto_description = serde_proto_description.0;

    for (ee, getter) in

    {

        ee.getter_for_incoming = getter;

    }