add peer print & fix tick issue

Author: erhant

compute/src/node.rs

@@ -319,8 +319,10 @@ impl DriaComputeNode {
         // prepare durations for sleeps
         let mut peer_refresh_interval =
             tokio::time::interval(Duration::from_secs(DIAGNOSTIC_REFRESH_INTERVAL_SECS));
+        peer_refresh_interval.tick().await; // move one tick
         let mut available_node_refresh_interval =
             tokio::time::interval(Duration::from_secs(AVAILABLE_NODES_REFRESH_INTERVAL_SECS));
+        available_node_refresh_interval.tick().await; // move one tick
 
         // subscribe to topics
         self.subscribe(PingpongHandler::LISTEN_TOPIC).await?;

monitor/src/node.rs

@@ -17,6 +17,7 @@ use tokio::sync::mpsc;
 use tokio_util::sync::CancellationToken;
 
 const TASK_PRINT_INTERVAL_SECS: u64 = 20;
+const PEER_PRINT_INTERVAL_SECS: u64 = 40;
 
 pub struct DriaMonitorNode {
     pub p2p: DriaP2PCommander,
@@ -39,13 +40,20 @@ impl DriaMonitorNode {
             results: HashMap::new(),
         }
     }
+
+    /// Setup the monitor node.
+    ///
+    /// Subscribes to task topics.
     pub async fn setup(&self) -> Result<()> {
         self.p2p.subscribe(WorkflowHandler::LISTEN_TOPIC).await?;
         self.p2p.subscribe(WorkflowHandler::RESPONSE_TOPIC).await?;
 
         Ok(())
     }
 
+    /// Shutdown the monitor node.
+    ///
+    /// Unsubscribes from task topics, closes channels.
     pub async fn shutdown(&mut self) -> Result<()> {
         log::info!("Shutting down monitor");
         self.p2p.unsubscribe(WorkflowHandler::LISTEN_TOPIC).await?;
@@ -62,9 +70,16 @@ impl DriaMonitorNode {
         Ok(())
     }
 
+    /// Run the monitor node.
     pub async fn run(&mut self, token: CancellationToken) {
         let mut task_print_interval =
             tokio::time::interval(tokio::time::Duration::from_secs(TASK_PRINT_INTERVAL_SECS));
+        let mut peer_print_interval =
+            tokio::time::interval(tokio::time::Duration::from_secs(PEER_PRINT_INTERVAL_SECS));
+
+        // move one ticks
+        task_print_interval.tick().await;
+        peer_print_interval.tick().await;
 
         loop {
             tokio::select! {
@@ -76,13 +91,27 @@ impl DriaMonitorNode {
                     }
                     None => break, // channel closed, we can return now
                 },
-                // print task counts
                 _ = task_print_interval.tick() => self.handle_task_print(),
+                _ = peer_print_interval.tick() => self.handle_peer_print().await,
                 _ = token.cancelled() => break,
             }
         }
     }
 
+    async fn handle_peer_print(&self) {
+        match self.p2p.peer_counts().await {
+            Ok((mesh, all)) => {
+                log::info!("Peer count: {} / {}", mesh, all);
+            }
+            Err(e) => {
+                log::error!("Error getting peer counts: {:?}", e);
+            }
+        }
+    }
+
+    /// Handle incoming gossipsub message.
+    ///
+    /// Records the `task` and `result` messages only, does not respond to anything else.
     async fn handle_message(
         &mut self,
         (peer_id, message_id, gossipsub_message): (PeerId, MessageId, Message),
@@ -120,6 +149,7 @@ impl DriaMonitorNode {
         Ok(())
     }
 
+    /// Print the tasks (ids) that have not been responded to.
     fn handle_task_print(&self) {
         let seen_task_ids = self.tasks.keys().collect::<Vec<_>>();
         let seen_result_ids = self.results.keys().collect::<Vec<_>>();