added "connecting" log and pings per sec

Author: erhant

Cargo.lock

@@ -7,7 +7,7 @@ default-members = ["compute"]
 
 [workspace.package]
 edition = "2021"
-version = "0.3.8"
+version = "0.3.9"
 license = "Apache-2.0"
 readme = "README.md"
 

Cargo.toml

@@ -14,6 +14,11 @@ debug:
 build:
 		cargo build --workspace
 
+.PHONY: trace #        | Run with TRACE logs
+trace:
+		RUST_LOG=warn,dkn_compute=trace,libp2p=debug \
+		cargo run --bin dkn-compute
+
 .PHONY: profile-cpu #  | Profile CPU usage with flamegraph
 profile-cpu:
 	  DKN_EXIT_TIMEOUT=120 cargo flamegraph --root --profile=profiling --bin dkn-compute

Makefile

@@ -65,8 +65,9 @@ impl PingpongHandler {
 
         log::info!("Received a {} for: {}", "ping".blue(), pingpong.uuid);
 
-        // record ping moment
+        // record ping
         node.last_pinged_at = Instant::now();
+        node.num_pings += 1;
 
         // respond
         let response_body = PingpongResponse {

compute/src/gossipsub/pingpong.rs

@@ -7,7 +7,7 @@ use workers::task::TaskWorker;
 
 #[tokio::main]
 async fn main() -> Result<()> {
-    // load a particular env, or `.env` by default
+    // load a particular environment file specified by DKN_COMPUTE_ENV, or `.env` by default
     let env_path = env::var("DKN_COMPUTE_ENV").unwrap_or_else(|_| ".env".to_string());
     let dotenv_result = dotenvy::from_path(&env_path);
 

compute/src/main.rs

@@ -22,10 +22,6 @@ impl DriaComputeNode {
     pub(crate) async fn handle_diagnostic_refresh(&self) {
         let mut diagnostics = vec![format!("Diagnostics (v{}):", DRIA_COMPUTE_NODE_VERSION)];
 
-        // if we have not received pings for a while, we are considered offline
-        let is_offline =
-            self.last_pinged_at < Instant::now() - Duration::from_secs(PING_LIVENESS_SECS);
-
         // print peer counts
         match self.p2p.peer_counts().await {
             Ok((mesh, all)) => diagnostics.push(format!(
@@ -74,14 +70,27 @@ impl DriaComputeNode {
         ));
 
         // add network status as well
-        diagnostics.push(format!(
-            "Node Status: {}",
-            if is_offline {
-                "OFFLINE".bold().red()
-            } else {
-                "ONLINE".bold().green()
-            }
-        ));
+        // if we have not received pings for a while, we are considered offline
+        let is_offline = Instant::now().duration_since(self.last_pinged_at)
+            > Duration::from_secs(PING_LIVENESS_SECS);
+        if self.num_pings == 0 {
+            // if we didnt have any pings, we might still be connecting
+            diagnostics.push(format!("Node Status: {}", "CONNECTING".yellow()));
+        } else {
+            diagnostics.push(format!(
+                "Node Status: {}",
+                if is_offline {
+                    "OFFLINE".red()
+                } else {
+                    "ONLINE".green()
+                }
+            ));
+        }
+
+        // add pings per second
+        let elapsed = Instant::now().duration_since(self.started_at).as_secs_f64();
+        let pings_per_second = self.num_pings as f64 / elapsed; // elapsed is always > 0
+        diagnostics.push(format!("Pings/sec: {:.3}", pings_per_second));
 
         log::info!("{}", diagnostics.join("\n  "));
 

compute/src/node/diagnostic.rs

@@ -33,7 +33,11 @@ pub struct DriaComputeNode {
     pub p2p: DriaP2PCommander,
     /// The last time the node was pinged by the network.
     /// If this is too much, we can say that the node is not reachable by RPC.
-    pub last_pinged_at: Instant,
+    pub(crate) last_pinged_at: Instant,
+    /// Number of pings received.
+    pub(crate) num_pings: u64,
+    /// The time the node was started.
+    pub(crate) started_at: Instant,
     /// Gossipsub message receiver, used by peer-to-peer client in a separate thread.
     ///
     /// It will publish messages sent to this channel to the network.
@@ -143,6 +147,8 @@ impl DriaComputeNode {
                 initial_steps,
                 spec_collector: SpecCollector::new(model_names),
                 last_pinged_at: Instant::now(),
+                num_pings: 0,
+                started_at: Instant::now(),
             },
             p2p_client,
             task_batch_worker,

compute/src/node/mod.rs

@@ -183,6 +183,7 @@ fn create_gossipsub_behaviour(author: PeerId) -> Result<gossipsub::Behaviour> {
         MessageId::from(digest.to_be_bytes())
     };
 
+    // TODO: make `flood_publish false?`
     Behaviour::new(
         MessageAuthenticity::Author(author),
         ConfigBuilder::default()

p2p/src/behaviour.rs

@@ -48,6 +48,8 @@ impl DriaP2PClient {
     ///
     /// The `version` is used to create the protocol strings for the client, and its very important that
     /// they match with the clients existing within the network.
+    ///
+    /// If for any reason the given `listen_addr` is not available, it will try to listen on a random port on `localhost`.
     #[allow(clippy::type_complexity)]
     pub fn new(
         keypair: Keypair,
@@ -62,7 +64,6 @@ impl DriaP2PClient {
     )> {
         // this is our peerId
         let peer_id = keypair.public().to_peer_id();
-        log::info!("Compute node peer address: {}", peer_id);
 
         let mut swarm = SwarmBuilder::with_existing_identity(keypair)
             .with_tokio()
@@ -123,19 +124,23 @@ impl DriaP2PClient {
 
         // listen on all interfaces for incoming connections
         log::info!("Listening p2p network on: {}", listen_addr);
-        swarm.listen_on(listen_addr)?;
+        if let Err(e) = swarm.listen_on(listen_addr) {
+            log::error!("Could not listen on address: {:?}", e);
+            log::warn!("Trying fallback address with localhost random port");
+            swarm.listen_on("/ip4/127.0.0.1/tcp/0".parse().unwrap())?;
+        }
 
         // listen on relay addresses with p2p circuit
-        for addr in &nodes.relay_nodes {
+        for addr in nodes.relay_nodes.iter().cloned() {
             log::info!("Listening to relay: {}", addr);
-            swarm.listen_on(addr.clone().with(Protocol::P2pCircuit))?;
+            swarm.listen_on(addr.with(Protocol::P2pCircuit))?;
         }
 
         // dial rpc nodes
-        for rpc_addr in &nodes.rpc_nodes {
+        for rpc_addr in nodes.rpc_nodes.iter().cloned() {
             log::info!("Dialing RPC node: {}", rpc_addr);
-            if let Err(e) = swarm.dial(rpc_addr.clone()) {
-                log::error!("Error dialing RPC node: {:?}", e);
+            if let Err(e) = swarm.dial(rpc_addr) {
+                log::error!("Could not dial RPC node: {:?}", e);
             };
         }