net: Implement mutable Settings which allow dynamic reconfiguration

bin/darkfid/src/proto/protocol_proposal.rs

@@ -158,7 +158,7 @@ impl ProtocolProposal {
             // Node waits for response
             let response = match self
                 .proposals_response_sub
-                .receive_with_timeout(self.p2p.settings().outbound_connect_timeout)
+                .receive_with_timeout(self.p2p.settings().read().await.outbound_connect_timeout)
                 .await
             {
                 Ok(r) => r,

bin/darkfid/src/task/sync.rs

@@ -138,7 +138,7 @@ async fn synced_peers(
     checkpoint: Option<(u32, HeaderHash)>,
 ) -> Result<HashMap<(u32, [u8; 32]), Vec<ChannelPtr>>> {
     info!(target: "darkfid::task::sync::synced_peers", "Receiving tip from peers...");
-    let comms_timeout = node.p2p.settings().outbound_connect_timeout;
+    let comms_timeout = node.p2p.settings().read().await.outbound_connect_timeout;
     let mut tips = HashMap::new();
     loop {
         // Grab channels
@@ -249,7 +249,7 @@ async fn retrieve_headers(
     for peer in peers {
         peer_subs.push(peer.subscribe_msg::<HeaderSyncResponse>().await?);
     }
-    let comms_timeout = node.p2p.settings().outbound_connect_timeout;
+    let comms_timeout = node.p2p.settings().read().await.outbound_connect_timeout;
 
     // We subtract 1 since tip_height is increased by one
     let total = tip_height - last_known - 1;
@@ -354,7 +354,7 @@ async fn retrieve_blocks(
     for peer in peers {
         peer_subs.push(peer.subscribe_msg::<SyncResponse>().await?);
     }
-    let comms_timeout = node.p2p.settings().outbound_connect_timeout;
+    let comms_timeout = node.p2p.settings().read().await.outbound_connect_timeout;
 
     let mut received_blocks = 0;
     let total = node.validator.blockchain.headers.len_sync();
@@ -443,8 +443,9 @@ async fn sync_best_fork(node: &Darkfid, peers: &[ChannelPtr], last_tip: &HeaderH
     channel.send(&request).await?;
 
     // Node waits for response
-    let response =
-        response_sub.receive_with_timeout(node.p2p.settings().outbound_connect_timeout).await?;
+    let response = response_sub
+        .receive_with_timeout(node.p2p.settings().read().await.outbound_connect_timeout)
+        .await?;
 
     // Verify and store retrieved proposals
     debug!(target: "darkfid::task::sync_task", "Processing received proposals");

bin/darkfid/src/tests/mod.rs

@@ -90,7 +90,7 @@ async fn sync_blocks_real(ex: Arc<Executor<'static>>) -> Result<()> {
 
     let charlie_url = Url::parse("tcp+tls://127.0.0.1:18342")?;
     settings.inbound_addrs = vec![charlie_url];
-    let bob_url = th.bob.p2p.settings().inbound_addrs[0].clone();
+    let bob_url = th.bob.p2p.settings().read().await.inbound_addrs[0].clone();
     settings.peers = vec![bob_url];
     let charlie = generate_node(
         &th.vks,

bin/darkfid/src/tests/sync_forks.rs

@@ -67,7 +67,7 @@ async fn sync_forks_real(ex: Arc<Executor<'static>>) -> Result<()> {
 
     let charlie_url = Url::parse("tcp+tls://127.0.0.1:18442")?;
     settings.inbound_addrs = vec![charlie_url];
-    let bob_url = th.bob.p2p.settings().inbound_addrs[0].clone();
+    let bob_url = th.bob.p2p.settings().read().await.inbound_addrs[0].clone();
     settings.peers = vec![bob_url];
     let charlie =
         generate_node(&th.vks, &th.validator_config, &settings, &ex, false, false, None).await?;

bin/lilith/src/main.rs

@@ -118,7 +118,7 @@ impl Spawn {
 
     async fn info(&self) -> JsonValue {
         let mut addr_vec = vec![];
-        for addr in &self.p2p.settings().inbound_addrs {
+        for addr in &self.p2p.settings().read().await.inbound_addrs {
             addr_vec.push(JsonValue::String(addr.as_ref().to_string()));
         }
 

src/event_graph/mod.rs

@@ -251,7 +251,7 @@ impl EventGraph {
             };
 
             let peer_tips = match timeout(
-                Duration::from_secs(self.p2p.settings().outbound_connect_timeout),
+                Duration::from_secs(self.p2p.settings().read().await.outbound_connect_timeout),
                 tip_rep_sub.receive(),
             )
             .await
@@ -355,7 +355,7 @@ impl EventGraph {
                 }
 
                 let parent = match timeout(
-                    Duration::from_secs(self.p2p.settings().outbound_connect_timeout),
+                    Duration::from_secs(self.p2p.settings().read().await.outbound_connect_timeout),
                     ev_rep_sub.receive(),
                 )
                 .await

src/event_graph/proto.rs

@@ -258,7 +258,7 @@ impl ProtocolEventGraph {
 
                     let parents = match timeout(
                         Duration::from_secs(
-                            self.event_graph.p2p.settings().outbound_connect_timeout,
+                            self.event_graph.p2p.settings().read().await.outbound_connect_timeout,
                         ),
                         self.ev_rep_sub.receive(),
                     )

src/net/acceptor.rs

@@ -63,13 +63,10 @@ impl Acceptor {
 
     /// Start accepting inbound socket connections
     pub async fn start(self: Arc<Self>, endpoint: Url, ex: Arc<Executor<'_>>) -> Result<()> {
-        let listener = Listener::new(
-            endpoint,
-            self.session.upgrade().unwrap().p2p().settings().datastore.clone(),
-        )
-        .await?
-        .listen()
-        .await?;
+        let datastore =
+            self.session.upgrade().unwrap().p2p().settings().read().await.datastore.clone();
+
+        let listener = Listener::new(endpoint, datastore).await?.listen().await?;
 
         self.accept(listener, ex);
         Ok(())
@@ -110,7 +107,9 @@ impl Acceptor {
 
         loop {
             // Refuse new connections if we're up to the connection limit
-            let limit = self.session.upgrade().unwrap().p2p().settings().inbound_connections;
+            let limit =
+                self.session.upgrade().unwrap().p2p().settings().read().await.inbound_connections;
+
             if self.clone().conn_count.load(SeqCst) >= limit {
                 // This will get notified every time an inbound channel is stopped.
                 // These channels are the channels spawned below on listener.next().is_ok().

src/net/connector.rs

@@ -16,13 +16,17 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::{sync::Arc, time::Duration};
+use std::{
+    sync::{atomic::Ordering, Arc},
+    time::Duration,
+};
 
 use futures::{
     future::{select, Either},
     pin_mut,
 };
 use log::warn;
+use smol::lock::RwLock as AsyncRwLock;
 use url::Url;
 
 use super::{
@@ -37,7 +41,7 @@ use crate::{system::CondVar, Error, Result};
 /// Create outbound socket connections
 pub struct Connector {
     /// P2P settings
-    settings: Arc<Settings>,
+    settings: Arc<AsyncRwLock<Settings>>,
     /// Weak pointer to the session
     pub session: SessionWeakPtr,
     /// Stop signal that aborts the connector if received.
@@ -46,7 +50,7 @@ pub struct Connector {
 
 impl Connector {
     /// Create a new connector with given network settings
-    pub fn new(settings: Arc<Settings>, session: SessionWeakPtr) -> Self {
+    pub fn new(settings: Arc<AsyncRwLock<Settings>>, session: SessionWeakPtr) -> Self {
         Self { settings, session, stop_signal: CondVar::new() }
     }
 
@@ -60,11 +64,17 @@ impl Connector {
             return Err(Error::ConnectFailed)
         }
 
-        let mut endpoint = url.clone();
+        let settings = self.settings.read().await;
+        let transports = settings.allowed_transports.clone();
+        let transport_mixing = settings.transport_mixing;
+        let datastore = settings.datastore.clone();
+        let outbound_connect_timeout = settings.outbound_connect_timeout;
+        drop(settings);
 
-        let transports = &self.settings.allowed_transports;
+        let mut endpoint = url.clone();
         let scheme = endpoint.scheme();
-        if !transports.contains(&scheme.to_string()) && self.settings.transport_mixing {
+
+        if !transports.contains(&scheme.to_string()) && transport_mixing {
             if transports.contains(&"tor".to_string()) && scheme == "tcp" {
                 endpoint.set_scheme("tor")?;
             } else if transports.contains(&"tor+tls".to_string()) && scheme == "tcp+tls" {
@@ -76,8 +86,8 @@ impl Connector {
             }
         }
 
-        let dialer = Dialer::new(endpoint.clone(), self.settings.datastore.clone()).await?;
-        let timeout = Duration::from_secs(self.settings.outbound_connect_timeout);
+        let dialer = Dialer::new(endpoint.clone(), datastore).await?;
+        let timeout = Duration::from_secs(outbound_connect_timeout);
 
         let stop_fut = async {
             self.stop_signal.wait().await;
@@ -102,8 +112,13 @@ impl Connector {
             Either::Left((Err(e), _)) => {
                 // If we get ENETUNREACH, we don't have IPv6 connectivity so note it down.
                 if e.raw_os_error() == Some(libc::ENETUNREACH) {
-                    *self.session.upgrade().unwrap().p2p().hosts().ipv6_available.lock().unwrap() =
-                        false;
+                    self.session
+                        .upgrade()
+                        .unwrap()
+                        .p2p()
+                        .hosts()
+                        .ipv6_available
+                        .store(false, Ordering::SeqCst);
                 }
                 Err(e.into())
             }

src/net/hosts.rs

@@ -20,12 +20,16 @@ use std::{
     collections::HashMap,
     fmt, fs,
     fs::File,
-    sync::{Arc, Mutex, RwLock},
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc, Mutex, RwLock,
+    },
     time::{Instant, UNIX_EPOCH},
 };
 
 use log::{debug, error, info, trace, warn};
 use rand::{prelude::IteratorRandom, rngs::OsRng, Rng};
+use smol::lock::RwLock as AsyncRwLock;
 use url::Url;
 
 use super::{settings::Settings, ChannelPtr};
@@ -814,15 +818,15 @@ pub struct Hosts {
     pub(in crate::net) last_connection: Mutex<Instant>,
 
     /// Marker for IPv6 availability
-    pub(in crate::net) ipv6_available: Mutex<bool>,
+    pub(in crate::net) ipv6_available: AtomicBool,
 
     /// Pointer to configured P2P settings
-    settings: Arc<Settings>,
+    settings: Arc<AsyncRwLock<Settings>>,
 }
 
 impl Hosts {
     /// Create a new hosts list
-    pub(in crate::net) fn new(settings: Arc<Settings>) -> HostsPtr {
+    pub(in crate::net) fn new(settings: Arc<AsyncRwLock<Settings>>) -> HostsPtr {
         Arc::new(Self {
             registry: Mutex::new(HashMap::new()),
             container: HostContainer::new(),
@@ -830,7 +834,7 @@ impl Hosts {
             channel_publisher: Publisher::new(),
             disconnect_publisher: Publisher::new(),
             last_connection: Mutex::new(Instant::now()),
-            ipv6_available: Mutex::new(true),
+            ipv6_available: AtomicBool::new(true),
             settings,
         })
     }
@@ -843,7 +847,7 @@ impl Hosts {
         // First filter these address to ensure this peer doesn't exist in our black, gold or
         // whitelist and apply transport filtering. If we don't support this transport,
         // store the peer on our dark list to broadcast to other nodes.
-        let filtered_addrs = self.filter_addresses(self.settings.clone(), addrs);
+        let filtered_addrs = self.filter_addresses(addrs).await;
         let mut addrs_len = 0;
 
         if filtered_addrs.is_empty() {
@@ -923,22 +927,28 @@ impl Hosts {
 
     // Loop through hosts selected by Outbound Session and see if any of them are
     // free to connect to.
-    pub(in crate::net) fn check_addrs(&self, hosts: Vec<(Url, u64)>) -> Option<(Url, u64)> {
+    pub(in crate::net) async fn check_addrs(&self, hosts: Vec<(Url, u64)>) -> Option<(Url, u64)> {
         trace!(target: "net::hosts::check_addrs()", "[START]");
+
+        let seeds = self.settings.read().await.seeds.clone();
+
         for (host, last_seen) in hosts {
             // Print a warning if we are trying to connect to a seed node in
             // Outbound session. This shouldn't happen as we reject configured
             // seed nodes from entering our hostlist in filter_addrs().
-            if self.settings.seeds.contains(&host) {
-                warn!(target: "net::hosts::check_addrs",
-                      "Seed addr={} has entered the hostlist! Skipping",
-                      host.clone());
+            if seeds.contains(&host) {
+                warn!(
+                    target: "net::hosts::check_addrs",
+                    "Seed addr={} has entered the hostlist! Skipping", host.clone(),
+                );
                 continue
             }
 
             if let Err(e) = self.try_register(host.clone(), HostState::Connect) {
-                trace!(target: "net::hosts::check_addrs", "Skipping addr={}, err={}",
-                       host.clone(), e);
+                trace!(
+                    target: "net::hosts::check_addrs",
+                    "Skipping addr={}, err={}", host.clone(), e,
+                );
                 continue
             }
 
@@ -1073,8 +1083,8 @@ impl Hosts {
     }
 
     /// Import blacklisted peers specified in the config file.
-    pub(in crate::net) fn import_blacklist(&self) -> Result<()> {
-        for (mut host, ports) in self.settings.blacklist.clone() {
+    pub(in crate::net) async fn import_blacklist(&self) -> Result<()> {
+        for (mut host, ports) in self.settings.read().await.blacklist.clone() {
             // If the ports are empty, simply store the host_str. We will use this to
             // blacklist all ports of a given peer in `block_all_ports()`.
             if ports.is_empty() {
@@ -1105,11 +1115,12 @@ impl Hosts {
 
     /// Filter given addresses based on certain rulesets and validity. Strictly called only on
     /// the first time learning of new peers.
-    fn filter_addresses(&self, settings: Arc<Settings>, addrs: &[(Url, u64)]) -> Vec<(Url, u64)> {
-        debug!(target: "net::hosts::filter_addresses()", "Filtering addrs: {:?}", addrs);
+    async fn filter_addresses(&self, addrs: &[(Url, u64)]) -> Vec<(Url, u64)> {
+        debug!(target: "net::hosts::filter_addresses", "Filtering addrs: {:?}", addrs);
         let mut ret = vec![];
-        let localnet = self.settings.localnet;
-        let ipv6_available: bool = { *self.ipv6_available.lock().unwrap() };
+
+        // Acquire read lock on P2P settings. Dropped when this function finishes.
+        let settings = self.settings.read().await;
 
         'addr_loop: for (addr_, last_seen) in addrs {
             // Validate that the format is `scheme://host_str:port`
@@ -1118,15 +1129,19 @@ impl Hosts {
                 addr_.cannot_be_a_base() ||
                 addr_.path_segments().is_some()
             {
-                debug!(target: "net::hosts::filter_addresses()",
-                       "[{}] has invalid addr format. Skipping", addr_);
+                debug!(
+                    target: "net::hosts::filter_addresses",
+                    "[{}] has invalid addr format. Skipping", addr_,
+                );
                 continue
             }
 
             // Configured seeds should never enter the hostlist.
-            if self.settings.seeds.contains(addr_) {
-                debug!(target: "net::hosts::filter_addresses()",
-                       "[{}] is a configured seed. Skipping", addr_);
+            if settings.seeds.contains(addr_) {
+                debug!(
+                    target: "net::hosts::filter_addresses",
+                    "[{}] is a configured seed. Skipping", addr_,
+                );
                 continue
             }
 
@@ -1134,19 +1149,23 @@ impl Hosts {
             if self.container.contains(HostColor::Black as usize, addr_) ||
                 self.block_all_ports(addr_.host_str().unwrap().to_string())
             {
-                warn!(target: "net::hosts::filter_addresses()",
-                      "[{}] is blacklisted", addr_);
+                warn!(
+                    target: "net::hosts::filter_addresses",
+                    "[{}] is blacklisted", addr_,
+                );
                 continue
             }
 
             let host_str = addr_.host_str().unwrap();
 
-            if !localnet {
+            if !settings.localnet {
                 // Our own external addresses should never enter the hosts set.
                 for ext in &settings.external_addrs {
                     if host_str == ext.host_str().unwrap() {
-                        debug!(target: "net::hosts::filter_addresses()",
-                               "[{}] is our own external addr. Skipping", addr_);
+                        debug!(
+                            target: "net::hosts::filter_addresses",
+                            "[{}] is our own external addr. Skipping", addr_,
+                        );
                         continue 'addr_loop
                     }
                 }
@@ -1154,8 +1173,10 @@ impl Hosts {
                 // On localnet, make sure ours ports don't enter the host set.
                 for ext in &settings.external_addrs {
                     if addr_.port() == ext.port() {
-                        debug!(target: "net::hosts::filter_addresses()",
-                               "[{}] is our own localnet port. Skipping", addr_);
+                        debug!(
+                            target: "net::hosts::filter_addresses",
+                            "[{}] is our own localnet port. Skipping", addr_,
+                        );
                         continue 'addr_loop
                     }
                 }
@@ -1168,9 +1189,11 @@ impl Hosts {
             // Filter non-global ranges if we're not allowing localnet.
             // Should never be allowed in production, so we don't really care
             // about some of them (e.g. 0.0.0.0, or broadcast, etc.).
-            if !localnet && self.is_local_host(addr) {
-                debug!(target: "net::hosts::filter_addresses()",
-                       "[{}] Filtering non-global ranges", addr_);
+            if !settings.localnet && self.is_local_host(addr) {
+                debug!(
+                    target: "net::hosts::filter_addresses",
+                    "[{}] Filtering non-global ranges", addr_,
+                );
                 continue
             }
 
@@ -1182,8 +1205,10 @@ impl Hosts {
                     if tor_hscrypto::pk::HsId::from_str(host_str).is_err() {
                         continue
                     }
-                    trace!(target: "net::hosts::filter_addresses()",
-                           "[Tor] Valid: {}", host_str);
+                    trace!(
+                        target: "net::hosts::filter_addresses",
+                        "[Tor] Valid: {}", host_str,
+                    );
                 }
 
                 #[cfg(feature = "p2p-nym")]
@@ -1191,8 +1216,10 @@ impl Hosts {
 
                 #[cfg(feature = "p2p-tcp")]
                 "tcp" | "tcp+tls" => {
-                    trace!(target: "net::hosts::filter_addresses()",
-                           "[TCP] Valid: {}", host_str);
+                    trace!(
+                        target: "net::hosts::filter_addresses",
+                        "[TCP] Valid: {}", host_str,
+                    );
                 }
 
                 _ => continue,
@@ -1203,7 +1230,7 @@ impl Hosts {
             // We will personally ignore this peer but still send it to others in
             // Protocol Addr to ensure all transports get propagated.
             if !settings.allowed_transports.contains(&addr_.scheme().to_string()) ||
-                (!ipv6_available && self.is_ipv6(addr_.clone()))
+                (!self.ipv6_available.load(Ordering::SeqCst) && self.is_ipv6(addr_.clone()))
             {
                 self.container.store_or_update(HostColor::Dark, addr_.clone(), *last_seen);
                 self.container.sort_by_last_seen(HostColor::Dark as usize);
@@ -1219,7 +1246,7 @@ impl Hosts {
                 self.container.contains(HostColor::White as usize, addr_) ||
                 self.container.contains(HostColor::Grey as usize, addr_)
             {
-                debug!(target: "net::hosts::filter_addresses()", "[{}] exists! Skipping", addr_);
+                debug!(target: "net::hosts::filter_addresses", "[{}] exists! Skipping", addr_);
                 continue
             }
 
@@ -1346,9 +1373,7 @@ impl Hosts {
 
 #[cfg(test)]
 mod tests {
-    use std::time::UNIX_EPOCH;
-
-    use super::{super::settings::Settings, *};
+    use super::*;
     use crate::system::sleep;
 
     #[test]
@@ -1361,7 +1386,7 @@ mod tests {
             ],
             ..Default::default()
         };
-        let hosts = Hosts::new(Arc::new(settings.clone()));
+        let hosts = Hosts::new(Arc::new(AsyncRwLock::new(settings)));
 
         let local_hosts: Vec<Url> = vec![
             Url::parse("tcp://localhost").unwrap(),
@@ -1389,7 +1414,7 @@ mod tests {
     #[test]
     fn test_is_ipv6() {
         let settings = Settings { ..Default::default() };
-        let hosts = Hosts::new(Arc::new(settings.clone()));
+        let hosts = Hosts::new(Arc::new(AsyncRwLock::new(settings)));
 
         let ipv6_hosts: Vec<Url> = vec![
             Url::parse("tcp+tls://[::1]").unwrap(),
@@ -1415,8 +1440,8 @@ mod tests {
     #[test]
     fn test_block_all_ports() {
         let settings = Settings { ..Default::default() };
+        let hosts = Hosts::new(Arc::new(AsyncRwLock::new(settings)));
 
-        let hosts = Hosts::new(Arc::new(settings.clone()));
         let blacklist1 = Url::parse("tcp+tls://nietzsche.king:333").unwrap();
         let blacklist2 = Url::parse("tcp+tls://agorism.xyz").unwrap();
 
@@ -1432,8 +1457,8 @@ mod tests {
         let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
 
         let settings = Settings { ..Default::default() };
+        let hosts = Hosts::new(Arc::new(AsyncRwLock::new(settings)));
 
-        let hosts = Hosts::new(Arc::new(settings.clone()));
         let grey_hosts = vec![
             Url::parse("tcp://localhost:3921").unwrap(),
             Url::parse("tor://[::1]:21481").unwrap(),
@@ -1479,7 +1504,7 @@ mod tests {
     fn test_get_last() {
         smol::block_on(async {
             let settings = Settings { ..Default::default() };
-            let hosts = Hosts::new(Arc::new(settings.clone()));
+            let hosts = Hosts::new(Arc::new(AsyncRwLock::new(settings)));
 
             // Build up a hostlist
             for i in 0..10 {

src/net/p2p.rs

@@ -26,6 +26,7 @@ use futures_rustls::rustls::crypto::{ring, CryptoProvider};
 use log::{debug, error, info, warn};
 use smol::{
     fs::{self, unix::PermissionsExt},
+    lock::RwLock as AsyncRwLock,
     stream::StreamExt,
 };
 use url::Url;
@@ -60,7 +61,7 @@ pub struct P2p {
     /// Protocol registry
     protocol_registry: ProtocolRegistry,
     /// P2P network settings
-    settings: Arc<Settings>,
+    settings: Arc<AsyncRwLock<Settings>>,
     /// Reference to configured [`ManualSession`]
     session_manual: ManualSessionPtr,
     /// Reference to configured [`InboundSession`]
@@ -87,8 +88,6 @@ impl P2p {
     /// Creates a weak pointer to self that is used by all sessions to access the
     /// p2p parent class.
     pub async fn new(settings: Settings, executor: ExecutorPtr) -> Result<P2pPtr> {
-        let settings = Arc::new(settings);
-
         // Create the datastore
         if let Some(ref datastore) = settings.datastore {
             let datastore = expand_path(datastore)?;
@@ -99,9 +98,12 @@ impl P2p {
         // Register a CryptoProvider for rustls
         let _ = CryptoProvider::install_default(ring::default_provider());
 
+        // Wrap the Settings into an Arc<RwLock>
+        let settings = Arc::new(AsyncRwLock::new(settings));
+
         let self_ = Arc::new(Self {
             executor,
-            hosts: Hosts::new(settings.clone()),
+            hosts: Hosts::new(Arc::clone(&settings)),
             protocol_registry: ProtocolRegistry::new(),
             settings,
             session_manual: ManualSession::new(),
@@ -114,11 +116,11 @@ impl P2p {
             dnet_publisher: Publisher::new(),
         });
 
-        self_.session_manual.p2p.init(self_.clone());
         self_.session_inbound.p2p.init(self_.clone());
+        self_.session_manual.p2p.init(self_.clone());
+        self_.session_seedsync.p2p.init(self_.clone());
         self_.session_outbound.p2p.init(self_.clone());
         self_.session_refine.p2p.init(self_.clone());
-        self_.session_seedsync.p2p.init(self_.clone());
 
         register_default_protocols(self_.clone()).await;
 
@@ -127,19 +129,18 @@ impl P2p {
 
     /// Starts inbound, outbound, and manual sessions.
     pub async fn start(self: Arc<Self>) -> Result<()> {
-        debug!(target: "net::p2p::start()", "P2P::start() [BEGIN]");
-        info!(target: "net::p2p::start()", "[P2P] Starting P2P subsystem");
-
-        // First attempt any set manual connections
-        self.session_manual().start().await;
+        debug!(target: "net::p2p::start", "P2P::start() [BEGIN]");
+        info!(target: "net::p2p::start", "[P2P] Starting P2P subsystem");
 
         // Start the inbound session
         if let Err(err) = self.session_inbound().start().await {
-            error!(target: "net::p2p::start()", "Failed to start inbound session!: {}", err);
-            self.session_manual().stop().await;
+            error!(target: "net::p2p::start", "Failed to start inbound session!: {}", err);
             return Err(err)
         }
 
+        // Start the manual session
+        self.session_manual().start().await;
+
         // Start the seedsync session. Seed connections will not
         // activate yet- they wait for a call to notify().
         self.session_seedsync().start().await;
@@ -150,7 +151,7 @@ impl P2p {
         // Start the refine session
         self.session_refine().start().await;
 
-        info!(target: "net::p2p::start()", "[P2P] P2P subsystem started");
+        info!(target: "net::p2p::start", "[P2P] P2P subsystem started");
         Ok(())
     }
 
@@ -223,8 +224,8 @@ impl P2p {
     }
 
     /// Return an atomic pointer to the set network settings
-    pub fn settings(&self) -> Arc<Settings> {
-        self.settings.clone()
+    pub fn settings(&self) -> Arc<AsyncRwLock<Settings>> {
+        Arc::clone(&self.settings)
     }
 
     /// Return an atomic pointer to the list of hosts

src/net/protocol/protocol_address.rs

@@ -20,7 +20,7 @@ use std::{sync::Arc, time::UNIX_EPOCH};
 
 use async_trait::async_trait;
 use log::debug;
-use smol::Executor;
+use smol::{lock::RwLock as AsyncRwLock, Executor};
 
 use super::{
     super::{
@@ -62,7 +62,7 @@ pub struct ProtocolAddress {
     addrs_sub: MessageSubscription<AddrsMessage>,
     get_addrs_sub: MessageSubscription<GetAddrsMessage>,
     hosts: HostsPtr,
-    settings: Arc<Settings>,
+    settings: Arc<AsyncRwLock<Settings>>,
     jobsman: ProtocolJobsManagerPtr,
 }
 
@@ -79,9 +79,6 @@ impl ProtocolAddress {
     /// and a get-address subscription and adds them to the address protocol
     /// instance.
     pub async fn init(channel: ChannelPtr, p2p: P2pPtr) -> ProtocolBasePtr {
-        let settings = p2p.settings();
-        let hosts = p2p.hosts();
-
         // Creates a subscription to address message
         let addrs_sub =
             channel.subscribe_msg::<AddrsMessage>().await.expect("Missing addrs dispatcher!");
@@ -94,9 +91,9 @@ impl ProtocolAddress {
             channel: channel.clone(),
             addrs_sub,
             get_addrs_sub,
-            hosts,
+            hosts: p2p.hosts(),
             jobsman: ProtocolJobsManager::new(PROTO_NAME, channel),
-            settings,
+            settings: p2p.settings(),
         })
     }
 
@@ -215,35 +212,48 @@ impl ProtocolAddress {
     /// last_seen field to now.
     async fn send_my_addrs(self: Arc<Self>) -> Result<()> {
         debug!(
-            target: "net::protocol_address::send_my_addrs()",
+            target: "net::protocol_address::send_my_addrs",
             "[START] channel address={}", self.channel.address(),
         );
 
         let type_id = self.channel.session_type_id();
         if type_id != SESSION_OUTBOUND {
-            debug!(target: "net::protocol_address::send_my_addrs()",
-            "Not an outbound session. Stopping");
+            debug!(
+                target: "net::protocol_address::send_my_addrs",
+                "Not an outbound session. Stopping",
+            );
             return Ok(())
         }
 
-        if self.settings.external_addrs.is_empty() {
-            debug!(target: "net::protocol_address::send_my_addrs()",
-            "External addr not configured. Stopping");
+        let external_addrs = self.settings.read().await.external_addrs.clone();
+
+        if external_addrs.is_empty() {
+            debug!(
+                target: "net::protocol_address::send_my_addrs",
+                "External addr not configured. Stopping",
+            );
             return Ok(())
         }
 
         let mut addrs = vec![];
 
-        for addr in self.settings.external_addrs.clone() {
+        for addr in external_addrs {
             let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
             addrs.push((addr, last_seen));
         }
-        debug!(target: "net::protocol_address::send_my_addrs()",
-        "Broadcasting {} addresses", addrs.len());
+
+        debug!(
+            target: "net::protocol_address::send_my_addrs",
+            "Broadcasting {} addresses", addrs.len(),
+        );
+
         let ext_addr_msg = AddrsMessage { addrs };
         self.channel.send(&ext_addr_msg).await?;
-        debug!(target: "net::protocol_address::send_my_addrs()",
-        "[END] channel address={}", self.channel.address());
+
+        debug!(
+            target: "net::protocol_address::send_my_addrs",
+            "[END] channel address={}", self.channel.address(),
+        );
 
         Ok(())
     }
@@ -256,8 +266,15 @@ impl ProtocolBase for ProtocolAddress {
     /// and get address protocols on the protocol task manager. Then send
     /// get-address msg.
     async fn start(self: Arc<Self>, ex: Arc<Executor<'_>>) -> Result<()> {
-        debug!(target: "net::protocol_address::start()",
-        "START => address={}", self.channel.address());
+        debug!(
+            target: "net::protocol_address::start()",
+            "START => address={}", self.channel.address(),
+        );
+
+        let settings = self.settings.read().await;
+        let outbound_connections = settings.outbound_connections;
+        let allowed_transports = settings.allowed_transports.clone();
+        drop(settings);
 
         self.jobsman.clone().start(ex.clone());
 
@@ -268,14 +285,14 @@ impl ProtocolBase for ProtocolAddress {
         self.jobsman.spawn(self.clone().handle_receive_get_addrs(), ex).await;
 
         // Send get_address message.
-        let get_addrs = GetAddrsMessage {
-            max: self.settings.outbound_connections as u32,
-            transports: self.settings.allowed_transports.clone(),
-        };
+        let get_addrs =
+            GetAddrsMessage { max: outbound_connections as u32, transports: allowed_transports };
         self.channel.send(&get_addrs).await?;
 
-        debug!(target: "net::protocol_address::start()",
-        "END => address={}", self.channel.address());
+        debug!(
+            target: "net::protocol_address::start()",
+            "END => address={}", self.channel.address(),
+        );
 
         Ok(())
     }

src/net/protocol/protocol_ping.rs

@@ -24,7 +24,7 @@ use std::{
 use async_trait::async_trait;
 use log::{debug, error, warn};
 use rand::{rngs::OsRng, Rng};
-use smol::Executor;
+use smol::{lock::RwLock as AsyncRwLock, Executor};
 
 use super::{
     super::{
@@ -47,7 +47,7 @@ pub struct ProtocolPing {
     channel: ChannelPtr,
     ping_sub: MessageSubscription<PingMessage>,
     pong_sub: MessageSubscription<PongMessage>,
-    settings: Arc<Settings>,
+    settings: Arc<AsyncRwLock<Settings>>,
     jobsman: ProtocolJobsManagerPtr,
 }
 
@@ -56,8 +56,6 @@ const PROTO_NAME: &str = "ProtocolPing";
 impl ProtocolPing {
     /// Create a new ping-pong protocol.
     pub async fn init(channel: ChannelPtr, p2p: P2pPtr) -> ProtocolBasePtr {
-        let settings = p2p.settings();
-
         // Creates a subscription to ping message
         let ping_sub =
             channel.subscribe_msg::<PingMessage>().await.expect("Missing ping dispatcher!");
@@ -70,7 +68,7 @@ impl ProtocolPing {
             channel: channel.clone(),
             ping_sub,
             pong_sub,
-            settings,
+            settings: p2p.settings(),
             jobsman: ProtocolJobsManager::new(PROTO_NAME, channel),
         })
     }
@@ -86,6 +84,11 @@ impl ProtocolPing {
         );
 
         loop {
+            let settings = self.settings.read().await;
+            let outbound_connect_timeout = settings.outbound_connect_timeout;
+            let channel_heartbeat_interval = settings.channel_heartbeat_interval;
+            drop(settings);
+
             // Create a random nonce.
             let nonce = Self::random_nonce();
 
@@ -98,7 +101,7 @@ impl ProtocolPing {
 
             // Wait for pong, check nonce matches.
             let pong_msg = match timeout(
-                Duration::from_secs(self.settings.outbound_connect_timeout),
+                Duration::from_secs(outbound_connect_timeout),
                 self.pong_sub.receive(),
             )
             .await
@@ -138,7 +141,7 @@ impl ProtocolPing {
             );
 
             // Sleep until next heartbeat
-            sleep(self.settings.channel_heartbeat_interval).await;
+            sleep(channel_heartbeat_interval).await;
         }
     }
 

src/net/protocol/protocol_seed.rs

@@ -20,7 +20,7 @@ use std::{sync::Arc, time::UNIX_EPOCH};
 
 use async_trait::async_trait;
 use log::debug;
-use smol::Executor;
+use smol::{lock::RwLock as AsyncRwLock, Executor};
 
 use super::{
     super::{
@@ -39,7 +39,7 @@ use crate::Result;
 pub struct ProtocolSeed {
     channel: ChannelPtr,
     hosts: HostsPtr,
-    settings: Arc<Settings>,
+    settings: Arc<AsyncRwLock<Settings>>,
     addr_sub: MessageSubscription<AddrsMessage>,
 }
 
@@ -48,41 +48,50 @@ const PROTO_NAME: &str = "ProtocolSeed";
 impl ProtocolSeed {
     /// Create a new seed protocol.
     pub async fn init(channel: ChannelPtr, p2p: P2pPtr) -> ProtocolBasePtr {
-        let hosts = p2p.hosts();
-        let settings = p2p.settings();
-
         // Create a subscription to address message
         let addr_sub =
             channel.subscribe_msg::<AddrsMessage>().await.expect("Missing addr dispatcher!");
 
-        Arc::new(Self { channel, hosts, settings, addr_sub })
+        Arc::new(Self { channel, hosts: p2p.hosts(), settings: p2p.settings(), addr_sub })
     }
 
     /// Send our own external addresses over a channel. Set the
     /// last_seen field to now.
     pub async fn send_my_addrs(&self) -> Result<()> {
-        debug!(target: "net::protocol_seed::send_my_addrs()",
-        "[START] channel address={}", self.channel.address());
+        debug!(
+            target: "net::protocol_seed::send_my_addrs",
+            "[START] channel address={}", self.channel.address(),
+        );
+
+        let external_addrs = self.settings.read().await.external_addrs.clone();
 
-        if self.settings.external_addrs.is_empty() {
-            debug!(target: "net::protocol_seed::send_my_addrs()",
-            "External address is not configured. Stopping");
+        if external_addrs.is_empty() {
+            debug!(
+                target: "net::protocol_seed::send_my_addrs",
+                "External address is not configured. Stopping",
+            );
             return Ok(())
         }
 
         let mut addrs = vec![];
 
-        for addr in self.settings.external_addrs.clone() {
+        for addr in external_addrs {
             let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
             addrs.push((addr, last_seen));
         }
 
-        debug!(target: "net::protocol_seed::send_my_addrs()",
-        "Broadcasting {} addresses", addrs.len());
+        debug!(
+            target: "net::protocol_seed::send_my_addrs",
+            "Broadcasting {} addresses", addrs.len(),
+        );
+
         let ext_addr_msg = AddrsMessage { addrs };
         self.channel.send(&ext_addr_msg).await?;
-        debug!(target: "net::protocol_seed::send_my_addrs()",
-        "[END] channel address={}", self.channel.address());
+
+        debug!(
+            target: "net::protocol_seed::send_my_addrs",
+            "[END] channel address={}", self.channel.address(),
+        );
 
         Ok(())
     }
@@ -100,11 +109,14 @@ impl ProtocolBase for ProtocolSeed {
         // Send own address to the seed server
         self.send_my_addrs().await?;
 
+        let settings = self.settings.read().await;
+        let outbound_connections = settings.outbound_connections;
+        let allowed_transports = settings.allowed_transports.clone();
+        drop(settings);
+
         // Send get address message
-        let get_addr = GetAddrsMessage {
-            max: self.settings.outbound_connections as u32,
-            transports: self.settings.allowed_transports.clone(),
-        };
+        let get_addr =
+            GetAddrsMessage { max: outbound_connections as u32, transports: allowed_transports };
         self.channel.send(&get_addr).await?;
 
         // Receive addresses

src/net/protocol/protocol_version.rs

@@ -26,7 +26,7 @@ use futures::{
     pin_mut,
 };
 use log::{debug, error};
-use smol::{Executor, Timer};
+use smol::{lock::RwLock as AsyncRwLock, Executor, Timer};
 
 use super::super::{
     channel::ChannelPtr,
@@ -42,13 +42,15 @@ pub struct ProtocolVersion {
     channel: ChannelPtr,
     version_sub: MessageSubscription<VersionMessage>,
     verack_sub: MessageSubscription<VerackMessage>,
-    settings: Arc<Settings>,
+    settings: Arc<AsyncRwLock<Settings>>,
 }
 
 impl ProtocolVersion {
     /// Create a new version protocol. Makes a version and version ack
     /// subscription, then adds them to a version protocol instance.
-    pub async fn new(channel: ChannelPtr, settings: Arc<Settings>) -> Arc<Self> {
+    // TODO: This function takes settings as a param, however, it is also reachable through Channel.
+    //       Maybe we want to navigate towards Settings through channel->session->p2p->settings
+    pub async fn new(channel: ChannelPtr, settings: Arc<AsyncRwLock<Settings>>) -> Arc<Self> {
         // Creates a version subscription
         let version_sub =
             channel.subscribe_msg::<VersionMessage>().await.expect("Missing version dispatcher!");
@@ -65,7 +67,8 @@ impl ProtocolVersion {
     /// version ack.
     pub async fn run(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
         debug!(target: "net::protocol_version::run()", "START => address={}", self.channel.address());
-        let timeout = Timer::after(Duration::from_secs(self.settings.channel_handshake_timeout));
+        let timeout =
+            Timer::after(Duration::from_secs(self.settings.read().await.channel_handshake_timeout));
         let version = self.clone().exchange_versions(executor);
 
         pin_mut!(timeout);
@@ -147,13 +150,19 @@ impl ProtocolVersion {
             "START => address={}", self.channel.address(),
         );
 
+        let settings = self.settings.read().await;
+        let node_id = settings.node_id.clone();
+        let app_version = settings.app_version.clone();
+        let external_addrs = settings.external_addrs.clone();
+        drop(settings);
+
         let version = VersionMessage {
-            node_id: self.settings.node_id.clone(),
-            version: self.settings.app_version.clone(),
+            node_id,
+            version: app_version.clone(),
             timestamp: UNIX_EPOCH.elapsed().unwrap().as_secs(),
             connect_recv_addr: self.channel.connect_addr().clone(),
             resolve_recv_addr: self.channel.resolve_addr().clone(),
-            ext_send_addr: self.settings.external_addrs.clone(),
+            ext_send_addr: external_addrs,
             /* NOTE: `features` is a list of enabled features in the
             format Vec<(service, version)>. In the future, Protocols will
             add their own data to this field when they are attached.*/
@@ -168,12 +177,12 @@ impl ProtocolVersion {
         debug!(
             target: "net::protocol_version::send_version()",
             "App version: {}, Recv version: {}",
-            self.settings.app_version, verack_msg.app_version,
+            app_version, verack_msg.app_version,
         );
 
         // MAJOR and MINOR should be the same.
-        if self.settings.app_version.major != verack_msg.app_version.major ||
-            self.settings.app_version.minor != verack_msg.app_version.minor
+        if app_version.major != verack_msg.app_version.major ||
+            app_version.minor != verack_msg.app_version.minor
         {
             error!(
                 target: "net::protocol_version::send_version()",
@@ -206,7 +215,7 @@ impl ProtocolVersion {
         self.channel.set_version(version).await;
 
         // Send verack
-        let verack = VerackMessage { app_version: self.settings.app_version.clone() };
+        let verack = VerackMessage { app_version: self.settings.read().await.app_version.clone() };
         self.channel.send(&verack).await?;
 
         debug!(

src/net/session/inbound_session.rs

@@ -67,7 +67,9 @@ impl InboundSession {
     /// if the addresses are not configured. Then runs the channel subscription
     /// loop.
     pub async fn start(self: Arc<Self>) -> Result<()> {
-        if self.p2p().settings().inbound_addrs.is_empty() {
+        let inbound_addrs = self.p2p().settings().read().await.inbound_addrs.clone();
+
+        if inbound_addrs.is_empty() {
             info!(target: "net::inbound_session", "[P2P] Not configured for inbound connections.");
             return Ok(())
         }
@@ -77,7 +79,7 @@ impl InboundSession {
         // Activate mutex lock on accept tasks.
         let mut accept_tasks = self.accept_tasks.lock().await;
 
-        for (index, accept_addr) in self.p2p().settings().inbound_addrs.iter().enumerate() {
+        for (index, accept_addr) in inbound_addrs.iter().enumerate() {
             // First initialize an Acceptor and its Subscriber.
             let parent = Arc::downgrade(&self);
             let acceptor = Acceptor::new(parent);
@@ -111,8 +113,8 @@ impl InboundSession {
 
     /// Stops the inbound session.
     pub async fn stop(&self) {
-        if self.p2p().settings().inbound_addrs.is_empty() {
-            info!(target: "net::inbound_session", "[P2P] Not configured for inbound connections.");
+        if self.p2p().settings().read().await.inbound_addrs.is_empty() {
+            info!(target: "net::inbound_session", "[P2P] Stopping inbound session.");
             return
         }
 

src/net/session/manual_session.rs

@@ -36,7 +36,7 @@ use std::sync::{Arc, Weak};
 use async_trait::async_trait;
 use futures::stream::{FuturesUnordered, StreamExt};
 use log::{debug, error, info, warn};
-use smol::lock::Mutex;
+use smol::lock::{Mutex as AsyncMutex, RwLock as AsyncRwLock};
 use url::Url;
 
 use super::{
@@ -57,13 +57,13 @@ pub type ManualSessionPtr = Arc<ManualSession>;
 /// Defines manual connections session.
 pub struct ManualSession {
     pub(in crate::net) p2p: LazyWeak<P2p>,
-    slots: Mutex<Vec<Arc<Slot>>>,
+    slots: AsyncMutex<Vec<Arc<Slot>>>,
 }
 
 impl ManualSession {
     /// Create a new manual session.
     pub fn new() -> ManualSessionPtr {
-        Arc::new(Self { p2p: LazyWeak::new(), slots: Mutex::new(Vec::new()) })
+        Arc::new(Self { p2p: LazyWeak::new(), slots: AsyncMutex::new(Vec::new()) })
     }
 
     pub(crate) async fn start(self: Arc<Self>) {
@@ -76,7 +76,7 @@ impl ManualSession {
 
         // Initialize a slot for each configured peer.
         // Connections will be started by not yet activated.
-        for peer in &self.p2p().settings().peers {
+        for peer in &self.p2p().settings().read().await.peers {
             let slot = Slot::new(self_.clone(), peer.clone(), self.p2p().settings());
             futures.push(slot.clone().start());
             slots.push(slot);
@@ -117,7 +117,11 @@ struct Slot {
 }
 
 impl Slot {
-    fn new(session: Weak<ManualSession>, addr: Url, settings: Arc<Settings>) -> Arc<Self> {
+    fn new(
+        session: Weak<ManualSession>,
+        addr: Url,
+        settings: Arc<AsyncRwLock<Settings>>,
+    ) -> Arc<Self> {
         Arc::new(Self {
             addr,
             process: StoppableTask::new(),
@@ -156,11 +160,18 @@ impl Slot {
                 self.addr, attempts
             );
 
+            let settings = self.p2p().settings().read_arc().await;
+            let seeds = settings.seeds.clone();
+            let outbound_connect_timeout = settings.outbound_connect_timeout;
+            drop(settings);
+
             // Do not establish a connection to a host that is also configured as a seed.
             // This indicates a user misconfiguration.
-            if self.p2p().settings().seeds.contains(&self.addr) {
-                error!(target: "net::manual_session", 
-                       "[P2P] Suspending manual connection to seed [{}]", self.addr.clone());
+            if seeds.contains(&self.addr) {
+                error!(
+                    target: "net::manual_session",
+                    "[P2P] Suspending manual connection to seed [{}]", self.addr.clone(),
+                );
                 return Ok(())
             }
 
@@ -210,9 +221,9 @@ impl Slot {
             info!(
                 target: "net::manual_session",
                 "[P2P] Waiting {} seconds until next manual outbound connection attempt [{}]",
-                self.p2p().settings().outbound_connect_timeout, self.addr,
+                outbound_connect_timeout, self.addr,
             );
-            sleep(self.p2p().settings().outbound_connect_timeout).await;
+            sleep(outbound_connect_timeout).await;
         }
     }
 

src/net/session/mod.rs

@@ -128,8 +128,11 @@ pub trait Session: Sync {
 
         // Perform the handshake protocol
         let protocol_version = ProtocolVersion::new(channel.clone(), p2p.settings().clone()).await;
-        debug!(target: "net::session::register_channel()",
-        "Performing handshake protocols {}", channel.clone().address());
+        debug!(
+            target: "net::session::register_channel()",
+            "Performing handshake protocols {}", channel.clone().address(),
+        );
+
         let handshake_task =
             self.perform_handshake_protocols(protocol_version, channel.clone(), executor.clone());
 

src/net/session/outbound_session.rs

@@ -82,8 +82,9 @@ impl OutboundSession {
 
     /// Start the outbound session. Runs the channel connect loop.
     pub(crate) async fn start(self: Arc<Self>) {
-        let n_slots = self.p2p().settings().outbound_connections;
+        let n_slots = self.p2p().settings().read().await.outbound_connections;
         info!(target: "net::outbound_session", "[P2P] Starting {} outbound connection slots.", n_slots);
+
         // Activate mutex lock on connection slots.
         let mut slots = self.slots.lock().await;
 
@@ -163,6 +164,7 @@ struct Slot {
 impl Slot {
     fn new(session: Weak<OutboundSession>, slot: u32) -> Arc<Self> {
         let settings = session.upgrade().unwrap().p2p().settings();
+
         Arc::new(Self {
             slot,
             process: StoppableTask::new(),
@@ -188,6 +190,7 @@ impl Slot {
             ex,
         );
     }
+
     async fn stop(self: Arc<Self>) {
         self.connector.stop();
         self.process.stop().await;
@@ -203,18 +206,23 @@ impl Slot {
     /// and healthy since we require the network retains some unreliable
     /// connections. A network that purely favors uptime over unreliable
     /// connections may be vulnerable to sybil by attackers with good uptime.
-    fn fetch_addrs(&self) -> Option<(Url, u64)> {
+    async fn fetch_addrs(&self) -> Option<(Url, u64)> {
         let hosts = self.p2p().hosts();
         let slot = self.slot as usize;
-        let settings = self.p2p().settings();
         let container = &self.p2p().hosts().container;
 
+        // Acquire Settings read lock
+        let settings = self.p2p().settings().read_arc().await;
+
         let white_count = (settings.white_connect_percent * settings.outbound_connections) / 100;
         let gold_count = settings.gold_connect_count;
 
-        let transports = &settings.allowed_transports;
+        let transports = settings.allowed_transports.clone();
         let transport_mixing = settings.transport_mixing;
-        let preference_strict = &settings.slot_preference_strict;
+        let preference_strict = settings.slot_preference_strict;
+
+        // Drop Settings read lock
+        drop(settings);
 
         let grey_only = hosts.container.is_empty(HostColor::White) &&
             hosts.container.is_empty(HostColor::Gold) &&
@@ -223,15 +231,16 @@ impl Slot {
         // If we only have grey entries, select from the greylist. Otherwise,
         // use the preference defined in settings.
         let addrs = if grey_only && !preference_strict {
-            container.fetch(HostColor::Grey, transports, transport_mixing)
+            container.fetch(HostColor::Grey, &transports, transport_mixing)
         } else if slot < gold_count {
-            container.fetch(HostColor::Gold, transports, transport_mixing)
+            container.fetch(HostColor::Gold, &transports, transport_mixing)
         } else if slot < white_count {
-            container.fetch(HostColor::White, transports, transport_mixing)
+            container.fetch(HostColor::White, &transports, transport_mixing)
         } else {
-            container.fetch(HostColor::Grey, transports, transport_mixing)
+            container.fetch(HostColor::Grey, &transports, transport_mixing)
         };
-        hosts.check_addrs(addrs)
+
+        hosts.check_addrs(addrs).await
     }
 
     // We first try to make connections to the addresses on our gold list. We then find some
@@ -267,7 +276,7 @@ impl Slot {
                 continue
             }
 
-            let addr = if let Some(addr) = self.fetch_addrs() {
+            let addr = if let Some(addr) = self.fetch_addrs().await {
                 debug!(target: "net::outbound_session::run()", "Fetched addr={}, slot #{}", addr.0,
                 self.slot);
                 addr
@@ -515,7 +524,15 @@ impl PeerDiscoveryBase for PeerDiscovery {
             // wait to be woken up by notify()
             let sleep_was_instant = self.wait().await;
 
-            let p2p = self.p2p();
+            // Read the current P2P settings
+            let settings = self.p2p().settings().read_arc().await;
+            let outbound_peer_discovery_cooloff_time =
+                settings.outbound_peer_discovery_cooloff_time;
+            let outbound_peer_discovery_attempt_time =
+                settings.outbound_peer_discovery_attempt_time;
+            let outbound_connections = settings.outbound_connections;
+            let allowed_transports = settings.allowed_transports.clone();
+            drop(settings);
 
             if sleep_was_instant {
                 // Try again
@@ -537,13 +554,13 @@ impl PeerDiscoveryBase for PeerDiscovery {
                     state: "sleep",
                 });
 
-                sleep(p2p.settings().outbound_peer_discovery_cooloff_time).await;
+                sleep(outbound_peer_discovery_cooloff_time).await;
                 current_attempt = 1;
             }
 
             // First 2 times try sending GetAddr to the network.
             // 3rd time do a seed sync.
-            if p2p.is_connected() && current_attempt <= 2 {
+            if self.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.
 
@@ -559,16 +576,17 @@ impl PeerDiscoveryBase for PeerDiscovery {
                 });
 
                 let get_addrs = GetAddrsMessage {
-                    max: p2p.settings().outbound_connections as u32,
-                    transports: p2p.settings().allowed_transports.clone(),
+                    max: outbound_connections as u32,
+                    transports: allowed_transports,
                 };
-                p2p.broadcast(&get_addrs).await;
+
+                self.p2p().broadcast(&get_addrs).await;
 
                 // Wait for a hosts store update event
                 let store_sub = self.p2p().hosts().subscribe_store().await;
 
                 let result = timeout(
-                    Duration::from_secs(p2p.settings().outbound_peer_discovery_attempt_time),
+                    Duration::from_secs(outbound_peer_discovery_attempt_time),
                     store_sub.receive(),
                 )
                 .await;
@@ -607,9 +625,9 @@ impl PeerDiscoveryBase for PeerDiscovery {
                     state: "seed",
                 });
 
-                p2p.clone().seed().await;
+                self.p2p().seed().await;
 
-                if p2p.clone().session_seedsync().failed().await {
+                if self.p2p().session_seedsync().failed().await {
                     error!(
                         target: "net::outbound_session::peer_discovery()",
                         "[P2P] Network reseed failed!"
@@ -621,7 +639,7 @@ impl PeerDiscoveryBase for PeerDiscovery {
             self.session().wakeup_slots().await;
 
             // Give some time for new connections to be established
-            sleep(p2p.settings().outbound_peer_discovery_attempt_time).await;
+            sleep(outbound_peer_discovery_attempt_time).await;
         }
     }
 

src/net/session/refine_session.rs

@@ -71,7 +71,7 @@ impl RefineSession {
 
     /// Start the refinery and self handshake processes.
     pub(crate) async fn start(self: Arc<Self>) {
-        if let Some(ref hostlist) = self.p2p().settings().hostlist {
+        if let Some(ref hostlist) = self.p2p().settings().read().await.hostlist {
             match self.p2p().hosts().container.load_all(hostlist) {
                 Ok(()) => {
                     debug!(target: "net::refine_session::start", "Load hosts successful!");
@@ -82,7 +82,7 @@ impl RefineSession {
             }
         }
 
-        match self.p2p().hosts().import_blacklist() {
+        match self.p2p().hosts().import_blacklist().await {
             Ok(()) => {
                 debug!(target: "net::refine_session::start", "Import blacklist successful!");
             }
@@ -101,7 +101,7 @@ impl RefineSession {
         debug!(target: "net::refine_session", "Stopping refinery process");
         self.refinery.clone().stop().await;
 
-        if let Some(ref hostlist) = self.p2p().settings().hostlist {
+        if let Some(ref hostlist) = self.p2p().settings().read().await.hostlist {
             match self.p2p().hosts().container.save_all(hostlist) {
                 Ok(()) => {
                     debug!(target: "net::refine_session::stop()", "Save hosts successful!");
@@ -224,11 +224,17 @@ impl GreylistRefinery {
     // Randomly select a peer on the greylist and probe it. This method will remove from the
     // greylist and store on the whitelist providing the peer is responsive.
     async fn run(self: Arc<Self>) {
-        let p2p = self.p2p();
-        let hosts = p2p.hosts();
-        let settings = p2p.settings();
+        let hosts = self.p2p().hosts();
+
         loop {
-            sleep(settings.greylist_refinery_interval).await;
+            // Acquire read lock on P2P settings and load necessary settings
+            let settings = self.p2p().settings().read_arc().await;
+            let greylist_refinery_interval = settings.greylist_refinery_interval;
+            let time_with_no_connections = settings.time_with_no_connections;
+            let allowed_transports = settings.allowed_transports.clone();
+            drop(settings);
+
+            sleep(greylist_refinery_interval).await;
 
             if hosts.container.is_empty(HostColor::Grey) {
                 debug!(target: "net::refinery",
@@ -239,12 +245,12 @@ impl GreylistRefinery {
 
             // Pause the refinery if we've had zero connections for longer than the configured
             // limit.
-            let offline_limit = Duration::from_secs(settings.time_with_no_connections);
+            let offline_limit = Duration::from_secs(time_with_no_connections);
 
             let offline_timer =
                 { Instant::now().duration_since(*hosts.last_connection.lock().unwrap()) };
 
-            if !p2p.is_connected() && offline_timer >= offline_limit {
+            if !self.p2p().is_connected() && offline_timer >= offline_limit {
                 warn!(target: "net::refinery", "No connections for {}s. GreylistRefinery paused.",
                           offline_timer.as_secs());
 
@@ -261,10 +267,7 @@ impl GreylistRefinery {
             }
 
             // Only attempt to refine peers that match our transports.
-            match hosts
-                .container
-                .fetch_random_with_schemes(HostColor::Grey, &settings.allowed_transports)
-            {
+            match hosts.container.fetch_random_with_schemes(HostColor::Grey, &allowed_transports) {
                 Some((entry, _)) => {
                     let url = &entry.0;
                     let last_seen = &entry.1;
@@ -275,7 +278,7 @@ impl GreylistRefinery {
                         continue
                     }
 
-                    if !self.session().handshake_node(url.clone(), p2p.clone()).await {
+                    if !self.session().handshake_node(url.clone(), self.p2p().clone()).await {
                         debug!(
                             target: "net::refinery",
                             "Peer {} handshake failed. Removed from greylist", url,

src/net/session/seedsync_session.rs

@@ -50,7 +50,7 @@ use std::sync::{
 use async_trait::async_trait;
 use futures::stream::{FuturesUnordered, StreamExt};
 use log::{debug, info, warn};
-use smol::lock::Mutex;
+use smol::lock::{Mutex as AsyncMutex, RwLock as AsyncRwLock};
 use url::Url;
 
 use super::{
@@ -73,13 +73,13 @@ pub type SeedSyncSessionPtr = Arc<SeedSyncSession>;
 /// Defines seed connections session
 pub struct SeedSyncSession {
     pub(in crate::net) p2p: LazyWeak<P2p>,
-    slots: Mutex<Vec<Arc<Slot>>>,
+    slots: AsyncMutex<Vec<Arc<Slot>>>,
 }
 
 impl SeedSyncSession {
     /// Create a new seed sync session instance
     pub(crate) fn new() -> SeedSyncSessionPtr {
-        Arc::new(Self { p2p: LazyWeak::new(), slots: Mutex::new(Vec::new()) })
+        Arc::new(Self { p2p: LazyWeak::new(), slots: AsyncMutex::new(Vec::new()) })
     }
 
     /// Initialize the seedsync session. Each slot is suspended while it waits
@@ -94,7 +94,7 @@ impl SeedSyncSession {
 
         // Initialize a slot for each configured seed.
         // Connections will be started by not yet activated.
-        for seed in &self.p2p().settings().seeds {
+        for seed in &self.p2p().settings().read().await.seeds {
             let slot = Slot::new(self_.clone(), seed.clone(), self.p2p().settings());
             futures.push(slot.clone().start());
             slots.push(slot);
@@ -154,7 +154,11 @@ struct Slot {
 }
 
 impl Slot {
-    fn new(session: Weak<SeedSyncSession>, addr: Url, settings: Arc<Settings>) -> Arc<Self> {
+    fn new(
+        session: Weak<SeedSyncSession>,
+        addr: Url,
+        settings: Arc<AsyncRwLock<Settings>>,
+    ) -> Arc<Self> {
         Arc::new(Self {
             addr,
             process: StoppableTask::new(),

src/net/tests.rs

@@ -169,7 +169,7 @@ async fn get_random_gold_host(
 ) -> ((Url, u64), usize) {
     let random_node = &outbound_instances[index];
     let hosts = random_node.hosts();
-    let external_addr = &random_node.settings().external_addrs[0];
+    let external_addr = random_node.settings().read().await.external_addrs[0].clone();
 
     info!("========================================================");
     info!("Getting gold addr from node={}", external_addr);
@@ -185,7 +185,7 @@ async fn get_random_gold_host(
 async fn _check_random_hostlist(outbound_instances: &Vec<Arc<P2p>>, rng: &mut ThreadRng) {
     let mut urls = HashSet::new();
     let random_node = outbound_instances.choose(rng).unwrap();
-    let external_addr = &random_node.settings().external_addrs[0];
+    let external_addr = random_node.settings().read().await.external_addrs[0].clone();
 
     info!("========================================================");
     info!("Checking node={}", external_addr);
@@ -209,7 +209,7 @@ async fn _check_random_hostlist(outbound_instances: &Vec<Arc<P2p>>, rng: &mut Th
 
 async fn check_all_hostlist(outbound_instances: &Vec<Arc<P2p>>) {
     for node in outbound_instances {
-        let external_addr = &node.settings().external_addrs[0];
+        let external_addr = &node.settings().read().await.external_addrs[0].clone();
         info!("========================================================");
         info!("Checking node={}", external_addr);
         info!("========================================================");
@@ -233,9 +233,9 @@ async fn check_all_hostlist(outbound_instances: &Vec<Arc<P2p>>) {
 }
 async fn kill_node(outbound_instances: &Vec<Arc<P2p>>, node: Url) {
     for p2p in outbound_instances {
-        if p2p.settings().external_addrs[0] == node {
+        if p2p.settings().read().await.external_addrs[0] == node {
             info!("========================================================");
-            info!("Shutting down node: {}", p2p.settings().external_addrs[0]);
+            info!("Shutting down node: {}", p2p.settings().read().await.external_addrs[0]);
             info!("========================================================");
             p2p.stop().await;
         }
@@ -312,7 +312,7 @@ async fn p2p_test_real(ex: Arc<Executor<'static>>) {
 
     for p2p in &outbound_instances {
         info!("========================================================");
-        info!("Starting node={}", p2p.settings().external_addrs[0]);
+        info!("Starting node={}", p2p.settings().read().await.external_addrs[0]);
         info!("========================================================");
         p2p.clone().start().await.unwrap();
     }
@@ -344,7 +344,7 @@ async fn p2p_test_real(ex: Arc<Executor<'static>>) {
     assert!(!seed.hosts().container.is_empty(HostColor::White));
 
     info!("========================================================");
-    info!("Checking seed={}", seed.settings().inbound_addrs[0]);
+    info!("Checking seed={}", seed.settings().read().await.inbound_addrs[0]);
     info!("========================================================");
 
     let mut urls = HashSet::new();
@@ -439,7 +439,7 @@ async fn p2p_test_real(ex: Arc<Executor<'static>>) {
 
     for p2p in &manual_instances {
         info!("========================================================");
-        info!("Starting node={}", p2p.settings().external_addrs[0]);
+        info!("Starting node={}", p2p.settings().read().await.external_addrs[0]);
         info!("========================================================");
         p2p.clone().start().await.unwrap();
     }
@@ -457,7 +457,7 @@ async fn p2p_test_real(ex: Arc<Executor<'static>>) {
         // We should have (N_CONNS outbound + N_CONNS inbound)
         // connections at this point.
         info!("========================================================");
-        info!("Checking manual node={}", p2p.settings().node_id);
+        info!("Checking manual node={}", p2p.settings().read().await.node_id);
         info!("========================================================");
         let channels = p2p.hosts().channels();
         assert!(channels.len() == N_CONNS * 2);