CSY/reowolf Changeset - 1a61d3b3d964 · Centrum Wiskunde & Informatica (CWI)

Changeset - 1a61d3b3d964

Parent rev.

Child rev.

[Not reviewed]

0 1 0

Christopher Esterhuyse - 5 years ago 2020-07-08 14:53:44
christopher.esterhuyse@gmail.com

setup procedure refactor and cleanup

1 file changed with 32 insertions and 27 deletions:

src/runtime/setup.rs

0 comments (0 inline, 0 general)

src/runtime/setup.rs

➞

Show inline comments

@@ @@ -145,328 +145,333 @@ fn new_endpoint_manager( @@
         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
+    }
     struct UdpTodo {
         local_port: PortId,
         sock: UdpSocket,
+    }
     enum TodoEndpoint {
         Accepting(TcpListener),
         NetEndpoint(NetEndpoint),
+    }
     ////////////////////////////////////////////
     // 1. Start to construct EndpointManager
     const WAKER_PERIOD: Duration = Duration::from_millis(300);
     struct WakerState {
         continue_signal: AtomicBool,
         waker: mio::Waker,
+    }
     let mut waker_state: Option<Arc<WakerState>> = None;
     let mut poll = Poll::new().map_err(|_| Ce::PollInitFailed)?;
     let mut events = Events::with_capacity(net_endpoint_setups.len() * 2 + 4);
     let mut net_polled_undrained = VecSet::default();
     let udp_polled_undrained = VecSet::default();
     let [mut net_polled_undrained, udp_polled_undrained] = [VecSet::default(), VecSet::default()];
     let mut delayed_messages = vec![];
     // 2. create a registered (TcpListener/Endpoint) for passive / active respectively
