net: change try_read() and try_write() to read() and write() and cleanup warnings

also experiment with a longer greylist refinery period

src/net/hosts/refinery.rs

@@ -19,16 +19,11 @@
 use std::{sync::Arc, time::UNIX_EPOCH};
 
 use log::{debug, warn};
-use rand::Rng;
 use url::Url;
 
 use super::super::p2p::{P2p, P2pPtr};
 use crate::{
-    net::{
-        connector::Connector,
-        protocol::ProtocolVersion,
-        session::{Session, SessionWeakPtr},
-    },
+    net::{connector::Connector, protocol::ProtocolVersion, session::Session},
     system::{sleep, LazyWeak, StoppableTask, StoppableTaskPtr},
     Error,
 };
@@ -102,7 +97,7 @@ impl GreylistRefinery {
 
             // TODO: create a custom net setting for this timer
             debug!(target: "net::greylist_refinery::run()", "Sleeping...");
-            sleep(5).await;
+            sleep(10).await;
         }
     }
 

src/net/hosts/store.rs

@@ -18,7 +18,7 @@
 
 use std::{collections::HashSet, sync::Arc};
 
-use log::{debug, trace, warn};
+use log::{debug, trace};
 use rand::{
     prelude::{IteratorRandom, SliceRandom},
     rngs::OsRng,
@@ -184,7 +184,7 @@ impl Hosts {
         debug!(target: "net::hosts::store::greylist_store_or_update()", "[START]");
 
         // We filter addresses before writing to the greylist.
-        // We don't need to do this for the whitelist because it reads from the greylist.
+        // We don't need to do this for the whitelist the whitelist is created from the greylist.
         let filtered_addrs = self.filter_addresses(addrs).await;
         let filtered_addrs_len = filtered_addrs.len();
         for (addr, last_seen) in filtered_addrs {
@@ -209,7 +209,7 @@ impl Hosts {
     pub async fn greylist_store(&self, addr: &Url, last_seen: u64) {
         debug!(target: "net::hosts::greylist_store()", "hosts::greylist_store() [START]");
 
-        let mut greylist = self.greylist.try_write().unwrap();
+        let mut greylist = self.greylist.write().await;
 
         // Remove oldest element if the greylist reaches max size.
         if greylist.len() == GREYLIST_MAX_LEN {
@@ -230,7 +230,7 @@ impl Hosts {
     pub async fn whitelist_store(&self, addr: &Url, last_seen: u64) {
         debug!(target: "net::hosts::whitelist_store()", "[START]");
 
-        let mut whitelist = self.whitelist.try_write().unwrap();
+        let mut whitelist = self.whitelist.write().await;
 
         // Remove oldest element if the whitelist reaches max size.
         if whitelist.len() == WHITELIST_MAX_LEN {
@@ -250,7 +250,7 @@ impl Hosts {
     pub async fn whitelist_update_last_seen(&self, addr: &Url, last_seen: u64, index: usize) {
         debug!(target: "net::hosts::store::whitelist_update_last_seen()", "[START]");
 
-        let mut whitelist = self.whitelist.try_write().unwrap();
+        let mut whitelist = self.whitelist.write().await;
 
         whitelist[index] = (addr.clone(), last_seen);
 
@@ -264,7 +264,7 @@ impl Hosts {
     pub async fn greylist_update_last_seen(&self, addr: &Url, last_seen: u64, index: usize) {
         debug!(target: "net::hosts::greylist_update_last_seen()", "[START]");
 
-        let mut greylist = self.greylist.try_write().unwrap();
+        let mut greylist = self.greylist.write().await;
 
         greylist[index] = (addr.clone(), last_seen);
 
@@ -278,7 +278,7 @@ impl Hosts {
         // First lookup the entry using its addr.
         let mut entry = vec![];
 
-        let whitelist = self.whitelist.try_read().unwrap();
+        let whitelist = self.whitelist.read().await;
         for (url, time) in whitelist.iter() {
             if url == addr {
                 entry.push((url.clone(), time.clone()));
@@ -289,7 +289,7 @@ impl Hosts {
         assert!(entry.len() == 1);
 
         // Remove this item from the whitelist.
-        let mut whitelist = self.whitelist.try_write().unwrap();
+        let mut whitelist = self.whitelist.write().await;
         // TODO: test!
         let index = whitelist.iter().position(|x| *x == entry[0]);
         // This should never fail since the entry exists.
@@ -303,7 +303,7 @@ impl Hosts {
 
     pub async fn greylist_remove(&self, addr: &Url, position: usize) {
         debug!(target: "net::refinery::run()", "Removing whitelisted peer {} from greylist", addr);
-        let mut greylist = self.greylist.try_write().unwrap();
+        let mut greylist = self.greylist.write().await;
 
         greylist.remove(position);
 
@@ -592,7 +592,7 @@ impl Hosts {
         limit: Option<usize>,
     ) -> Vec<(Url, u64)> {
         debug!(target: "store::whitelist_fetch_with_schemes", "[START]");
-        let whitelist = self.whitelist.try_read().unwrap();
+        let whitelist = self.whitelist.read().await;
 
         let mut limit = match limit {
             Some(l) => l.min(whitelist.len()),
@@ -618,7 +618,7 @@ impl Hosts {
         // If we didn't find any, pick some from the greylist
         if ret.is_empty() {
             debug!(target: "store::whitelist_fetch_with_schemes", "No matching schemes! We must look at greylist");
-            let greylist = self.greylist.try_read().unwrap();
+            let greylist = self.greylist.read().await;
             for (addr, last_seen) in greylist.iter() {
                 if schemes.contains(&addr.scheme().to_string()) {
                     ret.push((addr.clone(), *last_seen));

src/net/protocol/protocol_address.rs

@@ -29,7 +29,7 @@ use super::{
         message::{AddrsMessage, GetAddrsMessage},
         message_subscriber::MessageSubscription,
         p2p::P2pPtr,
-        session::{OutboundSessionPtr, SESSION_OUTBOUND},
+        session::SESSION_OUTBOUND,
         settings::SettingsPtr,
     },
     protocol_base::{ProtocolBase, ProtocolBasePtr},

src/net/protocol/protocol_seed.rs

@@ -29,7 +29,6 @@ use super::{
         message::{AddrsMessage, GetAddrsMessage},
         message_subscriber::MessageSubscription,
         p2p::P2pPtr,
-        session::OutboundSessionPtr,
         settings::SettingsPtr,
     },
     protocol_base::{ProtocolBase, ProtocolBasePtr},
@@ -53,7 +52,6 @@ impl ProtocolSeed {
     pub async fn init(channel: ChannelPtr, p2p: P2pPtr) -> ProtocolBasePtr {
         let hosts = p2p.hosts();
         let settings = p2p.settings();
-        let session = p2p.session_outbound();
 
         // Create a subscription to address message
         let addr_sub =

src/net/tests.rs

@@ -21,12 +21,12 @@
 use std::sync::Arc;
 
 use log::info;
-use rand::{prelude::SliceRandom, Rng};
+use rand::Rng;
 use smol::{channel, future, Executor};
 use url::Url;
 
 use crate::{
-    net::{P2p, Settings, SESSION_ALL},
+    net::{P2p, Settings},
     system::sleep,
 };
 
@@ -88,37 +88,19 @@ 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();
-
-    //info!("Initializing seed network");
-    //let settings = Settings {
-    //    localnet: true,
-    //    inbound_addrs: vec![seed_addr.clone()],
-    //    external_addrs: vec![seed_addr.clone()],
-    //    outbound_connections: 0,
-    //    outbound_connect_timeout: 2,
-    //    inbound_connections: usize::MAX,
-    //    peers: vec![],
-    //    allowed_transports: vec!["tcp".to_string()],
-    //    node_id: "seed".to_string(),
-    //    //advertise: true,
-    //    ..Default::default()
-    //};
-
-    //let p2p = P2p::new(settings, ex.clone()).await;
-    //p2p_instances.push(p2p);
+    let mut rng = rand::thread_rng();
 
     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());
-        //}
+        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()],
@@ -143,18 +125,26 @@ async fn hostlist_propagation(ex: Arc<Executor<'static>>) {
         p2p.clone().start().await.unwrap();
     }
 
-    info!("Waiting 10s until all peers connect");
-    sleep(15).await;
+    info!("Waiting until all peers connect");
+    sleep(30).await;
 
-    info!("Inspecting peerlists...");
+    info!("Inspecting hostlists...");
     for p2p in p2p_instances.iter() {
         let hosts = p2p.hosts();
         assert!(!hosts.is_empty_greylist().await);
+        //assert!(!hosts.is_empty_whitelist().await);
+
         let greylist = hosts.greylist.read().await;
-        info!("Peer {}", p2p.settings().node_id);
+        let whitelist = hosts.whitelist.read().await;
+
+        info!("Node {}", p2p.settings().node_id);
         for (i, (url, last_seen)) in greylist.iter().enumerate() {
             info!("Greylist entry {}: {}, {}", i, url, last_seen);
         }
+
+        for (i, (url, last_seen)) in whitelist.iter().enumerate() {
+            info!("Whitelist entry {}: {}, {}", i, url, last_seen);
+        }
     }
 
     // Stop the P2P network