Remove redundant listening hashmap from uTP listener and clear dead code

ogenev · ogenev · commit 9a930fe1d135 · 2022-05-25T14:37:08.000+03:00
diff --git a/trin-core/src/portalnet/overlay.rs b/trin-core/src/portalnet/overlay.rs
@@ -101,7 +101,7 @@ impl<TContentKey: OverlayContentKey + Send, TMetric: Metric + Send>
         config: OverlayConfig,
         discovery: Arc<Discovery>,
         utp_listener_tx: UnboundedSender<UtpListenerRequest>,
-        utp_listener_rx: Arc<TRwLock<UnboundedReceiver<UtpListenerEvent>>>,
+        _utp_listener_rx: Arc<TRwLock<UnboundedReceiver<UtpListenerEvent>>>,
         storage: Arc<RwLock<PortalStorage>>,
         data_radius: U256,
         protocol: ProtocolId,
@@ -124,7 +124,6 @@ impl<TContentKey: OverlayContentKey + Send, TMetric: Metric + Send>
             Arc::clone(&data_radius),
             protocol.clone(),
             utp_listener_tx.clone(),
-            utp_listener_rx,
             config.enable_metrics,
         )
         .await
@@ -285,13 +284,6 @@ impl<TContentKey: OverlayContentKey + Send, TMetric: Metric + Send>
         enr: Enr,
         conn_id: u16,
     ) -> Result<Content, OverlayRequestError> {
-        let utp_request = UtpListenerRequest::FindContentStream(conn_id);
-        if let Err(err) = self.utp_listener_tx.send(utp_request) {
-            return Err(OverlayRequestError::UtpError(format!(
-                "Unable to send uTP FindContent stream request: {err}"
-            )));
-        }
-
         // initiate the connection to the acceptor
         let (tx, rx) = tokio::sync::oneshot::channel::<anyhow::Result<UtpSocket>>();
         self.utp_listener_tx
@@ -401,11 +393,6 @@ impl<TContentKey: OverlayContentKey + Send, TMetric: Metric + Send>
             return Ok(response);
         }
 
-        let utp_request = UtpListenerRequest::OfferStream(conn_id);
-        if let Err(err) = self.utp_listener_tx.send(utp_request) {
-            return Err(anyhow!("Unable to send uTP Offer stream request: {err}"));
-        }
-
         // initiate the connection to the acceptor
         let (tx, rx) = tokio::sync::oneshot::channel::<anyhow::Result<UtpSocket>>();
 
diff --git a/trin-core/src/portalnet/overlay_service.rs b/trin-core/src/portalnet/overlay_service.rs
@@ -19,13 +19,7 @@ use crate::{
     utp::stream::UtpListenerRequest,
 };
 
