New Timer & Peer IDs for Host:

Id: 1 - 7: Host Systems.
id: 8 - 99: User Timers (Hope thats enough)
Id: 100+ : DevP2P Session Peers

Author: SurfingNerd

crates/net/network-devp2p/src/host.rs

@@ -70,9 +70,23 @@ const DISCOVERY_REFRESH: TimerToken = 4;
 const FAST_DISCOVERY_REFRESH: TimerToken = 5;
 const DISCOVERY_ROUND: TimerToken = 6;
 const NODE_TABLE: TimerToken = 7;
-const USER_TIMER: TimerToken = 8;
 
-const FIRST_SESSION: StreamToken = 9;
+// Maximum count of peer mappings we remember.  each node ID takes 64 bytes for nodeID, 8 bytes for peer id, and probabyl 3 * 8 bytes for internals. = about 100 bytes per entry.)
+// 10000 elements should take about 1 MB of memory.
+const MAX_NODE_TO_PEER_MAPPINGS: usize = 10000;
+
+// the user timers are a collection of timers that are registered by the protocol handlers.
+// therefore comming from other modules.
+// to be not in conflict with the token ID system, we choose a very
+// high number for the user timers,
+// that is realisticly unreachable by the peer stream tokens,
+// but still have enough number space for user timers.
+const FIRST_USER_TIMER: TimerToken = 8;
+const MAX_USER_TIMERS: TimerToken = 91;
+const LAST_USER_TIMER: TimerToken = FIRST_USER_TIMER + MAX_USER_TIMERS;
+
+const FIRST_SESSION: StreamToken = LAST_USER_TIMER + 1;
+const LAST_SESSION: StreamToken = StreamToken::MAX;
 
 // Timeouts
 // for IDLE TimerToken
