[package]

name = "reowolf_rs"

version = "0.1.4"

authors = [

	"Christopher Esterhuyse <esterhuy@cwi.nl, christopher.esterhuyse@gmail.com>",

	"Hans-Dieter Hiep <hdh@cwi.nl>"

]

edition = "2018"

[dependencies]

# convenience macros

maplit = "1.0.2"

derive_more = "0.99.2"

# runtime

bincode = "1.3.1"

serde = { version = "1.0.114", features = ["derive"] }

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

# network

mio = { version = "0.7.0", package = "mio", features = ["udp", "tcp", "os-poll"] }

socket2 = { version = "0.3.12", optional = true }

# protocol

backtrace = "0.3"

lazy_static = "1.4.0"

# ffi

# socket ffi

libc = { version = "^0.2", optional = true }

os_socketaddr = { version = "0.1.0", optional = true }

[dev-dependencies]

# test-generator = "0.3.0"

crossbeam-utils = "0.7.2"

lazy_static = "1.4.0"

[lib]

crate-type = [

	"rlib", # for use as a Rust dependency. 

	"cdylib" # for FFI use, typically C.

]

[features]

ffi = [] # see src/ffi/mod.rs

ffi_pseudo_socket_api = ["ffi", "libc", "os_socketaddr"]# see src/ffi/pseudo_socket_api.rs.

endpoint_logging = [] # see src/macros.rs

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

use crate::common::*;

use crate::runtime::*;

impl TokenTarget {

    // subdivides the domain of usize into

    // [NET_ENDPOINT][UDP_ENDPOINT  ]

    // ^0            ^usize::MAX/2   ^usize::MAX

    const HALFWAY_INDEX: usize = usize::MAX / 2;

    const MAX_INDEX: usize = usize::MAX;

}

impl From<Token> for TokenTarget {

    fn from(Token(index): Token) -> Self {

        if let Some(shifted) = index.checked_sub(Self::HALFWAY_INDEX) {

            TokenTarget::UdpEndpoint { index: shifted }

        } else {

            TokenTarget::NetEndpoint { index }

        }

    }

}

impl Into<Token> for TokenTarget {

    fn into(self) -> Token {

        match self {

            TokenTarget::UdpEndpoint { index } => Token(index + Self::HALFWAY_INDEX),

            TokenTarget::NetEndpoint { index } => Token(index),

        }

    }

}

impl Connector {

    /// Create a new connector structure with the given protocol description (via Arc to facilitate sharing).

    /// The resulting connector will start in the setup phase, and cannot be used for communication until the

    /// `connect` procedure completes.

    /// # Safety

    /// The correctness of the system's underlying distributed algorithms requires that no two

    /// connectors have the same ID. If the user does not know the identifiers of other connectors in the

    /// system, it is advised to guess it using Connector::random_id (relying on the exceptionally low probability of an error).

    /// Sessions with duplicate connector identifiers will not result in any memory unsafety, but cannot be guaranteed

    /// to preserve their configured protocols.

    /// Fortunately, in most realistic cases, the presence of duplicate connector identifiers will result in an

    /// error during `connect`, observed as a peer misbehaving.

    pub fn new(

        mut logger: Box<dyn Logger>,

        proto_description: Arc<ProtocolDescription>,

        connector_id: ConnectorId,

    ) -> Self {

        log!(&mut *logger, "Created with connector_id {:?}", connector_id);

        let mut id_manager = IdManager::new(connector_id);

        let native_component_id = id_manager.new_component_id();

        Self {

            unphased: ConnectorUnphased {

                proto_description,

                proto_components: Default::default(),

                logger,

                native_component_id,

                ips: IdAndPortState { id_manager, port_info: Default::default() },

            },

            phased: ConnectorPhased::Setup(Box::new(ConnectorSetup {

                net_endpoint_setups: Default::default(),

                udp_endpoint_setups: Default::default(),

            })),

        }

    }

    /// Conceptually, this returning [p0, g1] is sugar for:

    /// 1. create port pair [p0, g0]

    /// 2. create port pair [p1, g1]

    /// 3. create udp component with interface of moved ports [p1, g0]

    /// 4. return [p0, g1]

