Просмотр исходного кода

net: add `last_connection` timer to more empirically track connection status

Before we were simply relying on channels() returning an empty
vector to determine our connection status, however this does not
work for e.g. Lilith that frequently has no active connections in
channels(), but may have received a connection more recently than the
time_with_no_connections refinery timeout.
draoi 2 лет назад
Родитель
Сommit
0f5fcba2c2
2 измененных файлов с 29 добавлено и 27 удалено
  1. 17 22
      src/net/hosts/refinery.rs
  2. 12 5
      src/net/hosts/store.rs

+ 17 - 22
src/net/hosts/refinery.rs

@@ -66,6 +66,7 @@ impl GreylistRefinery {
         let ex = self.p2p().executor();
         let ex = self.p2p().executor();
         self.process.clone().start(
         self.process.clone().start(
             async move {
             async move {
+                //self.listen_for_channels().await;
                 self.run().await;
                 self.run().await;
                 unreachable!();
                 unreachable!();
             },
             },
@@ -93,10 +94,8 @@ impl GreylistRefinery {
     // This method will remove from the greylist and store on the whitelist
     // This method will remove from the greylist and store on the whitelist
     // providing the peer is responsive.
     // providing the peer is responsive.
     async fn run(self: Arc<Self>) {
     async fn run(self: Arc<Self>) {
-        let mut last_online = Instant::now();
         let settings = self.p2p().settings();
         let settings = self.p2p().settings();
         let hosts = self.p2p().hosts();
         let hosts = self.p2p().hosts();
-
         loop {
         loop {
             sleep(settings.greylist_refinery_interval).await;
             sleep(settings.greylist_refinery_interval).await;
 
 
@@ -109,27 +108,23 @@ impl GreylistRefinery {
 
 
             // Pause the refinery if we've had zero connections for longer than the configured
             // Pause the refinery if we've had zero connections for longer than the configured
             // limit.
             // limit.
-            if hosts.channels().await.is_empty() {
-                let time_offline = Instant::now().duration_since(last_online);
-                let offline_limit = Duration::from_secs(settings.time_with_no_connections);
-
-                if time_offline >= offline_limit {
-                    warn!(target: "net::refinery", "No connections for {}s. Refinery paused.",
-                          time_offline.as_secs());
-
-                    // It is neccessary to clear suspended hosts at this point, otherwise these
-                    // 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;
-                    for host in suspended_hosts {
-                        hosts.unregister(&host).await;
-                    }
-
-                    continue
+            let offline_limit = Duration::from_secs(settings.time_with_no_connections);
+            let offline_timer = Instant::now().duration_since(*hosts.last_connection.read().await);
+
+            if hosts.channels().await.is_empty() && offline_timer >= offline_limit {
+                warn!(target: "net::refinery", "No connections for {}s. Refinery paused.",
+                          offline_timer.as_secs());
+
+                // It is neccessary to clear suspended hosts at this point, otherwise these
+                // 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;
+                for host in suspended_hosts {
+                    hosts.unregister(&host).await;
                 }
                 }
-            } else {
-                last_online = Instant::now();
+
+                continue
             }
             }
 
 
             // Only attempt to refine peers that match our transports.
             // Only attempt to refine peers that match our transports.

+ 12 - 5
src/net/hosts/store.rs

@@ -16,7 +16,7 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
  */
 
 
-use std::{collections::HashMap, fmt, fs, fs::File, sync::Arc};
+use std::{collections::HashMap, fmt, fs, fs::File, sync::Arc, time::Instant};
 
 
 use log::{debug, error, info, trace, warn};
 use log::{debug, error, info, trace, warn};
 use rand::{prelude::IteratorRandom, rngs::OsRng, Rng};
 use rand::{prelude::IteratorRandom, rngs::OsRng, Rng};
@@ -702,6 +702,9 @@ pub struct Hosts {
     /// Subscriber for notifications of new channels
     /// Subscriber for notifications of new channels
     channel_subscriber: SubscriberPtr<Result<ChannelPtr>>,
     channel_subscriber: SubscriberPtr<Result<ChannelPtr>>,
 
 
+    /// Keeps track of the last time a connection was made.
+    pub last_connection: RwLock<Instant>,
+
     /// Pointer to configured P2P settings
     /// Pointer to configured P2P settings
     settings: SettingsPtr,
     settings: SettingsPtr,
 }
 }
@@ -714,6 +717,7 @@ impl Hosts {
             container: HostContainer::new(),
             container: HostContainer::new(),
             store_subscriber: Subscriber::new(),
             store_subscriber: Subscriber::new(),
             channel_subscriber: Subscriber::new(),
             channel_subscriber: Subscriber::new(),
+            last_connection: RwLock::new(Instant::now()),
             settings,
             settings,
         })
         })
     }
     }
@@ -736,7 +740,7 @@ impl Hosts {
         for (i, (addr, last_seen)) in filtered_addrs.iter().enumerate() {
         for (i, (addr, last_seen)) in filtered_addrs.iter().enumerate() {
             if self.try_register(addr.clone(), HostState::Insert).await.is_err() {
             if self.try_register(addr.clone(), HostState::Insert).await.is_err() {
                 debug!(target: "net::hosts::store_or_update()",
                 debug!(target: "net::hosts::store_or_update()",
-            "We are already tracking {}. Skipping...", addr);
+                "{} is already registered. Skipping...", addr);
                 continue
                 continue
             }
             }
 
 
@@ -853,13 +857,16 @@ impl Hosts {
 
 
         self.try_register(address.clone(), HostState::Connected(channel.clone())).await?;
         self.try_register(address.clone(), HostState::Connected(channel.clone())).await?;
 
 
-        self.channel_subscriber.notify(Ok(channel)).await;
+        self.channel_subscriber.notify(Ok(channel.clone())).await;
+
+        let mut last_online = self.last_connection.write().await;
+        *last_online = Instant::now();
+
         Ok(())
         Ok(())
     }
     }
 
 
     pub async fn subscribe_store(&self) -> Result<Subscription<usize>> {
     pub async fn subscribe_store(&self) -> Result<Subscription<usize>> {
-        let sub = self.store_subscriber.clone().subscribe().await;
-        Ok(sub)
+        Ok(self.store_subscriber.clone().subscribe().await)
     }
     }
 
 
     // Verify whether a URL is local.
     // Verify whether a URL is local.