-use crate::{
-    portalnet::types::content_key::RawContentKey,
-    utp::{
-        stream::{UtpListenerEvent, UtpPayload},
-        trin_helpers::UtpStreamId,
-    },
-};
+use crate::utp::trin_helpers::UtpStreamId;
 use delay_map::HashSetDelay;
 use discv5::{
     enr::NodeId,
@@ -43,10 +37,7 @@ use rand::seq::SliceRandom;
 use ssz::Encode;
 use ssz_types::{BitList, VariableList};
 use thiserror::Error;
-use tokio::sync::{
-    mpsc::{self, UnboundedReceiver, UnboundedSender},
-    RwLock as TRwLock,
-};
+use tokio::sync::mpsc::{self, UnboundedReceiver, UnboundedSender};
 
 /// Maximum number of ENRs in response to FindNodes.
 pub const FIND_NODES_MAX_NODES: usize = 32;
@@ -265,8 +256,6 @@ pub struct OverlayService<TContentKey, TMetric> {
     response_tx: UnboundedSender<OverlayResponse>,
     /// The sender half of a channel to send requests to uTP listener
     utp_listener_tx: UnboundedSender<UtpListenerRequest>,
-    /// Receiver for UtpListener emitted events
-    utp_listener_rx: Arc<TRwLock<UnboundedReceiver<UtpListenerEvent>>>,
     /// Phantom content key.
     phantom_content_key: PhantomData<TContentKey>,
     /// Phantom metric (distance function).
@@ -292,7 +281,6 @@ impl<TContentKey: OverlayContentKey + Send, TMetric: Metric + Send>
         data_radius: Arc<U256>,
         protocol: ProtocolId,
         utp_listener_sender: UnboundedSender<UtpListenerRequest>,
-        utp_listener_receiver: Arc<TRwLock<UnboundedReceiver<UtpListenerEvent>>>,
         enable_metrics: bool,
     ) -> Result<UnboundedSender<OverlayRequest>, String> {
         let (request_tx, request_rx) = mpsc::unbounded_channel();
@@ -325,7 +313,6 @@ impl<TContentKey: OverlayContentKey + Send, TMetric: Metric + Send>
                 response_rx,
                 response_tx,
                 utp_listener_tx: utp_listener_sender,
-                utp_listener_rx: utp_listener_receiver,
                 phantom_content_key: PhantomData,
                 phantom_metric: PhantomData,
                 metrics,
@@ -397,9 +384,6 @@ impl<TContentKey: OverlayContentKey + Send, TMetric: Metric + Send>
             tokio::time::interval(Duration::from_secs(BUCKET_REFRESH_INTERVAL));
 
         loop {
-            // let utp_listener_rx = self.utp_listener_rx.clone();
-            // let mut utp_listener_lock = utp_listener_rx.write_with_warn().await;
-
             tokio::select! {
                 Some(request) = self.request_rx.recv() => self.process_request(request),
                 Some(response) = self.response_rx.recv() => {
@@ -428,14 +412,6 @@ impl<TContentKey: OverlayContentKey + Send, TMetric: Metric + Send>
                         self.peers_to_ping.insert(node_id);
                     }
                 }
-                // Some(utp_event) = utp_listener_lock.recv() => {
-                //     match utp_event {
-                //         UtpListenerEvent::ProcessedClosedStreams(processed_streams) =>
-                //         {
-                //             self.handle_utp_payload(processed_streams);
-                //         }
-                //     }
-                // }
                 _ = OverlayService::<TContentKey, TMetric>::bucket_maintenance_poll(self.protocol.clone(), &self.kbuckets) => {}
                 _ = bucket_refresh_interval.tick() => {
                     debug!("[{:?}] Overlay bucket refresh lookup", self.protocol);
@@ -666,26 +642,11 @@ impl<TContentKey: OverlayContentKey + Send, TMetric: Metric + Send>
                 } else {
                     let conn_id: u16 = crate::utp::stream::rand();
 
-                    // listen for incoming connection request on conn_id, as part of utp handshake and
-                    // temporarily storing content data, so we can send it right after we receive
+                    // Listen for incoming uTP connection request on as part of uTP handshake and
+                    // storing content data, so we can send it inside UtpListener right after we receive
                     // SYN packet from the requester
-                    let utp_request = UtpListenerRequest::FindContentData(conn_id, content.clone());
-                    if let Err(err) = self.utp_listener_tx.send(utp_request) {
-                        return Err(OverlayRequestError::UtpError(format!(
-                            "Unable to send uTP FindContentData stream request: {err}"
-                        )));
-                    }
-
-                    // also listen on conn_id + 1 because this is the receive path
                     let conn_id_recv = conn_id.wrapping_add(1);
 
-                    let utp_request = UtpListenerRequest::FindContentStream(conn_id_recv);
-                    if let Err(err) = self.utp_listener_tx.send(utp_request) {
-                        return Err(OverlayRequestError::UtpError(format!(
-                            "Unable to send uTP FindContent stream request: {err}"
-                        )));
-                    }
-
                     self.add_utp_connection(
                         source,
                         conn_id_recv,
@@ -704,8 +665,7 @@ impl<TContentKey: OverlayContentKey + Send, TMetric: Metric + Send>
                 }
             }
             Err(msg) => Err(OverlayRequestError::Failure(format!(
-                "Unable to respond to FindContent: {}",
-                msg
+                "Unable to respond to FindContent: {msg}",
             ))),
         }
     }
@@ -748,24 +708,8 @@ impl<TContentKey: OverlayContentKey + Send, TMetric: Metric + Send>
                 })?;
         }
 
-        // listen for incoming connection request on conn_id, as part of utp handshake
+        // Listen for incoming connection request on conn_id, as part of utp handshake
         let conn_id: u16 = crate::utp::stream::rand();
-        let utp_request = UtpListenerRequest::OfferStream(conn_id);
-        if let Err(err) = self.utp_listener_tx.send(utp_request) {
-            return Err(OverlayRequestError::UtpError(format!(
-                "Unable to send uTP Offer stream request: {err}"
-            )));
-        }
-
-        // also listen on conn_id + 1 because this is the actual receive path for acceptor
-        let conn_id_recv = conn_id.wrapping_add(1);
-
-        let utp_request = UtpListenerRequest::AcceptStream(conn_id_recv, accept_keys.clone());
-        if let Err(err) = self.utp_listener_tx.send(utp_request) {
-            return Err(OverlayRequestError::UtpError(format!(
-                "Unable to send uTP Accept stream request: {err}"
-            )));
-        }
 
         self.add_utp_connection(source, conn_id, UtpStreamId::AcceptStream(accept_keys))?;
 