    pub fn new_udp_mediator_component(

        &mut self,

        local_addr: SocketAddr,

        peer_addr: SocketAddr,

    ) -> Result<[PortId; 2], WrongStateError> {

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

        match phased {

            ConnectorPhased::Communication(..) => Err(WrongStateError),

            ConnectorPhased::Setup(setup) => {

                let udp_index = setup.udp_endpoint_setups.len();

                let udp_cid = cu.ips.id_manager.new_component_id();

                // allocates 4 new port identifiers, two for each logical channel,

                // one channel per direction (into and out of the component)

                let mut npid = || cu.ips.id_manager.new_port_id();

                let [nin, nout, uin, uout] = [npid(), npid(), npid(), npid()];

                // allocate the native->udp_mediator channel's ports

                cu.ips.port_info.map.insert(

                    nout,

                    PortInfo {

                        route: Route::LocalComponent,

                        polarity: Putter,

                        peer: Some(uin),

                        owner: cu.native_component_id,

                    },

                );

                cu.ips.port_info.map.insert(

                    uin,

                    PortInfo {

                        route: Route::UdpEndpoint { index: udp_index },

                        polarity: Getter,

                        peer: Some(uin),

                        owner: udp_cid,

                    },

                );

                // allocate the udp_mediator->native channel's ports

                cu.ips.port_info.map.insert(

                    uout,

                    PortInfo {

                        route: Route::UdpEndpoint { index: udp_index },

                        polarity: Putter,

                        peer: Some(uin),

                        owner: udp_cid,

                    },

                );

                cu.ips.port_info.map.insert(

                    nin,

                    PortInfo {

                        route: Route::LocalComponent,

                        polarity: Getter,

                        peer: Some(uout),

                        owner: cu.native_component_id,

                    },

                );

                // allocate the two ports owned by the UdpMediator component

                // Remember to setup this UdpEndpoint setup during `connect` later.

                setup.udp_endpoint_setups.push(UdpEndpointSetup {

                    local_addr,

                    peer_addr,

                    getter_for_incoming: nin,

                });

                // update owned sets

                cu.ips

                    .port_info

                    .owned

                    .entry(cu.native_component_id)

                    .or_default()

                    .extend([nin, nout].iter().copied());

                cu.ips.port_info.owned.insert(udp_cid, maplit::hashset! {uin, uout});

                // Return the native's output, input port pair

                Ok([nout, nin])

            }

        }

    }

    /// Adds a "dangling" port to the connector in the setup phase,

    /// to be formed into channel during the connect procedure with the given

    /// transport layer information.

    pub fn new_net_port(

        &mut self,

        polarity: Polarity,

        sock_addr: SocketAddr,

        endpoint_polarity: EndpointPolarity,

    ) -> Result<PortId, WrongStateError> {

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

        match phased {

            ConnectorPhased::Communication(..) => Err(WrongStateError),

            ConnectorPhased::Setup(setup) => {

                // allocate a single dangling port with a `None` peer (for now)

                let new_pid = cu.ips.id_manager.new_port_id();

                cu.ips.port_info.map.insert(

                    new_pid,

                    PortInfo {

                        route: Route::LocalComponent,

                        peer: None,

                        owner: cu.native_component_id,

                        polarity,

                    },

                );

                log!(

                    cu.logger,

                    "Added net port {:?} with polarity {:?} addr {:?} endpoint_polarity {:?}",

                    new_pid,

                    polarity,

                    &sock_addr,

                    endpoint_polarity

                );

                // Remember to setup this NetEndpoint setup during `connect` later.

                setup.net_endpoint_setups.push(NetEndpointSetup {

                    sock_addr,

                    endpoint_polarity,

                    getter_for_incoming: new_pid,

                });

                // update owned set

                cu.ips.port_info.owned.entry(cu.native_component_id).or_default().insert(new_pid);

                Ok(new_pid)

            }

        }

    }

    /// Finalizes the connector's setup procedure and forms a distributed system with

    /// all other connectors reachable through network channels. This procedure represents

    /// a synchronization barrier, and upon successful return, the connector can no longer add new network ports,

    /// but is ready to begin the first communication round.

    /// Initially, the connector has a singleton set of _batches_, the only element of which is empty.

    /// This single element starts off selected. The selected batch is modified with `put` and `get`,

    /// and new batches are added and selected with `next_batch`. See `sync` for an explanation of the

    /// purpose of these batches.

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

        use ConnectError as Ce;

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