-    let mut todos = net_endpoint_setups
+    let mut net_todos = net_endpoint_setups
         .iter()
         .enumerate()
         .map(|(index, (local_port, endpoint_setup))| {
             let token = TokenTarget::NetEndpoint { index }.into();
             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();
                 TodoEndpoint::NetEndpoint(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();
                 TodoEndpoint::Accepting(listener)
             };
             Ok(Todo {
                 todo_endpoint,
                 local_port: *local_port,
                 sent_local_port: false,
                 recv_peer_port: None,
                 endpoint_setup: endpoint_setup.clone(),
             })
         })
         .collect::<Result<Vec<Todo>, ConnectError>>()?;
     let udp_todos = udp_endpoint_setups
         .iter()
         .enumerate()
         .map(|(index, (local_port, endpoint_setup))| {
             let mut sock = UdpSocket::bind(endpoint_setup.local_addr)
                 .map_err(|_| Ce::BindFailed(endpoint_setup.local_addr))?;
             poll.registry()
                 .register(&mut sock, TokenTarget::UdpEndpoint { index }.into(), Interest::WRITABLE)
                 .unwrap();
             Ok(UdpTodo { sock, local_port: *local_port })
         })
         .collect::<Result<Vec<UdpTodo>, ConnectError>>()?;
     // 3. Using poll to drive progress:
     //    - accept an incoming connection for each TcpListener (turning them into endpoints too)
     //    - for each endpoint, send the local PortId
     //    - for each endpoint, recv the peer's PortId, and
     // all in connect_failed are NOT registered with Poll
     let mut connect_failed: HashSet<usize> = Default::default();
     // TODO register udps, and all them to incomplete list
     let mut setup_incomplete: HashSet<TokenTarget> = (0..todos.len())
         .map(|index| TokenTarget::NetEndpoint { index })
         .chain((0..udp_todos.len()).map(|index| TokenTarget::UdpEndpoint { index }))
         .collect();
     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()
     };
     while !setup_incomplete.is_empty() {
         let remaining = if let Some(deadline) = deadline {
             Some(deadline.checked_duration_since(Instant::now()).ok_or(Ce::Timeout)?)
         } else {
             None
         };
         poll.poll(&mut events, remaining).map_err(|_| Ce::PollFailed)?;
         for event in events.iter() {
             let token = event.token();
             let token_target = TokenTarget::from(token);
             if !setup_incomplete.contains(&token_target) {
                 // spurious wakeup
                 continue;
+            }
             match token_target {
                 TokenTarget::Waker => {
                     log!(
                         logger,
                         "Notification from waker. connect_failed is {:?}",
                         connect_failed.iter()
                     );
                     assert!(waker_state.is_some());
                     for net_index in connect_failed.drain() {
-                        let todo: &mut Todo = &mut todos[net_index];
+                        let net_todo = &mut net_todos[net_index];
                         log!(
                             logger,
                             "Restarting connection with endpoint {:?} {:?}",
                             net_index,
-                            todo.endpoint_setup.sock_addr
+                            net_todo.endpoint_setup.sock_addr
                         );
-                        match &mut todo.todo_endpoint {
+                        match &mut net_todo.todo_endpoint {
                             TodoEndpoint::NetEndpoint(endpoint) => {
                                 let mut new_stream =
-                                    TcpStream::connect(todo.endpoint_setup.sock_addr)
+                                    TcpStream::connect(net_todo.endpoint_setup.sock_addr)
                                         .expect("mio::TcpStream connect should not fail!");
                                 std::mem::swap(&mut endpoint.stream, &mut new_stream);
                                 let token = TokenTarget::NetEndpoint { index: net_index }.into();
                                 poll.registry()
                                     .register(&mut endpoint.stream, token, BOTH)
                                     .unwrap();
+                            }
                             _ => unreachable!(),
+                        }
+                    }
+                }
                 TokenTarget::UdpEndpoint { index } => {
                     let udp_todo: &UdpTodo = &udp_todos[index];
                     if event.is_error() {
                         return Err(Ce::BindFailed(udp_todo.sock.local_addr().unwrap()));
+                    }
                     setup_incomplete.remove(&token_target);
+                }
                 TokenTarget::NetEndpoint { index } => {
                     let todo: &mut Todo = &mut todos[index];
                     if let TodoEndpoint::Accepting(listener) = &mut todo.todo_endpoint {
                     let net_todo = &mut net_todos[index];
                     if let TodoEndpoint::Accepting(listener) = &mut net_todo.todo_endpoint {
                         // FIRST try complete this connection
                         match listener.accept() {
                             Err(e) if would_block(&e) => {
                                 log!(logger, "Spurious wakeup on listener {:?}", index)
+                            }
                             Err(_) => {
                                 log!(logger, "accept() failure on index {}", index);
                                 return Err(Ce::AcceptFailed(listener.local_addr().unwrap()));
+                            }
                             Ok((mut stream, peer_addr)) => {
                                 // success!
                                 poll.registry().deregister(listener).unwrap();
                                 // reusing original token as-is
                                 poll.registry().register(&mut stream, token, BOTH).unwrap();
                                 log!(
                                     logger,
                                     "Endpoint[{}] accepted a connection from {:?}",
                                     index,
                                     peer_addr
                                 );
                                 let net_endpoint = NetEndpoint { stream, inbox: vec![] };
-                                todo.todo_endpoint = TodoEndpoint::NetEndpoint(net_endpoint);
+                                net_todo.todo_endpoint = TodoEndpoint::NetEndpoint(net_endpoint);
+                            }
+                        }
+                    }
-                    if let TodoEndpoint::NetEndpoint(net_endpoint) = &mut todo.todo_endpoint {
+                    if let TodoEndpoint::NetEndpoint(net_endpoint) = &mut net_todo.todo_endpoint {
                         if event.is_error() {
                             if todo.endpoint_setup.endpoint_polarity == EndpointPolarity::Passive {
                             if net_todo.endpoint_setup.endpoint_polarity
                                 == EndpointPolarity::Passive
+                            {
                                 // right now you cannot retry an acceptor. return failure
                                 return Err(Ce::AcceptFailed(
                                     net_endpoint.stream.local_addr().unwrap(),
                                 ));
+                            }
                             // this actively-connecting endpoint failed to connect!
                             if connect_failed.insert(index) {
                                 log!(
                                     logger,
                                     "Connection failed for {:?}. List is {:?}",
                                     index,
                                     connect_failed.iter()
                                 );
                                 poll.registry().deregister(&mut net_endpoint.stream).unwrap();
                             } else {
                                 // spurious wakeup.
                                 continue;
+                            }
                             if waker_state.is_none() {
                                 log!(logger, "First connect failure. Starting waker thread");
                                 let arc = Arc::new(WakerState {
                                     waker: mio::Waker::new(
                                         poll.registry(),
                                         TokenTarget::Waker.into(),
+                                    )
                                     .unwrap(),
                                     continue_signal: true.into(),
                                 });
                                 let moved_arc = arc.clone();
                                 waker_state = Some(arc);
                                 std::thread::spawn(move || {
                                     while moved_arc
                                         .continue_signal
                                         .load(std::sync::atomic::Ordering::SeqCst)
+                                    {
                                         std::thread::sleep(WAKER_PERIOD);
                                         let _ = moved_arc.waker.wake();
+                                    }
                                 });
+                            }
                             continue;
+                        }
                         // event wasn't ERROR
                         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 local_polarity =
                             *port_info.polarities.get(&net_todo.local_port).unwrap();
                         if event.is_writable() && !net_todo.sent_local_port {
                             // can write and didn't send setup msg yet? Do so!
                             let msg = Msg::SetupMsg(SetupMsg::MyPortInfo(MyPortInfo {
                                 polarity: local_polarity,
-                                port: todo.local_port,
+                                port: net_todo.local_port,
                             }));
                             net_endpoint
                                 .send(&msg)
                                 .map_err(|e| {
                                     Ce::NetEndpointSetupError(
                                         net_endpoint.stream.local_addr().unwrap(),
                                         e,
+                                    )
                                 })
                                 .unwrap();
                             log!(logger, "endpoint[{}] sent msg {:?}", index, &msg);
-                            todo.sent_local_port = true;
+                            net_todo.sent_local_port = true;
+                        }
-                        if event.is_readable() && todo.recv_peer_port.is_none() {
+                        if event.is_readable() && net_todo.recv_peer_port.is_none() {
                             // can read and didn't recv setup msg yet? Do so!
                             let maybe_msg = net_endpoint.try_recv(logger).map_err(|e| {
                                 Ce::NetEndpointSetupError(
                                     net_endpoint.stream.local_addr().unwrap(),
                                     e,
+                                )
                             })?;
                             if maybe_msg.is_some() && !net_endpoint.inbox.is_empty() {
                                 net_polled_undrained.insert(index);
+                            }
                             match maybe_msg {
                                 None => {} // msg deserialization incomplete
                                 Some(Msg::SetupMsg(SetupMsg::MyPortInfo(peer_info))) => {
                                     log!(
                                         logger,
                                         "endpoint[{}] got peer info {:?}",
                                         index,
                                         peer_info
                                     );
                                     if peer_info.polarity == local_polarity {
                                         return Err(ConnectError::PortPeerPolarityMismatch(
-                                            todo.local_port,
+                                            net_todo.local_port,
                                         ));
+                                    }
-                                    todo.recv_peer_port = Some(peer_info.port);
+                                    net_todo.recv_peer_port = Some(peer_info.port);
                                     // 1. finally learned the peer of this port!
-                                    port_info.peers.insert(todo.local_port, peer_info.port);
+                                    port_info.peers.insert(net_todo.local_port, peer_info.port);
                                     // 2. learned the info of this peer port
                                     port_info.polarities.insert(peer_info.port, peer_info.polarity);
-                                    port_info.peers.insert(peer_info.port, todo.local_port);
+                                    port_info.peers.insert(peer_info.port, net_todo.local_port);
                                     if let Some(route) = port_info.routes.get(&peer_info.port) {
                                         // check just for logging purposes
                                         log!(
                                             logger,
                                             "Special case! Route to peer {:?} already known to be {:?}. Leave untouched",
                                             peer_info.port,
                                             route
                                         );
+                                    }
                                     port_info
                                         .routes
                                         .entry(peer_info.port)
                                         .or_insert(Route::NetEndpoint { index });
+                                }
                                 Some(inappropriate_msg) => {
                                     log!(
                                         logger,
                                         "delaying msg {:?} during channel setup phase",
                                         inappropriate_msg
                                     );
                                     delayed_messages.push((index, inappropriate_msg));
+                                }
+                            }
+                        }
                         // is the setup for this net_endpoint now complete?
-                        if todo.sent_local_port && todo.recv_peer_port.is_some() {
+                        if net_todo.sent_local_port && net_todo.recv_peer_port.is_some() {
                             // yes! connected, sent my info and received peer's info
                             setup_incomplete.remove(&token_target);
                             log!(logger, "endpoint[{}] is finished!", index);
+                        }
+                    }
+                }
+            }
+        }
         events.clear();
+    }
     log!(logger, "Endpoint setup complete! Cleaning up and building structures");
     if let Some(arc) = waker_state {
         log!(logger, "Sending waker the stop signal");
         arc.continue_signal.store(false, std::sync::atomic::Ordering::SeqCst);
         // TODO leave the waker registered?
+    }
-    let net_endpoint_exts = todos
+    let net_endpoint_exts = net_todos
         .into_iter()
         .enumerate()
         .map(|(index, Todo { todo_endpoint, local_port, .. })| NetEndpointExt {
             net_endpoint: match todo_endpoint {
                 TodoEndpoint::NetEndpoint(mut net_endpoint) => {
                     let token = TokenTarget::NetEndpoint { index }.into();
                     poll.registry()
                         .reregister(&mut net_endpoint.stream, token, Interest::READABLE)
                         .unwrap();
                     net_endpoint
+                }
                 _ => unreachable!(),
             },
             getter_for_incoming: local_port,
         })
         .collect();
     let udp_endpoint_exts = udp_todos
         .into_iter()
         .enumerate()
         .map(|(index, udp_todo)| {
             let UdpTodo { mut sock, local_port } = udp_todo;
             let token = TokenTarget::UdpEndpoint { index }.into();
             poll.registry().reregister(&mut sock, token, Interest::READABLE).unwrap();
             UdpEndpointExt {

0 comments (0 inline, 0 general)