@@ -777,19 +721,6 @@ impl<TContentKey: OverlayContentKey + Send, TMetric: Metric + Send>
         Ok(accept)
     }
 
-    /// Handle all closed uTP streams, currently we process only AcceptStream here.
-    /// FindContent payload is processed explicitly when we send FindContent request.
-    fn handle_utp_payload(&self, streams: Vec<(UtpPayload, UtpStreamId)>) {
-        for stream in streams {
-            match stream {
-                (payload, UtpStreamId::AcceptStream(content_keys)) => {
-                    self.process_accept_utp_payload(content_keys, payload);
-                }
-                _ => {}
-            }
-        }
-    }
-
     /// Sends a TALK request via Discovery v5 to some destination node.
     fn send_talk_req(&self, request: Request, request_id: OverlayRequestId, destination: Enr) {
         let discovery = Arc::clone(&self.discovery);
@@ -821,12 +752,6 @@ impl<TContentKey: OverlayContentKey + Send, TMetric: Metric + Send>
         });
     }
 
-    /// Process accepted uTP payload of the OFFER?ACCEPT stream
-    fn process_accept_utp_payload(&self, content_keys: Vec<RawContentKey>, payload: UtpPayload) {
-        // TODO: Verify the payload, store the content and propagate gossip.
-        debug!("DEBUG: Processing content keys: {content_keys:?}, with payload: {payload:?}, protocol: {:?}", self.protocol);
-    }
-
     /// Processes an incoming request from some source node.
     fn process_incoming_request(&mut self, request: Request, _id: RequestId, source: NodeId) {
         // Look up the node in the routing table.
@@ -1354,7 +1279,6 @@ mod tests {
         let discovery = Arc::new(Discovery::new(portal_config).unwrap());
 
         let (utp_listener_tx, _) = unbounded_channel::<UtpListenerRequest>();
-        let (_, utp_listener_rx) = unbounded_channel::<UtpListenerEvent>();
 
         // Initialize DB config
         let storage_capacity: u32 = DEFAULT_STORAGE_CAPACITY.parse().unwrap();
@@ -1392,7 +1316,6 @@ mod tests {
             response_tx,
             response_rx,
             utp_listener_tx,
-            utp_listener_rx: Arc::new(TRwLock::new(utp_listener_rx)),
             phantom_content_key: PhantomData,
             phantom_metric: PhantomData,
             metrics,
diff --git a/trin-core/src/utp/stream.rs b/trin-core/src/utp/stream.rs
@@ -21,10 +21,7 @@ use tokio::{
 
 use crate::{
     locks::RwLoggingExt,
-    portalnet::types::{
-        content_key::RawContentKey,
-        messages::{ByteList, ProtocolId},
-    },
+    portalnet::types::messages::ProtocolId,
     utp::{
         packets::{ExtensionType, Packet, PacketType, HEADER_SIZE},
         time::{now_microseconds, Delay, Timestamp},
@@ -114,8 +111,6 @@ struct DelayDifferenceSample {
 /// Represent overlay to uTP listener request. It is used as a way to communicate between the overlay protocol
 /// and uTP listener
 pub enum UtpListenerRequest {
-    /// Request to listen for Accept stream
-    AcceptStream(ConnId, Vec<RawContentKey>),
     /// Request to initialize uTP stream with remote node
     Connect(
         ConnId,
@@ -126,12 +121,6 @@ pub enum UtpListenerRequest {
     ),
     /// Request to add uTP stream to the active connections
     AddActiveConnection(Enr, ProtocolId, UtpStreamId, ConnId),
-    /// Request to listen for FindCOntent stream and send content data
-    FindContentData(ConnId, ByteList),
-    /// Request to listen for FindContent stream
-    FindContentStream(ConnId),
-    /// Request to listen for Offer stream
-    OfferStream(ConnId),
 }
 
 /// Result from processing all closed uTP streams. Includes a tuple with the payload and the stream id.
@@ -152,13 +141,13 @@ pub enum UtpStreamEvent {
 }
 
 /// Main uTP service used to listen and handle all uTP connections and streams
+// FIXME: Deny dead_code
+#[allow(dead_code)]
 pub struct UtpListener {
     /// Base discv5 layer
     discovery: Arc<Discovery>,
     /// Store all active connections
     utp_connections: HashMap<ConnectionKey, UtpSocket>,
-    /// uTP connection ids to listen for
-    listening: HashMap<ConnId, UtpStreamId>,
     /// Receiver for uTP events sent from the main portal event handler
     utp_event_rx: UnboundedReceiver<TalkRequest>,
     /// Sender to overlay layer with processed uTP stream
@@ -196,7 +185,6 @@ impl UtpListener {
             UtpListener {
                 discovery,
                 utp_connections: HashMap::new(),
-                listening: HashMap::new(),
                 utp_event_rx,
                 overlay_tx,
                 overlay_rx: utp_listener_rx,
@@ -218,14 +206,6 @@ impl UtpListener {
                     Some(overlay_request) = self.overlay_rx.recv() => {
                         self.process_overlay_request(overlay_request).await
                     },
-                //     _ = process_utp_streams_interval.tick() => {
-                //         let processed_streams = self.process_closed_streams();
-                //
-                //         if let Err(err) = self.overlay_tx.send(UtpListenerEvent::ProcessedClosedStreams(processed_streams)) {
-                //             error!("Unable to send ProcessClosedStreams event to overlay layer: {err}");
-                //             continue
-                //         }
-                // }
             }
         }
     }
@@ -270,15 +250,9 @@ impl UtpListener {
                                 return;
                             }
 
-                            // Get ownership of FindContentData and re-add the receiver connection
-                            let utp_message_id = self.listening.remove(&conn.sender_connection_id);
-
-                            // TODO: Probably there is a better way with lifetimes to pass the HashMap value to a
-                            // different thread without removing the key and re-adding it.
-                            self.listening
-                                .insert(conn.sender_connection_id, UtpStreamId::FindContentStream);
-
-                            if let Some(UtpStreamId::FindContentData(content_data)) = utp_message_id
+                            // Send content data if the stream is listening for FindContent SYN packet
+                            if let UtpStreamId::FindContentData(content_data) =
+                                conn.stream_id.clone()
                             {
                                 // We want to send uTP data only if the content is Content(ByteList)
                                 debug!(
@@ -390,27 +364,12 @@ impl UtpListener {
                 let conn_key = ConnectionKey::new(connected_to.node_id(), conn_id_recv);
                 self.utp_connections.insert(conn_key, conn);
             }
-            UtpListenerRequest::FindContentStream(conn_id) => {
-                self.listening
-                    .insert(conn_id, UtpStreamId::FindContentStream);
-            }
             UtpListenerRequest::Connect(conn_id, node_id, protocol_id, stream_id, tx) => {
                 let conn = self.connect(conn_id, node_id, protocol_id, stream_id).await;
                 if tx.send(conn).is_err() {
                     error!("Unable to send uTP socket to requester")
                 };
             }
-            UtpListenerRequest::OfferStream(conn_id) => {
-                self.listening.insert(conn_id, UtpStreamId::OfferStream);
-            }
-            UtpListenerRequest::FindContentData(conn_id, content) => {
-                self.listening
-                    .insert(conn_id, UtpStreamId::FindContentData(content));
-            }
-            UtpListenerRequest::AcceptStream(conn_id, accepted_keys) => {
-                self.listening
-                    .insert(conn_id, UtpStreamId::AcceptStream(accepted_keys));
-            }
         }
     }
 
@@ -440,34 +399,6 @@ impl UtpListener {
             Err(anyhow!("Trying to connect to unknow Enr"))
         }
     }
-
-    /// Return and cleanup all active uTP streams where socket state is "Closed"
-    pub fn process_closed_streams(&mut self) -> Vec<(UtpPayload, UtpStreamId)> {
-        // This seems to be a hot loop, we may need to optimise it and find a better way to filter by closed
-        // connections without cloning all records. One reasonable way is to use some data-oriented
-        // design principles like Struct of Arrays vs. Array of Structs.
-        self.utp_connections
-            .clone()
-            .iter()
-            .filter(|conn| conn.1.state == SocketState::Closed)
-            .map(|conn| {
-                // Remove the closed connections from active connections
-                let receiver_stream_id = self
-                    .listening
-                    .remove(&conn.1.receiver_connection_id)
-                    .expect("Receiver connection id should match active listening connections.");
-                self.listening
-                    .remove(&conn.1.sender_connection_id)
-                    .expect("Sender connection id should match active listening connections.");
-                let utp_socket = self
-                    .utp_connections
-                    .remove(conn.0)
-                    .expect("uTP socket should match asctive utp connections.");
-
-                (utp_socket.recv_data_stream, receiver_stream_id)
-            })
-            .collect()
-    }
 }
 
 // Used to be MicroTransportProtocol impl but it is basically just called UtpStream compared to the
diff --git a/utp-testing/src/main.rs b/utp-testing/src/main.rs
