p2pnet: peer discovery for outbound slots implemented

src/net/hosts.rs

@@ -3,22 +3,44 @@ use async_std::sync::{Arc, Mutex};
 use fxhash::FxHashSet;
 use url::Url;
 
+const LOCALNET: [&str; 5] = ["localhost", "0.0.0.0", "[::]", "127.0.0.1", "[::1]"];
+
 /// Pointer to hosts class.
 pub type HostsPtr = Arc<Hosts>;
 
 /// Manages a store of network addresses.
 pub struct Hosts {
     addrs: Mutex<FxHashSet<Url>>,
+    localnet: bool,
 }
 
 impl Hosts {
     /// Create a new host list.
-    pub fn new() -> Arc<Self> {
-        Arc::new(Self { addrs: Mutex::new(FxHashSet::default()) })
+    pub fn new(localnet: bool) -> Arc<Self> {
+        Arc::new(Self { addrs: Mutex::new(FxHashSet::default()), localnet })
     }
 
-    /// Add a new host to the host list.
-    pub async fn store(&self, addrs: Vec<Url>) {
+    /// Add a new host to the host list, after filtering localnet hosts,
+    /// if configured to do so.
+    pub async fn store(&self, input_addrs: Vec<Url>) {
+        let addrs = if !self.localnet {
+            let mut filtered = vec![];
+            for addr in &input_addrs {
+                match addr.host_str() {
+                    Some(host_str) => {
+                        if LOCALNET.contains(&host_str) {
+                            continue
+                        }
+                    }
+                    None => continue,
+                }
+                filtered.push(addr.clone());
+            }
+            filtered
+        } else {
+            input_addrs
+        };
+
         for addr in addrs {
             self.addrs.lock().await.insert(addr);
         }

src/net/p2p.rs

@@ -5,6 +5,7 @@ use async_executor::Executor;
 use futures::{select, try_join, FutureExt};
 use fxhash::{FxHashMap, FxHashSet};
 use log::{debug, error, warn};
+use rand::Rng;
 use serde_json::json;
 use url::Url;
 
@@ -72,6 +73,9 @@ pub struct P2p {
     state: Mutex<P2pState>,
 
     settings: SettingsPtr,
+
+    /// Flag to check if on discovery mode
+    discovery: Mutex<bool>,
 }
 
 impl P2p {
@@ -90,13 +94,14 @@ impl P2p {
             channels: Mutex::new(FxHashMap::default()),
             channel_subscriber: Subscriber::new(),
             stop_subscriber: Subscriber::new(),
-            hosts: Hosts::new(),
+            hosts: Hosts::new(settings.localnet),
             protocol_registry: ProtocolRegistry::new(),
             session_manual: Mutex::new(None),
             session_inbound: Mutex::new(None),
             session_outbound: Mutex::new(None),
             state: Mutex::new(P2pState::Open),
             settings,
+            discovery: Mutex::new(false),
         });
 
         let parent = Arc::downgrade(&self_);
@@ -405,4 +410,31 @@ impl P2p {
     pub fn channels(&self) -> &ConnectedChannels {
         &self.channels
     }
+
+    /// Try to start discovery mode.
+    /// Returns false if already on discovery mode.
+    pub async fn start_discovery(self: Arc<Self>) -> bool {
+        if *self.discovery.lock().await {
+            return false
+        }
+        *self.discovery.lock().await = true;
+        true
+    }
+
+    /// Stops discovery mode.
+    pub async fn stop_discovery(self: Arc<Self>) {
+        *self.discovery.lock().await = false;
+    }
+
+    /// Retrieves a random connected channel
+    pub async fn random_channel(self: Arc<Self>) -> Option<Arc<Channel>> {
+        let channels_map = self.channels().lock().await;
+        let mut values = channels_map.values();
+
+        if values.len() == 0 {
+            return None
+        }
+
+        Some(values.nth(rand::thread_rng().gen_range(0..values.len())).unwrap().clone())
+    }
 }

src/net/protocol/protocol_address.rs

@@ -15,7 +15,6 @@ use super::{
 };
 
 const SEND_ADDR_SLEEP_SECONDS: u64 = 900;
-const LOCALNET: [&str; 5] = ["localhost", "0.0.0.0", "[::]", "127.0.0.1", "[::1]"];
 
 /// Defines address and get-address messages.
 pub struct ProtocolAddress {
@@ -60,35 +59,13 @@ impl ProtocolAddress {
 
     /// Handles receiving the address message. Loops to continually recieve
     /// address messages on the address subsciption. Adds the recieved
-    /// addresses to the list of hosts, after filtering localnet hosts,
-    /// if configured to do so.
+    /// addresses to the list of hosts.
     async fn handle_receive_addrs(self: Arc<Self>) -> Result<()> {
         debug!(target: "net", "ProtocolAddress::handle_receive_addrs() [START]");
         loop {
             let addrs_msg = self.addrs_sub.receive().await?;
             debug!(target: "net", "ProtocolAddress::handle_receive_addrs() received {} addrs", addrs_msg.addrs.len());
-            let addrs = if !self.settings.localnet {
-                let mut filtered = vec![];
-                for addr in &addrs_msg.addrs {
-                    match addr.host_str() {
-                        Some(host_str) => {
-                            if LOCALNET.contains(&host_str) {
-                                debug!(target: "net", "  localnet host({}) detected, ignoring", host_str);
-                                continue
-                            }
-                        }
-                        None => {
-                            debug!(target: "net", "  empty host({}) detected, ignoring...", addr);
-                            continue
-                        }
-                    }
-                    filtered.push(addr.clone());
-                }
-                filtered
-            } else {
-                addrs_msg.addrs.clone()
-            };
-            self.hosts.store(addrs).await;
+            self.hosts.store(addrs_msg.addrs.clone()).await;
         }
     }
 

src/net/session/outbound_session.rs

@@ -9,7 +9,7 @@ use serde_json::{json, Value};
 use url::Url;
 
 use crate::{
-    net::TransportName,
+    net::{message, TransportName},
     system::{StoppableTask, StoppableTaskPtr, Subscriber, SubscriberPtr, Subscription},
     util::async_util,
     Error, Result,
@@ -249,8 +249,8 @@ impl OutboundSession {
     /// Loops through host addresses to find a outbound address that we can
     /// connect to. Checks whether address is valid by making sure it isn't
     /// our own inbound address, then checks whether it is already connected
-    /// (exists) or connecting (pending). Keeps looping until address is
-    /// found that passes all checks.
+    /// (exists) or connecting (pending). If no address was found, we try to
+    /// to discover new peers. Keeps looping until address is found that passes all checks.
     async fn load_address(&self, slot_number: u32) -> Result<Url> {
         loop {
             let p2p = self.p2p();
@@ -287,12 +287,60 @@ impl OutboundSession {
                 return Ok(addr)
             }
 
-            debug!(target: "net", "Hosts address pool is empty. Retrying connect slot #{}", slot_number);
+            // Peer discovery
+            if p2p.settings().peer_discovery {
+                debug!(target: "net", "#{} No available address found, entering peer discovery mode.", slot_number);
+                self.peer_discovery(slot_number).await?;
+                debug!(target: "net", "#{} Discovery mode ended.", slot_number);
+            }
 
+            // Sleep and then retry
+            debug!(target: "net", "Retrying connect slot #{}", slot_number);
             async_util::sleep(p2p.settings().outbound_retry_seconds).await;
         }
     }
 
+    /// Try to find new peers to update available hosts.
+    async fn peer_discovery(&self, slot_number: u32) -> Result<()> {
+        // Check that another slot(thread) already tries to update hosts
+        let p2p = self.p2p();
+        if !p2p.clone().start_discovery().await {
+            debug!(target: "net", "#{} P2P already on discovery mode.", slot_number);
+            return Ok(())
+        }
+
+        debug!(target: "net", "#{} Discovery mode started.", slot_number);
+
+        // Getting a random connected channel to ask for peers
+        let channel = match p2p.clone().random_channel().await {
+            Some(c) => c,
+            None => {
+                debug!(target: "net", "#{} No peers found.", slot_number);
+                p2p.clone().stop_discovery().await;
+                return Ok(())
+            }
+        };
+
+        // Ask peer
+        debug!(target: "net", "#{} Asking peer: {}", slot_number, channel.address());
+
+        // Communication setup
+        let response_sub = channel.subscribe_msg::<message::AddrsMessage>().await?;
+        let get_addr_msg = message::GetAddrsMessage {};
+
+        // Executing request and waiting for response
+        channel.send(get_addr_msg).await?;
+        let resp = response_sub.receive().await?;
+
+        // Store response data
+        debug!(target: "net", "#{} response: {:?}", slot_number, resp.addrs);
+        p2p.hosts().store(resp.addrs.clone()).await;
+
+        p2p.stop_discovery().await;
+
+        Ok(())
+    }
+
     /// Subscribe to a channel.
     pub async fn subscribe_channel(&self) -> Subscription<Result<ChannelPtr>> {
         self.channel_subscriber.clone().subscribe().await

src/net/settings.rs

@@ -28,6 +28,7 @@ pub struct Settings {
     pub app_version: Option<String>,
     pub outbound_transports: Vec<TransportName>,
     pub localnet: bool,
+    pub peer_discovery: bool,
 }
 
 impl Default for Settings {
@@ -48,6 +49,7 @@ impl Default for Settings {
             app_version: Some(option_env!("CARGO_PKG_VERSION").unwrap_or("").to_string()),
             outbound_transports: get_outbound_transports(vec![]),
             localnet: false,
+            peer_discovery: true,
         }
     }
 }
@@ -110,6 +112,11 @@ pub struct SettingsOpt {
     #[serde(default)]
     #[structopt(long)]
     pub localnet: bool,
+
+    /// Enable peer discovery
+    #[serde(default = "default_as_true")]
+    #[structopt(long)]
+    pub peer_discovery: bool,
 }
 
 impl From<SettingsOpt> for Settings {
@@ -130,6 +137,7 @@ impl From<SettingsOpt> for Settings {
             app_version: settings_opt.app_version,
             outbound_transports: get_outbound_transports(settings_opt.outbound_transports),
             localnet: settings_opt.localnet,
+            peer_discovery: settings_opt.peer_discovery,
         }
     }
 }
@@ -152,3 +160,8 @@ pub fn get_outbound_transports(opt_outbound_transports: Vec<String>) -> Vec<Tran
 
     outbound_transports
 }
+
+/// Auxiliary function to set serde bool value to true.
+fn default_as_true() -> bool {
+    true
+}