        match &phased {

            ConnectorPhased::Communication { .. } => {

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

                Err(Ce::AlreadyConnected)

            }

            ConnectorPhased::Setup(setup) => {

        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

                &mut *cu.logger,

                &setup.net_endpoint_setups,

                &setup.udp_endpoint_setups,

                &deadline,

            )?;

            log!(

                cu.logger,

                "Successfully connected {} endpoints. info now {:#?} {:#?}",

                endpoint_manager.net_endpoint_store.endpoint_exts.len(),

                &cu.ips.port_info,

                &endpoint_manager,

            );

            // leader election and tree construction. Learn our role in the consensus tree,

            // from learning who are our children/parents (neighbors) in the consensus tree.

            let neighborhood = init_neighborhood(

                cu.ips.id_manager.connector_id,

                &mut *cu.logger,

                &mut endpoint_manager,

                &deadline,

            )?;

            log!(cu.logger, "Successfully created neighborhood {:?}", &neighborhood);

            // Put it all together with an initial round index of zero.

            let mut comm = ConnectorCommunication {

                round_index: 0,

                endpoint_manager,

                neighborhood,

                native_batches: vec![Default::default()],

                round_result: Ok(None), // no previous round yet

            };

            if cfg!(feature = "session_optimization") {

                // Perform the session optimization procedure, which may modify the

                // internals of the connector, rerouting ports, moving around connectors etc.

            }

            log!(cu.logger, "connect() finished. setup phase complete");

            }

        }

    }

}

// Given a set of net_ and udp_ endpoints to setup,

// port information to flesh out (by discovering peers through channels)

// and a deadline in which to do it,

// try to return:

// - An EndpointManager, containing all the set up endpoints

// - new information about ports acquired through the newly-created channels

fn setup_endpoints_and_pair_ports(

    logger: &mut dyn Logger,

    net_endpoint_setups: &[NetEndpointSetup],

    udp_endpoint_setups: &[UdpEndpointSetup],

    deadline: &Option<Instant>,

    use ConnectError as Ce;

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

    const RETRY_PERIOD: Duration = Duration::from_millis(200);

    // The data for a net endpoint's setup in progress

    struct NetTodo {

        // becomes completed once sent_local_port && recv_peer_port.is_some()

        // we send local port if we haven't already and we receive a writable event

        // we recv peer port if we haven't already and we receive a readbale event

        todo_endpoint: NetTodoEndpoint,

        endpoint_setup: NetEndpointSetup,

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

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

    }

    // The data for a udp endpoint's setup in progress

    struct UdpTodo {

        // becomes completed once we receive our first writable event

        getter_for_incoming: PortId,

        sock: UdpSocket,

    }

    // Substructure of `NetTodo`, which represents the endpoint itself

    enum NetTodoEndpoint {

        Accepting(TcpListener),       // awaiting it's peer initiating the connection

        PeerInfoRecving(NetEndpoint), // awaiting info about peer port through the channel

    }

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

    // Start to construct our return values

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

    let mut events =

        Events::with_capacity((net_endpoint_setups.len() + udp_endpoint_setups.len()) * 2 + 4);

    let [mut net_polled_undrained, udp_polled_undrained] = [VecSet::default(), VecSet::default()];

    let mut delayed_messages = vec![];

    let mut last_retry_at = Instant::now();

    let mut io_byte_buffer = IoByteBuffer::default();

    // Create net/udp todo structures, each already registered with poll

    let mut net_todos = net_endpoint_setups

        .iter()

        .enumerate()

        .map(|(index, endpoint_setup)| {

            let token = TokenTarget::NetEndpoint { index }.into();

            log!(logger, "Net endpoint {} beginning setup with {:?}", index, &endpoint_setup);

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

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

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

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

                NetTodoEndpoint::PeerInfoRecving(NetEndpoint { stream, inbox: vec![] })

            } else {

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

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

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

                NetTodoEndpoint::Accepting(listener)

            };

            Ok(NetTodo {

                todo_endpoint,

                sent_local_port: false,

                recv_peer_port: None,

                endpoint_setup: endpoint_setup.clone(),

            })

        })

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

    let udp_todos = udp_endpoint_setups

        .iter()

        .enumerate()

        .map(|(index, endpoint_setup)| {

            let mut sock = UdpSocket::bind(endpoint_setup.local_addr)

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

            sock.connect(endpoint_setup.peer_addr)

                .map_err(|_| Ce::UdpConnectFailed(endpoint_setup.peer_addr))?;

            poll.registry()

                .register(&mut sock, TokenTarget::UdpEndpoint { index }.into(), Interest::WRITABLE)

                .unwrap();

            Ok(UdpTodo { sock, getter_for_incoming: endpoint_setup.getter_for_incoming })

        })

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

    // Initially no net connections have failed, and all udp and net endpoint setups are incomplete

    let mut net_connect_to_retry: HashSet<usize> = Default::default();

    let mut setup_incomplete: HashSet<TokenTarget> = {

        let net_todo_targets_iter =

            (0..net_todos.len()).map(|index| TokenTarget::NetEndpoint { index });

        let udp_todo_targets_iter =

            (0..udp_todos.len()).map(|index| TokenTarget::UdpEndpoint { index });

        net_todo_targets_iter.chain(udp_todo_targets_iter).collect()

    };

    // progress by reacting to poll events. continue until every endpoint is set up