p2pnet/wait_for_outbound: connection timeout added

bin/darkfid/src/main.rs

@@ -406,7 +406,7 @@ async fn realmain(args: Args, ex: Arc<Executor<'_>>) -> Result<()> {
     .detach();
 
     info!("Waiting for sync P2P outbound connections");
-    sync_p2p.clone().unwrap().wait_for_outbound().await?;
+    sync_p2p.clone().unwrap().wait_for_outbound(ex.clone()).await?;
 
     match block_sync_task(sync_p2p.clone().unwrap(), state.clone()).await {
         Ok(()) => *darkfid.synced.lock().await = true,
@@ -427,7 +427,7 @@ async fn realmain(args: Args, ex: Arc<Executor<'_>>) -> Result<()> {
         .detach();
 
         info!("Waiting for consensus P2P outbound connections");
-        consensus_p2p.clone().unwrap().wait_for_outbound().await?;
+        consensus_p2p.clone().unwrap().wait_for_outbound(ex.clone()).await?;
 
         info!("Starting consensus protocol task");
         ex.spawn(proposal_task(consensus_p2p.unwrap(), sync_p2p.unwrap(), state)).detach();

bin/darkwikid/src/main.rs

@@ -575,7 +575,7 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
 
     executor.spawn(p2p.clone().run(executor.clone())).detach();
 
-    p2p.clone().wait_for_outbound().await?;
+    p2p.clone().wait_for_outbound(executor.clone()).await?;
 
     //
     // Darkwiki start

bin/faucetd/src/main.rs

@@ -410,7 +410,7 @@ async fn realmain(args: Args, ex: Arc<Executor<'_>>) -> Result<()> {
     .detach();
 
     info!("Waiting for sync P2P outbound connections");
-    sync_p2p.clone().wait_for_outbound().await?;
+    sync_p2p.clone().wait_for_outbound(ex).await?;
 
     match block_sync_task(sync_p2p, state.clone()).await {
         Ok(()) => *faucetd.synced.lock().await = true,

bin/tau/taud/src/main.rs

@@ -221,7 +221,7 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
 
     executor.spawn(p2p.clone().run(executor.clone())).detach();
 
-    p2p.clone().wait_for_outbound().await?;
+    p2p.clone().wait_for_outbound(executor.clone()).await?;
 
     //
     // RPC interface

script/research/fud/fud/src/main.rs

@@ -407,7 +407,7 @@ async fn realmain(args: Args, ex: Arc<Executor<'_>>) -> Result<()> {
     .detach();
 
     info!("Waiting for P2P outbound connections");
-    p2p.wait_for_outbound().await?;
+    p2p.wait_for_outbound(ex).await?;
 
     fud.init().await?;
 

src/dht/dht.rs

@@ -5,7 +5,7 @@ use futures::{select, FutureExt};
 use fxhash::FxHashMap;
 use log::{debug, error, warn};
 use rand::Rng;
-use std::{collections::HashSet, time::Duration};
+use std::collections::HashSet;
 
 use crate::{
     net,
@@ -21,7 +21,6 @@ use super::{
 };
 
 // Constants configuration
-const REQUEST_TIMEOUT: u64 = 2400;
 const SEEN_DURATION: i64 = 120;
 
 /// Atomic pointer to DHT state
@@ -265,14 +264,18 @@ impl Dht {
 
 // Auxilary function to wait for a key response from the P2P network.
 pub async fn waiting_for_response(dht: DhtPtr) -> Result<Option<KeyResponse>> {
-    let (p2p_recv_channel, stop_signal) = {
+    let (p2p_recv_channel, stop_signal, timeout) = {
         let _dht = dht.read().await;
-        (_dht.p2p_recv_channel.clone(), _dht.stop_signal.clone())
+        (
+            _dht.p2p_recv_channel.clone(),
+            _dht.stop_signal.clone(),
+            _dht.p2p.settings().connect_timeout_seconds as u64,
+        )
     };
     let ex = Arc::new(async_executor::Executor::new());
     let (timeout_s, timeout_r) = async_channel::unbounded::<()>();
     ex.spawn(async move {
-        sleep(Duration::from_millis(REQUEST_TIMEOUT).as_secs()).await;
+        sleep(timeout).await;
         timeout_s.send(()).await.unwrap_or(());
     })
     .detach();

src/net/p2p.rs

@@ -2,13 +2,15 @@ use async_std::sync::{Arc, Mutex};
 use std::fmt;
 
 use async_executor::Executor;
+use futures::{select, FutureExt};
 use fxhash::{FxHashMap, FxHashSet};
-use log::debug;
+use log::{debug, warn};
 use serde_json::json;
 use url::Url;
 
 use crate::{
     system::{Subscriber, SubscriberPtr, Subscription},
+    util::sleep,
     Result,
 };
 
@@ -183,7 +185,7 @@ impl P2p {
     }
 
     /// Wait for outbound connections to be established.
-    pub async fn wait_for_outbound(self: Arc<Self>) -> Result<()> {
+    pub async fn wait_for_outbound(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
         debug!(target: "net", "P2p::wait_for_outbound() [BEGIN]");
         // To verify that the network needs initialization, we check if we have seeds or peers configured,
         // and have configured outbound slots.
@@ -202,18 +204,44 @@ impl P2p {
             let outbound_sub =
                 self.session_outbound.lock().await.as_ref().unwrap().subscribe_channel().await;
 
+            // Retrieve sto subscriber
+            let stop_sub = self.subscribe_stop().await;
+
+            // Retrieve timeout config
+            let timeout = self.settings().connect_timeout_seconds as u64;
+
             // Wait for the result for each of the addresses, excluding our own inbound addresses
             for addr in addrs {
                 if self_inbound_addr.contains(&addr) {
                     continue
                 }
 
-                // Wait for address to be processed
-                if let Err(e) = outbound_sub.receive().await {
-                    debug!(
-                        "P2p::wait_for_outbound(): Outbound connection failed [{}]: {}",
-                        &addr, e
-                    );
+                // Wait for address to be processed.
+                // We use a timeout to eliminate the following cases:
+                //  1. Network timeout
+                //  2. Thread reaching the receiver after peer has signal it
+                let (timeout_s, timeout_r) = async_channel::unbounded::<()>();
+                executor
+                    .spawn(async move {
+                        sleep(timeout).await;
+                        timeout_s.send(()).await.unwrap_or(());
+                    })
+                    .detach();
+
+                select! {
+                    msg = outbound_sub.receive().fuse() => {
+                            if let Err(e) = msg {
+                                warn!(
+                                    "P2p::wait_for_outbound(): Outbound connection failed [{}]: {}",
+                                    &addr, e
+                                );
+                            }
+                    },
+                    _ = stop_sub.receive().fuse() => debug!("P2p::wait_for_outbound(): stop signal received!"),
+                    _ = timeout_r.recv().fuse() => {
+                        warn!("P2p::wait_for_outbound(): Timeout on outbound connection: {}", &addr);
+                        continue
+                    },
                 }
             }