hosts: use sync Mutex for HostRegistry

We don't need to use an AsyncMutex here since we're not holding across .await points or for long periods of time.

Using a sync Mutex here also fixes some really weird fairness behaviors we observed in the smol::lock::Mutex where writers in the priority queue were occasionally getting ignored. This was apparently not a deadlock since subsequent and prior readers and writers were able to access the data with no problems.

bin/darkfid/src/rpc_tx.rs

@@ -134,7 +134,7 @@ impl Darkfid {
         };
 
         self.p2p.broadcast(&tx).await;
-        if self.p2p.hosts().channels().await.is_empty() {
+        if self.p2p.hosts().channels().is_empty() {
             warn!(target: "darkfid::rpc::tx_broadcast", "No connected channels to broadcast tx");
         }
 

bin/darkfid/src/task/miner.rs

@@ -331,7 +331,7 @@ async fn mine_next_block(
     extended_fork.module.verify_current_block(&next_block)?;
 
     // Check if we are connected to the network
-    if !skip_sync && node.p2p.hosts().channels().await.is_empty() {
+    if !skip_sync && node.p2p.hosts().channels().is_empty() {
         return Err(Error::NetworkNotConnected)
     }
 

bin/darkfid/src/task/sync.rs

@@ -142,7 +142,7 @@ async fn synced_peers(
     let mut tips = HashMap::new();
     loop {
         // Grab channels
-        let peers = node.p2p.hosts().channels().await;
+        let peers = node.p2p.hosts().channels();
 
         // Check anyone is connected
         if !peers.is_empty() {

bin/lilith/src/main.rs

@@ -198,7 +198,7 @@ impl Lilith {
             let url = &entry.0;
             let last_seen = &entry.1;
 
-            if !hosts.refinable(url.clone()).await {
+            if !hosts.refinable(url.clone()) {
                 debug!(target: "net::refinery::whitelist_refinery", "Addr={} not available!",
                        url.clone());
 

src/event_graph/mod.rs

@@ -213,7 +213,7 @@ impl EventGraph {
         //   from the beginning
 
         // Get references to all our peers.
-        let channels = self.p2p.hosts().channels().await;
+        let channels = self.p2p.hosts().channels();
         let mut communicated_peers = channels.len();
         info!(
             target: "event_graph::dag_sync()",

src/net/hosts.rs

@@ -844,7 +844,7 @@ impl Hosts {
 
         // Then ensure we aren't currently trying to add this peer to the hostlist.
         for (i, (addr, last_seen)) in filtered_addrs.iter().enumerate() {
-            if let Err(e) = self.try_register(addr.clone(), HostState::Insert).await {
+            if let Err(e) = self.try_register(addr.clone(), HostState::Insert) {
                 debug!(target: "net::hosts::store_or_update", "Cannot insert addr={}, err={}",
                        addr.clone(), e);
 
@@ -856,7 +856,7 @@ impl Hosts {
 
             // Free up this peer for usage by other parts of the code base.
             // This is a safe since the hostlist modification is now complete.
-            self.unregister(addr).await;
+            self.unregister(addr);
         }
 
         self.store_publisher.notify(addrs_len).await;
@@ -865,13 +865,13 @@ impl Hosts {
 
     /// Check whether a peer is available to be refined currently. Returns true
     /// if available, false otherwise.
-    pub async fn refinable(&self, addr: Url) -> bool {
-        self.try_register(addr.clone(), HostState::Refine).await.is_ok()
+    pub fn refinable(&self, addr: Url) -> bool {
+        self.try_register(addr.clone(), HostState::Refine).is_ok()
     }
 
     /// Try to update the registry. If the host already exists, try to update its state.
     /// Otherwise add the host to the registry along with its state.
-    pub(in crate::net) async fn try_register(
+    pub(in crate::net) fn try_register(
         &self,
         addr: Url,
         new_state: HostState,
@@ -926,7 +926,7 @@ impl Hosts {
                 continue
             }
 
-            if let Err(e) = self.try_register(host.clone(), HostState::Connect).await {
+            if let Err(e) = self.try_register(host.clone(), HostState::Connect) {
                 trace!(target: "net::hosts::check_addrs", "Skipping addr={}, err={}",
                        host.clone(), e);
                 continue
@@ -946,13 +946,13 @@ impl Hosts {
     /// Misuse of this call is dangerous since it frees up the peer to be used by
     /// the refinery or outbound connect loop, and may result in invalid states. It should
     /// only be called when it is completely safe to do so.
-    pub(in crate::net) async fn unregister(&self, addr: &Url) {
+    pub(in crate::net) fn unregister(&self, addr: &Url) {
         self.registry.lock().unwrap().remove(addr);
         debug!(target: "net::hosts::unregister()", "Removed {} from HostRegistry", addr);
     }
 
     /// Returns the list of connected channels.
-    pub async fn channels(&self) -> Vec<ChannelPtr> {
+    pub fn channels(&self) -> Vec<ChannelPtr> {
         let registry = self.registry.lock().unwrap();
         let mut channels = Vec::new();
 
@@ -965,7 +965,7 @@ impl Hosts {
     }
 
     /// Returns the list of suspended channels.
-    pub(in crate::net) async fn suspended(&self) -> Vec<Url> {
+    pub(in crate::net) fn suspended(&self) -> Vec<Url> {
         let registry = self.registry.lock().unwrap();
         let mut addrs = Vec::new();
 
@@ -978,8 +978,8 @@ impl Hosts {
     }
 
     /// Retrieve a random connected channel
-    pub async fn random_channel(&self) -> ChannelPtr {
-        let channels = self.channels().await;
+    pub fn random_channel(&self) -> ChannelPtr {
+        let channels = self.channels();
         let position = rand::thread_rng().gen_range(0..channels.len());
         channels[position].clone()
     }
@@ -991,7 +991,7 @@ impl Hosts {
         // This will panic if we are already connected to this peer, this peer
         // is suspended, or this peer is currently being inserted into the hostlist.
         // None of these scenarios should ever happen.
-        self.try_register(address.clone(), HostState::Connected(channel.clone())).await.unwrap();
+        self.try_register(address.clone(), HostState::Connected(channel.clone())).unwrap();
 
         // Notify that channel processing was successful
         self.channel_publisher.notify(Ok(channel.clone())).await;
@@ -1244,7 +1244,7 @@ impl Hosts {
         self.move_host(addr, last_seen, HostColor::Grey).await?;
 
         // Free up this addr for future operations.
-        self.unregister(addr).await;
+        self.unregister(addr);
 
         Ok(())
     }
@@ -1266,7 +1266,7 @@ impl Hosts {
                addr, destination);
 
         // This should never panic. Failure indicates a misuse of the HostState API.
-        self.try_register(addr.clone(), HostState::Move).await.unwrap();
+        self.try_register(addr.clone(), HostState::Move).unwrap();
 
         match destination {
             // Downgrade to grey. Remove from white and gold.

src/net/p2p.rs

@@ -170,7 +170,7 @@ impl P2p {
     /// the ones provided in `exclude_list`.
     pub async fn broadcast_with_exclude<M: Message>(&self, message: &M, exclude_list: &[Url]) {
         let mut channels = Vec::new();
-        for channel in self.hosts().channels().await {
+        for channel in self.hosts().channels() {
             if exclude_list.contains(channel.address()) {
                 continue
             }
@@ -204,8 +204,8 @@ impl P2p {
         let _results: Vec<_> = futures.collect().await;
     }
 
-    pub async fn is_connected(&self) -> bool {
-        !self.hosts().channels().await.is_empty()
+    pub fn is_connected(&self) -> bool {
+        !self.hosts().channels().is_empty()
     }
 
     /// Return an atomic pointer to the set network settings

src/net/session/manual_session.rs

@@ -164,7 +164,7 @@ impl Slot {
                 return Ok(())
             }
 
-            match self.p2p().hosts().try_register(self.addr.clone(), HostState::Connect).await {
+            match self.p2p().hosts().try_register(self.addr.clone(), HostState::Connect) {
                 Ok(_) => {
                     match self.connector.connect(&self.addr).await {
                         Ok((url, channel)) => {
@@ -197,7 +197,7 @@ impl Slot {
 
                             // Stop tracking this peer, to avoid it getting stuck in the Connect
                             // state. This is safe since we have failed to connect at this point.
-                            self.p2p().hosts().unregister(&self.addr).await;
+                            self.p2p().hosts().unregister(&self.addr);
                         }
                     }
                 }

src/net/session/mod.rs

@@ -83,9 +83,9 @@ pub async fn remove_sub_on_stop(p2p: P2pPtr, channel: ChannelPtr, type_id: Sessi
     }
 
     // Remove channel from the HostRegistry. Free up this addr for any future operation.
-    hosts.unregister(channel.address()).await;
+    hosts.unregister(channel.address());
 
-    if hosts.channels().await.is_empty() {
+    if hosts.channels().is_empty() {
         hosts.disconnect_publisher.notify(Error::NetworkNotConnected).await;
     }
     debug!(target: "net::session::remove_sub_on_stop()", "[END]");

src/net/session/outbound_session.rs

@@ -363,7 +363,7 @@ impl Slot {
                 self.p2p().hosts().move_host(&addr, last_seen, HostColor::Grey).await?;
 
                 // Mark its state as Suspend, which sends this node to the Refinery for processing.
-                self.p2p().hosts().try_register(addr.clone(), HostState::Suspend).await.unwrap();
+                self.p2p().hosts().try_register(addr.clone(), HostState::Suspend).unwrap();
 
                 continue
             }
@@ -406,7 +406,7 @@ impl Slot {
                 self.p2p().hosts().move_host(&addr, last_seen, HostColor::Grey).await?;
 
                 // Mark its state as Suspend, which sends it to the Refinery for processing.
-                self.p2p().hosts().try_register(addr.clone(), HostState::Suspend).await.unwrap();
+                self.p2p().hosts().try_register(addr.clone(), HostState::Suspend).unwrap();
 
                 // Notify that channel processing failed
                 self.p2p().hosts().channel_publisher.notify(Err(Error::ConnectFailed)).await;
@@ -543,7 +543,7 @@ impl PeerDiscoveryBase for PeerDiscovery {
 
             // First 2 times try sending GetAddr to the network.
             // 3rd time do a seed sync.
-            if p2p.is_connected().await && current_attempt <= 2 {
+            if p2p.is_connected() && current_attempt <= 2 {
                 // Broadcast the GetAddrs message to all active channels.
                 // If we have no active channels, we will perform a SeedSyncSession instead.
 

src/net/session/refine_session.rs

@@ -238,7 +238,7 @@ impl GreylistRefinery {
             let offline_timer =
                 { Instant::now().duration_since(*hosts.last_connection.lock().unwrap()) };
 
-            if hosts.channels().await.is_empty() && offline_timer >= offline_limit {
+            if hosts.channels().is_empty() && offline_timer >= offline_limit {
                 warn!(target: "net::refinery", "No connections for {}s. GreylistRefinery paused.",
                           offline_timer.as_secs());
 
@@ -246,9 +246,9 @@ impl GreylistRefinery {
                 // hosts cannot be connected to in Outbound Session. Failure to do this could
                 // result in the refinery being paused forver (since connections could never be
                 // made).
-                let suspended_hosts = hosts.suspended().await;
+                let suspended_hosts = hosts.suspended();
                 for host in suspended_hosts {
-                    hosts.unregister(&host).await;
+                    hosts.unregister(&host);
                 }
 
                 continue
@@ -263,7 +263,7 @@ impl GreylistRefinery {
                 Some((entry, _)) => {
                     let url = &entry.0;
 
-                    if let Err(e) = hosts.try_register(url.clone(), HostState::Refine).await {
+                    if let Err(e) = hosts.try_register(url.clone(), HostState::Refine) {
                         debug!(target: "net::refinery", "Unable to refine addr={}, err={}",
                                url.clone(), e);
                         continue
@@ -291,7 +291,7 @@ impl GreylistRefinery {
                         // Remove this entry from HostRegistry to avoid this host getting
                         // stuck in the Refining state. This is a safe since the hostlist
                         // modification is now complete.
-                        hosts.unregister(url).await;
+                        hosts.unregister(url);
 
                         continue
                     }
@@ -305,7 +305,7 @@ impl GreylistRefinery {
                     hosts.move_host(url, last_seen, HostColor::White).await.unwrap();
 
                     // When move is complete we can safely stop tracking this peer.
-                    hosts.unregister(url).await;
+                    hosts.unregister(url);
 
                     debug!(target: "net::refinery", "GreylistRefinery complete!");
                     continue

src/net/tests.rs

@@ -455,7 +455,7 @@ async fn p2p_test_real(ex: Arc<Executor<'static>>) {
         info!("========================================================");
         info!("Checking manual node={}", p2p.settings().node_id);
         info!("========================================================");
-        let channels = p2p.hosts().channels().await;
+        let channels = p2p.hosts().channels();
         assert!(channels.len() == N_CONNS * 2);
     }
 

src/rpc/p2p_method.rs

@@ -28,7 +28,7 @@ use crate::net;
 pub trait HandlerP2p: Sync + Send {
     async fn p2p_get_info(&self, id: u16, _params: JsonValue) -> JsonResult {
         let mut channels = Vec::new();
-        for channel in self.p2p().hosts().channels().await {
+        for channel in self.p2p().hosts().channels() {
             let session = match channel.session_type_id() {
                 net::session::SESSION_INBOUND => "inbound",
                 net::session::SESSION_OUTBOUND => "outbound",