@@ -423,11 +437,11 @@ impl Host {
             discovery: Mutex::new(None),
             udp_socket: Mutex::new(None),
             tcp_listener: Mutex::new(tcp_listener),
-            sessions: SessionContainer::new(FIRST_SESSION, MAX_SESSIONS),
+            sessions: SessionContainer::new(FIRST_SESSION, MAX_SESSIONS, MAX_NODE_TO_PEER_MAPPINGS),
             nodes: RwLock::new(NodeTable::new(path)),
             handlers: RwLock::new(HashMap::new()),
             timers: RwLock::new(HashMap::new()),
-            timer_counter: RwLock::new(USER_TIMER),
+            timer_counter: RwLock::new(FIRST_USER_TIMER),
             reserved_nodes: RwLock::new(HashSet::new()),
             stopping: AtomicBool::new(false),
             filter,
@@ -1299,7 +1313,7 @@ impl IoHandler<NetworkIoMessage> for Host {
             return;
         }
         match stream {
-            FIRST_SESSION.. => self.session_readable(stream, io),
+            FIRST_SESSION..LAST_SESSION => self.session_readable(stream, io),
             DISCOVERY => self.discovery_readable(io),
             TCP_ACCEPT => self.accept(io),
             _ => panic!("Received unknown readable token"),
@@ -1311,22 +1325,20 @@ impl IoHandler<NetworkIoMessage> for Host {
             return;
         }
         match stream {
-            FIRST_SESSION.. => self.session_writable(stream, io),
+            FIRST_SESSION..=LAST_SESSION => self.session_writable(stream, io),
             DISCOVERY => self.discovery_writable(io),
             _ => panic!("Received unknown writable token"),
         }
     }
 
     fn timeout(&self, io: &IoContext<NetworkIoMessage>, token: TimerToken) {
+        trace!(target: "network", "HOST timout: {}", token);
         if self.stopping.load(AtomicOrdering::SeqCst) {
             return;
         }
         match token {
             IDLE => self.maintain_network(io),
-            FIRST_SESSION.. => {
-                trace!(target: "network", "Timeout from Host impl: {}", token);
-                self.connection_timeout(token, io);
-            }
+
             DISCOVERY_REFRESH => {
                 // Run the _slow_ discovery if enough peers are connected
                 if !self.has_enough_peers() {
@@ -1356,7 +1368,7 @@ impl IoHandler<NetworkIoMessage> for Host {
                 nodes.clear_useless();
                 nodes.save();
             }
-            _ => match self.timers.read().get(&token).cloned() {
+            FIRST_USER_TIMER..=LAST_USER_TIMER => match self.timers.read().get(&token).cloned() {
                 Some(timer) => match self.handlers.read().get(&timer.protocol).cloned() {
                     None => {
                         warn!(target: "network", "No handler found for protocol: {:?}", timer.protocol)
@@ -1380,6 +1392,13 @@ impl IoHandler<NetworkIoMessage> for Host {
                     warn!("Unknown timer token: {}", token);
                 } // timer is not registerd through us
             },
+            FIRST_SESSION..=LAST_SESSION => {
+                trace!(target: "network", "Timeout from Host impl: {}", token);
+                self.connection_timeout(token, io);
+            }
+            _ => {
+                warn!(target: "network", "HOST: Unknown timer token tick: {}", token);
+            }
         }
     }
 
@@ -1419,20 +1438,33 @@ impl IoHandler<NetworkIoMessage> for Host {
                 ref token,
             } => {
                 trace!(target: "network", "Adding timer for protocol: {:?}, delay: {:?}, token: {}", protocol, delay.as_millis(), token);
+
+                if token >= &MAX_USER_TIMERS {
+                    warn!(target: "network", "Tried to register timer with token {} which is larger than MAX_USER_TIMERS {}", token, MAX_USER_TIMERS);
+                    return;
+                }
+
                 let handler_token = {
                     let mut timer_counter = self.timer_counter.write();
                     let counter = &mut *timer_counter;
                     let handler_token = *counter;
                     *counter += 1;
                     handler_token
                 };
+
+                if handler_token > LAST_USER_TIMER {
+                    warn!(target: "network", "Tried to register timer witch Index {token} what would be over the boundaries of usertimers: {LAST_USER_TIMER}");
+                    return;
+                }
                 self.timers.write().insert(
                     handler_token,
                     ProtocolTimer {
                         protocol: *protocol,
                         token: *token,
                     },
                 );
+
+                trace!(target: "network", "Registering handler_token {} token {} for protocol {:?} with delay {:?}", handler_token, token, protocol, delay);
                 io.register_timer(handler_token, *delay)
                     .unwrap_or_else(|e| debug!("Error registering timer {}: {:?}", token, e));
             }

crates/net/network-devp2p/src/session_container.rs

@@ -1,5 +1,6 @@
 use crate::host::HostInfo;
 use ethereum_types::H256;
+use lru_cache::LruCache;
 use mio::net::TcpStream;
 use std::{collections::BTreeMap, sync::Arc, time::Duration};
 
@@ -19,16 +20,16 @@ pub struct SessionContainer {
     max_sessions: usize,
     sessions: Arc<RwLock<std::collections::BTreeMap<usize, SharedSession>>>,
     expired_sessions: Arc<RwLock<Vec<SharedSession>>>,
-    node_id_to_session: Mutex<BTreeMap<ethereum_types::H512, usize>>, // used to map Node IDs to last used session tokens.
+    node_id_to_session: Mutex<LruCache<ethereum_types::H512, usize>>, // used to map Node IDs to last used session tokens.
     sessions_token_max: Mutex<usize>, // Used to generate new session tokens
 }
 
 impl SessionContainer {
-    pub fn new(first_session_token: usize, max_sessions: usize) -> Self {
+    pub fn new(first_session_token: usize, max_sessions: usize, max_node_mappings: usize) -> Self {
         SessionContainer {
             sessions: Arc::new(RwLock::new(std::collections::BTreeMap::new())),
             expired_sessions: Arc::new(RwLock::new(Vec::new())),
-            node_id_to_session: Mutex::new(BTreeMap::new()),
+            node_id_to_session: Mutex::new(LruCache::new(max_node_mappings)),
             sessions_token_max: Mutex::new(first_session_token),
             max_sessions,
         }
@@ -47,7 +48,7 @@ impl SessionContainer {
     fn create_token_id(
         &self,
         node_id: &NodeId,
-        tokens: &mut BTreeMap<ethereum_types::H512, usize>,
+        tokens: &mut LruCache<ethereum_types::H512, usize>,
     ) -> usize {
         let mut session_token_max = self.sessions_token_max.lock();
         let next_id = session_token_max.clone();
@@ -132,7 +133,7 @@ impl SessionContainer {
 
         if let Some(node_id) = id {
             // check if there is already a connection for the given node id.
-            if let Some(existing_peer_id) = node_ids.get(node_id) {
+            if let Some(existing_peer_id) = node_ids.get_mut(node_id) {
                 let existing_session_mutex_o = sessions.get(existing_peer_id).clone();
 
                 if let Some(existing_session_mutex) = existing_session_mutex_o {
@@ -279,7 +280,7 @@ impl SessionContainer {
     pub fn get_session_for(&self, id: &NodeId) -> Option<SharedSession> {
         self.node_id_to_session
             .lock()
-            .get(id)
+            .get_mut(id)
             .cloned()
             .map_or(None, |peer_id| {
                 let sessions = self.sessions.read();
@@ -294,7 +295,7 @@ impl SessionContainer {
     ) -> Option<usize> {
         self.node_id_to_session
             .lock()
-            .get(node_id)
+            .get_mut(node_id)
             .map_or(None, |peer_id| {
                 if !only_available_sessions {
                     return Some(*peer_id);