net: invoke GreylistRefinery in p2p.rs and cleanup

src/net/hosts/refinery.rs

@@ -22,34 +22,36 @@ use log::{debug, warn};
 use rand::Rng;
 use url::Url;
 
-use super::super::p2p::P2pPtr;
+use super::super::p2p::{P2p, P2pPtr};
 use crate::{
     net::{
         connector::Connector,
         protocol::ProtocolVersion,
         session::{Session, SessionWeakPtr},
     },
-    system::{sleep, StoppableTask, StoppableTaskPtr},
+    system::{sleep, LazyWeak, StoppableTask, StoppableTaskPtr},
     Error,
 };
 
+pub type GreylistRefineryPtr = Arc<GreylistRefinery>;
+
 //// Probe random peers on the greylist. If a peer is responsive, update the last_seen field and
 //// add it to the whitelist. If a node does not respond, remove it from the greylist.
 //// Called periodically.
 // NOTE: in monero this is called "greylist housekeeping" but that's a bit verbose.
-struct GreylistRefinery {
-    p2p: P2pPtr,
+pub struct GreylistRefinery {
+    /// Weak pointer to parent p2p object
+    pub(in crate::net) p2p: LazyWeak<P2p>,
     process: StoppableTaskPtr,
-    session: SessionWeakPtr,
 }
 
 impl GreylistRefinery {
-    fn new(p2p: P2pPtr, session: SessionWeakPtr) -> Arc<Self> {
-        Arc::new(Self { p2p, process: StoppableTask::new(), session })
+    pub fn new() -> Arc<Self> {
+        Arc::new(Self { p2p: LazyWeak::new(), process: StoppableTask::new() })
     }
 
-    async fn start(self: Arc<Self>) {
-        let ex = self.p2p.executor();
+    pub async fn start(self: Arc<Self>) {
+        let ex = self.p2p().executor();
         self.process.clone().start(
             async move {
                 self.run().await;
@@ -62,7 +64,7 @@ impl GreylistRefinery {
         );
     }
 
-    async fn stop(self: Arc<Self>) {
+    pub async fn stop(self: Arc<Self>) {
         self.process.stop().await
     }
 
@@ -71,11 +73,12 @@ impl GreylistRefinery {
     async fn run(self: Arc<Self>) {
         debug!(target: "net::refinery::run()", "START");
         loop {
-            let hosts = self.p2p.hosts();
+            let hosts = self.p2p().hosts();
 
             if hosts.is_empty_greylist().await {
-                warn!(target: "net::refinery::run()", "Greylist is empty. Aborting");
-                break
+                warn!(target: "net::refinery::run()",
+                "Greylist is empty! Cannot start refinery process. Sleeping...");
+                sleep(5).await;
             } else {
                 debug!(target: "net::refinery::run()", "Starting refinery process");
                 // Randomly select an entry from the greylist.
@@ -84,7 +87,7 @@ impl GreylistRefinery {
                 let entry = &greylist[position];
                 let url = &entry.0;
 
-                if ping_node(url, self.p2p.clone(), self.session.clone()).await {
+                if ping_node(url, self.p2p().clone()).await {
                     let whitelist = hosts.whitelist.read().await;
                     // Remove oldest element if the whitelist reaches max size.
                     if whitelist.len() == 1000 {
@@ -119,16 +122,21 @@ impl GreylistRefinery {
 
             // TODO: create a custom net setting for this timer
             debug!(target: "net::greylist_refinery::run()", "Sleeping...");
-            sleep(self.p2p.settings().outbound_peer_discovery_attempt_time).await;
+            sleep(5).await;
         }
     }
+
+    fn p2p(&self) -> P2pPtr {
+        self.p2p.upgrade()
+    }
 }
 
 // Ping a node to check it's online.
 // TODO: make this an actual ping-pong method, rather than a version exchange.
-pub async fn ping_node(addr: &Url, p2p: P2pPtr, session: SessionWeakPtr) -> bool {
-    let connector = Connector::new(p2p.settings(), session.clone());
-    let outbound_session = p2p.session_outbound();
+pub async fn ping_node(addr: &Url, p2p: P2pPtr) -> bool {
+    let session_outbound = p2p.session_outbound();
+    let parent = Arc::downgrade(&session_outbound);
+    let connector = Connector::new(p2p.settings(), parent);
 
     debug!(target: "net::refinery::ping_node()", "Attempting to connect to {}", addr);
     match connector.connect(addr).await {
@@ -136,7 +144,7 @@ pub async fn ping_node(addr: &Url, p2p: P2pPtr, session: SessionWeakPtr) -> bool
             debug!(target: "net::refinery::ping_node()", "Connected successfully!");
             let proto_ver = ProtocolVersion::new(channel.clone(), p2p.settings()).await;
 
-            let handshake_task = outbound_session.perform_handshake_protocols(
+            let handshake_task = session_outbound.perform_handshake_protocols(
                 proto_ver,
                 channel.clone(),
                 p2p.executor(),

src/net/hosts/store.rs

@@ -18,7 +18,7 @@
 
 use std::{collections::HashSet, sync::Arc};
 
-use log::{debug, trace};
+use log::{debug, trace, warn};
 use rand::{
     prelude::{IteratorRandom, SliceRandom},
     rngs::OsRng,
@@ -525,6 +525,8 @@ impl Hosts {
         // Retrieve all peers corresponding to that transport schemes
         let hosts = self.whitelist_fetch_with_schemes(schemes, None).await;
         if hosts.is_empty() {
+            warn!(target: "store::whitelist_fetch_n_random_with_schemes",
+                  "Whitelist is empty! Exiting...");
             return hosts
         }
 
@@ -547,6 +549,8 @@ impl Hosts {
         // Retrieve all peers not corresponding to that transport schemes
         let hosts = self.whitelist_fetch_excluding_schemes(schemes, None).await;
         if hosts.is_empty() {
+            warn!(target: "store::whitelist_fetch_n_random_excluding_schemes",
+                  "Whitelist is empty! Exiting...");
             return hosts
         }
 
@@ -646,7 +650,7 @@ impl Hosts {
 
 #[cfg(test)]
 mod tests {
-    use super::{super::settings::Settings, *};
+    use super::{super::super::settings::Settings, *};
     use std::time::UNIX_EPOCH;
 
     #[test]

src/net/p2p.rs

@@ -30,7 +30,10 @@ use url::Url;
 use super::{
     channel::ChannelPtr,
     dnet::DnetEvent,
-    hosts::store::{Hosts, HostsPtr},
+    hosts::{
+        refinery::{GreylistRefinery, GreylistRefineryPtr},
+        store::{Hosts, HostsPtr},
+    },
     message::Message,
     protocol::{protocol_registry::ProtocolRegistry, register_default_protocols},
     session::{
@@ -81,6 +84,9 @@ pub struct P2p {
     pub dnet_enabled: Mutex<bool>,
     /// The subscriber for which we can give dnet info over
     dnet_subscriber: SubscriberPtr<DnetEvent>,
+
+    // Greylist refinery process
+    greylist_refinery: Arc<GreylistRefinery>,
 }
 
 impl P2p {
@@ -111,12 +117,15 @@ impl P2p {
 
             dnet_enabled: Mutex::new(false),
             dnet_subscriber: Subscriber::new(),
+
+            greylist_refinery: GreylistRefinery::new(),
         });
 
         self_.session_manual.p2p.init(self_.clone());
         self_.session_inbound.p2p.init(self_.clone());
         self_.session_outbound.p2p.init(self_.clone());
 
+        self_.greylist_refinery.p2p.init(self_.clone());
         register_default_protocols(self_.clone()).await;
 
         self_
@@ -142,6 +151,9 @@ impl P2p {
         // Start the outbound session
         self.session_outbound().start().await;
 
+        info!(target: "net::p2p::start()", "Starting greylist refinery process");
+        self.greylist_refinery.clone().start().await;
+
         info!(target: "net::p2p::start()", "[P2P] P2P subsystem started");
         Ok(())
     }
@@ -166,6 +178,9 @@ impl P2p {
         self.session_manual().stop().await;
         self.session_inbound().stop().await;
         self.session_outbound().stop().await;
+
+        // Stop greylist refinery process
+        self.greylist_refinery().stop().await;
     }
 
     /// Broadcasts a message concurrently across all active channels.
@@ -289,6 +304,11 @@ impl P2p {
         self.session_outbound.clone()
     }
 
+    /// Get pointer to greylist refinery
+    pub fn greylist_refinery(&self) -> GreylistRefineryPtr {
+        self.greylist_refinery.clone()
+    }
+
     /// Enable network debugging
     pub async fn dnet_enable(&self) {
         *self.dnet_enabled.lock().await = true;

src/net/protocol/protocol_address.rs

@@ -49,10 +49,9 @@ pub struct ProtocolAddress {
     get_addrs_sub: MessageSubscription<GetAddrsMessage>,
     hosts: HostsPtr,
     settings: SettingsPtr,
+    jobsman: ProtocolJobsManagerPtr,
     // We require this to access ping_self() method.
     p2p: P2pPtr,
-    session: OutboundSessionPtr,
-    jobsman: ProtocolJobsManagerPtr,
 }
 
 const PROTO_NAME: &str = "ProtocolAddress";
@@ -64,7 +63,6 @@ impl ProtocolAddress {
     pub async fn init(channel: ChannelPtr, p2p: P2pPtr) -> ProtocolBasePtr {
         let settings = p2p.settings();
         let hosts = p2p.hosts();
-        let session = p2p.session_outbound();
 
         // Creates a subscription to address message
         let addrs_sub =
@@ -80,9 +78,8 @@ impl ProtocolAddress {
             get_addrs_sub,
             hosts,
             jobsman: ProtocolJobsManager::new(PROTO_NAME, channel),
-            session,
-            p2p,
             settings,
+            p2p,
         })
     }
 
@@ -197,9 +194,8 @@ impl ProtocolAddress {
         for addr in self.settings.external_addrs.clone() {
             debug!(target: "net::protocol_address::send_my_addrs()", "Attempting to ping self");
 
-            let parent = Arc::downgrade(&self.session);
             // See if we can do a version exchange with ourself.
-            if ping_node(&addr, self.p2p.clone(), parent).await {
+            if ping_node(&addr, self.p2p.clone()).await {
                 // We're online. Update last_seen and broadcast our address.
                 let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
                 addrs.push((addr, last_seen));
@@ -208,25 +204,12 @@ impl ProtocolAddress {
                 return Ok(())
             }
         }
-        //// See if we can do a version exchange with ourself.
         debug!(target: "net::protocol_address::send_my_addrs()", "Broadcasting address");
         let ext_addr_msg = AddrsMessage { addrs };
         self.channel.send(&ext_addr_msg).await?;
         debug!(target: "net::protocol_address::send_my_addrs()", "[END]");
 
         Ok(())
-        //debug!(target: "net::protocol_address", "Attempting to ping self");
-        //if self.session.ping_node(self.channel.address()).await {
-        //    // We're online. Broadcast our address.
-        //    let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
-        //    let addrs = vec![(self.channel.address().clone(), last_seen)];
-        //    let addrs_msg = AddrsMessage { addrs };
-        //    self.channel.send(&addrs_msg).await?;
-        //} else {
-        //    debug!(target: "net::protocol_address", "Ping self failed");
-        //}
-
-        //Ok(())
     }
 }
 
@@ -242,6 +225,7 @@ impl ProtocolBase for ProtocolAddress {
         self.jobsman.clone().spawn(self.clone().send_my_addrs(), ex.clone()).await;
 
         self.jobsman.clone().spawn(self.clone().handle_receive_addrs(), ex.clone()).await;
+
         self.jobsman.spawn(self.clone().handle_receive_get_addrs(), ex).await;
 
         // Send get_address message.

src/net/protocol/protocol_seed.rs

@@ -43,7 +43,6 @@ pub struct ProtocolSeed {
     settings: SettingsPtr,
     addr_sub: MessageSubscription<AddrsMessage>,
     // We require this to access ping_self() method.
-    session: OutboundSessionPtr,
     p2p: P2pPtr,
 }
 
@@ -60,7 +59,7 @@ impl ProtocolSeed {
         let addr_sub =
             channel.subscribe_msg::<AddrsMessage>().await.expect("Missing addr dispatcher!");
 
-        Arc::new(Self { channel, hosts, settings, addr_sub, session, p2p })
+        Arc::new(Self { channel, hosts, settings, addr_sub, p2p })
     }
 
     /// Sends own external addresses over a channel. Imports own external addresses
@@ -87,9 +86,7 @@ impl ProtocolSeed {
             debug!(target: "net::protocol_seed::send_my_addrs()", "Attempting to ping self");
 
             // See if we can do a version exchange with ourself.
-            let parent = Arc::downgrade(&self.session);
-            // See if we can do a version exchange with ourself.
-            if ping_node(&addr, self.p2p.clone(), parent).await {
+            if ping_node(&addr, self.p2p.clone()).await {
                 // We're online. Update last_seen and broadcast our address.
                 let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
                 addrs.push((addr, last_seen));

src/net/session/outbound_session.rs

@@ -70,8 +70,6 @@ pub struct OutboundSession {
     slots: Mutex<Vec<Arc<Slot>>>,
     /// Peer discovery task
     peer_discovery: Arc<PeerDiscovery>,
-    ///// Greylist refinery task
-    //greylist_refinery: Arc<GreylistRefinery>,
 }
 
 impl OutboundSession {
@@ -82,10 +80,8 @@ impl OutboundSession {
             channel_subscriber: Subscriber::new(),
             slots: Mutex::new(Vec::new()),
             peer_discovery: PeerDiscovery::new(),
-            //greylist_refinery: GreylistRefinery::new(),
         });
         self_.peer_discovery.session.init(self_.clone());
-        //self_.greylist_refinery.session.init(self_.clone());
         self_
     }
 
@@ -105,7 +101,6 @@ impl OutboundSession {
         }
 
         self.peer_discovery.clone().start().await;
-        //self.greylist_refinery.clone().start().await;
     }
 
     /// Stops the outbound session.
@@ -117,7 +112,6 @@ impl OutboundSession {
         }
 
         self.peer_discovery.clone().stop().await;
-        //self.greylist_refinery.clone().stop().await;
     }
 
     pub async fn slot_info(&self) -> Vec<u32> {

src/net/tests.rs

@@ -18,7 +18,7 @@
 
 // cargo +nightly test --release --all-features --lib p2p -- --include-ignored
 
-use std::{sync::Arc, time::SystemTime};
+use std::sync::Arc;
 
 use log::info;
 use rand::{prelude::SliceRandom, Rng};
@@ -31,7 +31,7 @@ use crate::{
 };
 
 // Number of nodes to spawn and number of peers each node connects to
-const N_NODES: usize = 5;
+const N_NODES: usize = 1;
 const N_CONNS: usize = 2;
 
 // TODO: test whitelist propagation between peers
@@ -88,7 +88,7 @@ async fn hostlist_propagation(ex: Arc<Executor<'static>>) {
     let seed_addr = Url::parse(&format!("tcp://127.0.0.1:{}", 51505)).unwrap();
 
     let mut p2p_instances = vec![];
-    let mut rng = rand::thread_rng();
+    //let mut rng = rand::thread_rng();
 
     info!("Initializing seed network");
     let settings = Settings {
@@ -110,6 +110,15 @@ async fn hostlist_propagation(ex: Arc<Executor<'static>>) {
 
     info!("Initializing outbound nodes");
     for i in 0..N_NODES {
+        // Everyone will connect to N_CONNS random peers.
+        let mut peers = vec![];
+        //for _ in 0..N_CONNS {
+        //    let mut port = 13200 + i;
+        //    while port == 13200 + i {
+        //        port = 13200 + rng.gen_range(0..N_NODES);
+        //    }
+        //    peers.push(Url::parse(&format!("tcp://127.0.0.1:{}", port)).unwrap());
+        //}
         let settings = Settings {
             localnet: true,
             inbound_addrs: vec![Url::parse(&format!("tcp://127.0.0.1:{}", 13200 + i)).unwrap()],
@@ -118,7 +127,7 @@ async fn hostlist_propagation(ex: Arc<Executor<'static>>) {
             outbound_connect_timeout: 10,
             inbound_connections: usize::MAX,
             seeds: vec![seed_addr.clone()],
-            peers: vec![],
+            peers,
             allowed_transports: vec!["tcp".to_string()],
             node_id: i.to_string(),
             //advertise: true,