Changeset - 0b5fa1cf6ccf
[Not reviewed]
0 1 0
Christopher Esterhuyse - 5 years ago 2020-10-08 12:50:21
christopher.esterhuyse@gmail.com
more detailed doc comments
1 file changed with 15 insertions and 5 deletions:
0 comments (0 inline, 0 general)
src/runtime/setup.rs
Show inline comments
 
@@ -109,223 +109,233 @@ impl Connector {
 
                        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) => {
 
                // Ideally, we'd simply clone `cu` in its entirety, and pass it to
 
                // `connect_inner`, such that a successful connection discards the original.
 
                // We need to work around the `logger` not being clonable;
 
                // solution? create a dummy clone, and use mem-swap to ensure the
 
                // single real logger is wherever it needs to be.
 
                // Idea: Clone `self.unphased`, and then pass the replica to
 
                // `connect_inner` to do the work, attempting to create a new connector structure
 
                // in connected state without encountering any errors.
 
                // If anything goes wrong during `connect_inner`, we simply keep the original `cu`.
 

	
 
                // Ideally, we'd simply clone `cu` in its entirety.
 
                // However, it isn't clonable, because of the pesky logger.
 
                // Solution: the original and clone ConnectorUnphased structures
 
                // 'share' the original logger by using `mem::swap` strategically to pass a dummy back and forth,
 
                // such that the real logger is wherever we need it to be without violating any invariants.
 
                let mut cu_clone = ConnectorUnphased {
 
                    logger: Box::new(DummyLogger),
 
                    proto_components: cu.proto_components.clone(),
 
                    native_component_id: cu.native_component_id.clone(),
 
                    ips: cu.ips.clone(),
 
                    proto_description: cu.proto_description.clone(),
 
                };
 
                // cu has REAL logger...
 
                std::mem::swap(&mut cu.logger, &mut cu_clone.logger);
 
                // ... cu_clone has REAL logger.
 
                match Self::connect_inner(cu_clone, setup, timeout) {
 
                    Ok(connected_connector) => {
 
                        *self = connected_connector;
 
                        Ok(())
 
                    }
 
                    Err((err, mut logger)) => {
 
                        // Put the original logger back in place (in self.unphased, AKA `cu`).
 
                        // cu_clone has REAL logger...
 
                        std::mem::swap(&mut cu.logger, &mut logger);
 
                        // ... cu has REAL logger.
 
                        Err(err)
 
                    }
 
                }
 
            }
 
        }
 
    }
 

	
 
    // Given an immutable setup structure, and my own (cloned) ConnetorUnphased,
 
    // attempt to complete the setup procedure and return a new connector in Connected state.
 
    // If anything goes wrong, throw everything in the bin, except for the Logger, which is
 
    // the only structure that sees lasting effects of the failed attempt.
 
    fn connect_inner(
 
        mut cu: ConnectorUnphased,
 
        setup: &ConnectorSetup,
 
        timeout: Option<Duration>,
 
    ) -> Result<Self, (ConnectError, Box<dyn Logger>)> {
 
        log!(cu.logger, "~~~ CONNECT called timeout {:?}", timeout);
 
        let deadline = timeout.map(|to| Instant::now() + to);
 
        let mut try_complete = || {
 
            // connect all endpoints in parallel; send and receive peer ids through ports
 
            let mut endpoint_manager = setup_endpoints_and_pair_ports(
 
                &mut *cu.logger,
 
                &setup.net_endpoint_setups,
 
                &setup.udp_endpoint_setups,
 
                &mut cu.ips.port_info,
 
                &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.
 
                session_optimize(&mut cu, &mut comm, &deadline)?;
 
            }
 
            log!(cu.logger, "connect() finished. setup phase complete");
 
            Ok(comm)
 
        };
 
        match try_complete() {
 
            Ok(comm) => {
 
                Ok(Self { unphased: cu, phased: ConnectorPhased::Communication(Box::new(comm)) })
 
            }
 
            Err(err) => Err((err, cu.logger)),
 
        }
 
    }
 
}
 

	
 
// 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],
 
    port_info: &mut PortInfoMap,
 
    deadline: &Option<Instant>,
 
) -> Result<EndpointManager, ConnectError> {
 
    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
0 comments (0 inline, 0 general)