net/outbound_session: add PeerDiscovery process to OutboundSession. Slots and PeerDiscovery communicate through CondVar

script/nodetool.py

@@ -14,7 +14,7 @@
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
-import asyncio, json, random, sys
+import asyncio, json, random, sys, time
 
 
 class JsonRpc:
@@ -85,10 +85,21 @@ async def main(argv):
 
     while True:
         data = await rpc.reader.readline()
-        data = json.loads(data.decode().strip())
-        if data["params"][0]["event"] in ["send", "recv"]:
+        #with open("rpclog", "a") as f:
+        #    f.write(data.decode())
+        data = json.loads(data)
+
+        params = data["params"][0]
+        ev = params["event"]
+        if ev in ["send", "recv"]:
             continue
-        print(data)
+        info = params["info"]
+        slot = info["slot"]
+
+        t = time.localtime()
+        current_time = time.strftime("%H:%M:%S", t)
+        print(f"{current_time}  slot {slot}: {ev}")
+        #print(data)
 
     await rpc.dnet_switch(False)
     await rpc.stop()

src/net/p2p.rs

@@ -24,7 +24,7 @@ use std::{
 use futures::{stream::FuturesUnordered, TryFutureExt};
 use log::{debug, error, info, warn};
 use rand::{prelude::IteratorRandom, rngs::OsRng};
-use smol::{lock::Mutex, stream::StreamExt, Executor};
+use smol::{lock::Mutex, stream::StreamExt};
 use url::Url;
 
 use super::{
@@ -40,7 +40,7 @@ use super::{
     settings::{Settings, SettingsPtr},
 };
 use crate::{
-    system::{Subscriber, SubscriberPtr, Subscription},
+    system::{ExecutorPtr, Subscriber, SubscriberPtr, Subscription},
     Result,
 };
 
@@ -54,7 +54,7 @@ pub type P2pPtr = Arc<P2p>;
 /// Toplevel peer-to-peer networking interface
 pub struct P2p {
     /// Global multithreaded executor reference
-    executor: Arc<Executor<'static>>,
+    executor: ExecutorPtr,
     /// Channels pending connection
     pending: PendingChannels,
     /// Connected channels
@@ -92,7 +92,7 @@ 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: Arc<Executor<'static>>) -> P2pPtr {
+    pub async fn new(settings: Settings, executor: ExecutorPtr) -> P2pPtr {
         let settings = Arc::new(settings);
 
         let self_ = Arc::new(Self {
@@ -117,7 +117,7 @@ impl P2p {
 
         *self_.session_manual.lock().await = Some(ManualSession::new(parent.clone()));
         *self_.session_inbound.lock().await = Some(InboundSession::new(parent.clone()));
-        *self_.session_outbound.lock().await = Some(OutboundSession::new(parent));
+        *self_.session_outbound.lock().await = Some(OutboundSession::new(parent).await);
 
         register_default_protocols(self_.clone()).await;
 
@@ -130,22 +130,20 @@ impl P2p {
         info!(target: "net::p2p::start()", "[P2P] Starting P2P subsystem");
 
         // First attempt any set manual connections
-        let manual = self.session_manual().await;
         for peer in &self.settings.peers {
-            manual.clone().connect(peer.clone()).await;
+            self.session_manual().await.connect(peer.clone()).await;
         }
 
         // Start the inbound session
         let inbound = self.session_inbound().await;
-        if let Err(err) = inbound.clone().start().await {
+        if let Err(err) = inbound.start().await {
             error!(target: "net::p2p::start()", "Failed to start inbound session!: {}", err);
-            manual.stop().await;
+            self.session_manual().await.stop().await;
             return Err(err)
         }
 
         // Start the outbound session
-        let outbound = self.session_outbound().await;
-        outbound.clone().start().await;
+        self.session_outbound().await.start().await;
 
         info!(target: "net::p2p::start()", "[P2P] P2P subsystem started");
         Ok(())
@@ -240,16 +238,20 @@ impl P2p {
     }
 
     /// Return reference to connected channels map
-    pub fn channels(&self) -> &ConnectedChannels {
+    pub async fn channels(&self) -> &ConnectedChannels {
         &self.channels
     }
 
     /// Retrieve a random connected channel from the
     pub async fn random_channel(&self) -> Option<ChannelPtr> {
-        let channels = self.channels().lock().await;
+        let channels = self.channels().await.lock().await;
         channels.values().choose(&mut OsRng).cloned()
     }
 
+    pub async fn is_connected(&self) -> bool {
+        !self.channels().await.lock().await.is_empty()
+    }
+
     /// Return an atomic pointer to the set network settings
     pub fn settings(&self) -> SettingsPtr {
         self.settings.clone()
@@ -261,7 +263,7 @@ impl P2p {
     }
 
     /// Reference the global executor
-    pub fn executor(&self) -> Arc<Executor<'static>> {
+    pub fn executor(&self) -> ExecutorPtr {
         self.executor.clone()
     }
 

src/net/session/manual_session.rs

@@ -66,7 +66,7 @@ impl ManualSession {
     pub fn new(p2p: Weak<P2p>) -> ManualSessionPtr {
         Arc::new(Self {
             p2p,
-            connect_slots: Mutex::new(vec![]),
+            connect_slots: Mutex::new(Vec::new()),
             channel_subscriber: Subscriber::new(),
             notify: Mutex::new(false),
         })

src/net/session/outbound_session.rs

@@ -45,37 +45,58 @@ use super::{
     Session, SessionBitFlag, SESSION_OUTBOUND,
 };
 use crate::{
-    system::{sleep, StoppableTask, StoppableTaskPtr, Subscriber, SubscriberPtr},
+    system::{sleep, CondVar, StoppableTask, StoppableTaskPtr, Subscriber, SubscriberPtr},
     Error, Result,
 };
 
+use std::sync::OnceLock;
+
+pub struct LazyWeak<Parent>(OnceLock<Weak<Parent>>);
+
+impl<Parent> LazyWeak<Parent> {
+    fn new() -> Self {
+        Self(OnceLock::new())
+    }
+
+    pub fn init(&self, parent: Arc<Parent>) {
+        assert!(self.0.get().is_none());
+        let parent = Arc::downgrade(&parent);
+        self.0.set(parent).unwrap();
+        assert!(self.0.get().is_some());
+    }
+
+    pub fn upgrade(&self) -> Arc<Parent> {
+        assert!(self.0.get().is_some());
+        self.0.get().unwrap().upgrade().unwrap()
+    }
+}
+
 pub type OutboundSessionPtr = Arc<OutboundSession>;
 
 /// Defines outbound connections session.
 pub struct OutboundSession {
     /// Weak pointer to parent p2p object
     p2p: Weak<P2p>,
-    /// Outbound connection slots
-    connect_slots: Mutex<Vec<StoppableTaskPtr>>,
     /// Subscriber used to signal channels processing
     channel_subscriber: SubscriberPtr<Result<ChannelPtr>>,
-    /// Flag to toggle channel_subscriber notifications
-    notify: Mutex<bool>,
 
     /// Outbound connection slots
     slots: Mutex<Vec<Arc<Slot>>>,
+    /// Peer discovery task
+    peer_discovery: Arc<PeerDiscovery>,
 }
 
 impl OutboundSession {
     /// Create a new outbound session.
-    pub(crate) fn new(p2p: Weak<P2p>) -> OutboundSessionPtr {
-        Arc::new(Self {
+    pub(crate) async fn new(p2p: Weak<P2p>) -> OutboundSessionPtr {
+        let self_ = Arc::new(Self {
             p2p,
-            connect_slots: Mutex::new(vec![]),
             channel_subscriber: Subscriber::new(),
-            notify: Mutex::new(false),
             slots: Mutex::new(Vec::new()),
-        })
+            peer_discovery: PeerDiscovery::new(),
+        });
+        self_.peer_discovery.session.init(self_.clone());
+        self_
     }
 
     /// Start the outbound session. Runs the channel connect loop.
@@ -92,14 +113,28 @@ impl OutboundSession {
             slot.clone().start().await;
             slots.push(slot);
         }
+
+        self.peer_discovery.clone().start().await;
     }
 
     /// Stops the outbound session.
     pub(crate) async fn stop(&self) {
-        let connect_slots = &*self.connect_slots.lock().await;
+        let slots = &*self.slots.lock().await;
 
-        for slot in connect_slots {
-            slot.stop().await;
+        for slot in slots {
+            slot.clone().stop().await;
+        }
+
+        self.peer_discovery.clone().stop().await;
+    }
+
+    fn wakeup_peer_discovery(&self) {
+        self.peer_discovery.notify()
+    }
+    async fn wakeup_slots(&self) {
+        let slots = &*self.slots.lock().await;
+        for slot in slots {
+            slot.notify();
         }
     }
 }
@@ -115,31 +150,16 @@ impl Session for OutboundSession {
     }
 }
 
-#[derive(PartialEq, Clone, Debug)]
-pub enum SlotState {
-    // Before start() is called
-    Inactive,
-    Active,
-    Discovery,
-    Seed,
-    Sleep,
-}
-
 pub struct Slot {
     slot: u32,
     process: StoppableTaskPtr,
-    state: Mutex<SlotState>,
+    wakeup_self: CondVar,
     session: Weak<OutboundSession>,
 }
 
 impl Slot {
-    pub fn new(session: Weak<OutboundSession>, slot: u32) -> Arc<Slot> {
-        Arc::new(Self {
-            slot,
-            process: StoppableTask::new(),
-            state: Mutex::new(SlotState::Inactive),
-            session,
-        })
+    fn new(session: Weak<OutboundSession>, slot: u32) -> Arc<Self> {
+        Arc::new(Self { slot, process: StoppableTask::new(), wakeup_self: CondVar::new(), session })
     }
 
     async fn start(self: Arc<Self>) {
@@ -149,7 +169,7 @@ impl Slot {
         self.process.clone().start(
             async move {
                 self.run().await;
-                Ok(())
+                unreachable!();
             },
             // Ignore stop handler
             |_| async {},
@@ -162,7 +182,6 @@ impl Slot {
     }
 
     async fn run(self: Arc<Self>) {
-        assert_eq!(*self.state.lock().await, SlotState::Inactive);
         // This is the main outbound connection loop where we try to establish
         // a connection in the slot. The `try_connect` function will block in
         // case the connection was sucessfully established. If it fails, then
@@ -176,45 +195,6 @@ impl Slot {
         // and attempt to fill the slot with another peer.
         loop {
             // Activate the slot
-            *self.state.lock().await = SlotState::Active;
-
-            /*
-            let mut addr = None;
-
-            if !load_addr {
-                if self.session().exists_inactive_slot() {
-                    *self.state.lock().await = SlotState::Sleep;
-                    wait for wakeup
-                    continue;
-                }
-
-                if !self.p2p().channels().lock.await.is_empty() {
-                    *self.state.lock().await = SlotState::Discovery;
-                    send get addr
-                    wait for addr back
-                    if load_addr {
-                        return addr
-                    }
-                }
-
-                // Finally try seed phase
-                *self.state.lock().await = SlotState::Seed;
-                self.p2p().seed().await;
-
-                if load_addr {
-                    return addr
-                } else {
-                    *self.state.lock().await = SlotState::Sleep;
-                    wait for wakeup
-                    continue
-                }
-            }
-
-            match try_connect() {
-                ...
-            }
-
-            */
             debug!(
                 target: "net::outbound_session::try_connect()",
                 "[P2P] Finding a host to connect to for outbound slot #{}",
@@ -228,13 +208,16 @@ impl Slot {
             let addr = if let Some(addr) = self.fetch_address_with_lock(transports).await {
                 addr
             } else {
+                self.wakeup_self.reset();
                 // Peer discovery
-                self.peer_discovery().await;
+                self.session().wakeup_peer_discovery();
+                // Wait to be woken up by peer discovery
+                self.wakeup_self.wait().await;
                 continue
             };
 
             let (addr_final, channel) = match self.try_connect(addr.clone()).await {
-                Ok((addr_final, channel)) => (addr_final, channel),
+                Ok(connect_info) => connect_info,
                 Err(err) => {
                     error!(
                         target: "net::outbound_session",
@@ -348,18 +331,6 @@ impl Slot {
     async fn fetch_address_with_lock(&self, transports: &[String]) -> Option<Url> {
         let p2p = self.p2p();
 
-        // TODO: REMOVE !!!!!
-        let retry_sleep = p2p.settings().outbound_connect_timeout;
-        if *p2p.peer_discovery_running.lock().await {
-            debug!(
-                target: "net::outbound_session::load_address()",
-                "[P2P] #{} Peer discovery active, waiting {} seconds...",
-                self.slot, retry_sleep,
-            );
-            sleep(retry_sleep).await;
-        }
-        // TODO: REMOVE !!!!!
-
         // Collect hosts
         let mut hosts = vec![];
 
@@ -415,6 +386,50 @@ impl Slot {
         None
     }
 
+    fn notify(&self) {
+        self.wakeup_self.notify()
+    }
+
+    fn session(&self) -> OutboundSessionPtr {
+        self.session.upgrade().unwrap()
+    }
+    fn p2p(&self) -> P2pPtr {
+        self.session().p2p()
+    }
+}
+
+struct PeerDiscovery {
+    process: StoppableTaskPtr,
+    wakeup_self: CondVar,
+    session: LazyWeak<OutboundSession>,
+}
+
+impl PeerDiscovery {
+    fn new() -> Arc<Self> {
+        Arc::new(Self {
+            process: StoppableTask::new(),
+            wakeup_self: CondVar::new(),
+            session: LazyWeak::new(),
+        })
+    }
+
+    async fn start(self: Arc<Self>) {
+        let ex = self.p2p().executor();
+        self.process.clone().start(
+            async move {
+                self.run().await;
+                unreachable!();
+            },
+            // Ignore stop handler
+            |_| async {},
+            Error::NetworkServiceStopped,
+            ex,
+        );
+    }
+    async fn stop(self: Arc<Self>) {
+        self.process.stop().await
+    }
+
     /// Activate peer discovery if not active already. This will loop through all
     /// connected P2P channels and send out a `GetAddrs` message to request more
     /// peers. Other parts of the P2P stack will then handle the incoming addresses
@@ -422,68 +437,51 @@ impl Slot {
     /// This function will also sleep `Settings::outbound_connect_timeout` seconds
     /// after broadcasting in order to let the P2P stack receive and work through
     /// the addresses it is expecting.
-    async fn peer_discovery(&self) {
-        let p2p = self.p2p();
+    async fn run(self: Arc<Self>) {
+        loop {
+            // wait to be woken up by notify()
+            self.wakeup_self.wait().await;
 
-        if *p2p.peer_discovery_running.lock().await {
-            info!(
-                target: "net::outbound_session::peer_discovery()",
-                "[P2P] Outbound #{}: Peer discovery already active",
-                self.slot,
-            );
-            return
-        }
+            let p2p = self.p2p();
 
-        info!(
-            target: "net::outbound_session::peer_discovery()",
-            "[P2P] Outbound #{}: Started peer discovery",
-            self.slot,
-        );
-        *p2p.peer_discovery_running.lock().await = true;
-
-        // Broadcast the GetAddrs message to all active channels.
-        // If we have no active channels, we will perform a SeedSyncSession instead.
-        if p2p.random_channel().await.is_some() {
-            let get_addrs = GetAddrsMessage { max: p2p.settings().outbound_connections as u32 };
-            info!(
-                target: "net::outbound_session::peer_discovery()",
-                "[P2P] Outbound #{}: Broadcasting GetAddrs across active channels",
-                self.slot,
-            );
-            p2p.broadcast(&get_addrs).await;
-        } else {
-            warn!(
-                target: "net::outbound_session::peer_discovery()",
-                "[P2P] No connected channels found for peer discovery. Reseeding.",
-            );
+            // Broadcast the GetAddrs message to all active channels.
+            // If we have no active channels, we will perform a SeedSyncSession instead.
+            if p2p.is_connected().await {
+                info!(
+                    target: "net::outbound_session::peer_discovery()",
+                    "[P2P] Outbound: Broadcasting GetAddrs across active channels",
+                );
 
-            if let Err(e) = p2p.clone().seed().await {
-                error!(
+                let get_addrs = GetAddrsMessage { max: p2p.settings().outbound_connections as u32 };
+                p2p.broadcast(&get_addrs).await;
+                // Temporary workaround. Sleep until the nodes respond back and
+                // we process the addr messages.
+                sleep(p2p.settings().outbound_connect_timeout).await;
+            } else {
+                warn!(
                     target: "net::outbound_session::peer_discovery()",
-                    "[P2P] Network reseed failed: {}", e,
+                    "[P2P] No connected channels found for peer discovery. Reseeding.",
                 );
+
+                if let Err(e) = p2p.clone().seed().await {
+                    error!(
+                        target: "net::outbound_session::peer_discovery()",
+                        "[P2P] Network reseed failed: {}", e,
+                    );
+                }
             }
-        }
 
-        // Now sleep to let the GetAddrs propagate, and hopefully
-        // in the meantime we'll get some peers.
-        debug!(
-            target: "net::outbound_session::peer_discovery()",
-            "[P2P] Outbound #{}: Sleeping {} seconds",
-            self.slot, p2p.settings().outbound_connect_timeout,
-        );
-        sleep(p2p.settings().outbound_connect_timeout).await;
-        *p2p.peer_discovery_running.lock().await = false;
+            self.session().wakeup_slots().await;
+            self.wakeup_self.reset();
+        }
     }
 
-    async fn populate_hosts() {}
-
-    async fn wakeup(self: Arc<Self>) {
-        // wakey :)
+    fn notify(&self) {
+        self.wakeup_self.notify()
     }
 
     fn session(&self) -> OutboundSessionPtr {
-        self.session.upgrade().unwrap()
+        self.session.upgrade()
     }
     fn p2p(&self) -> P2pPtr {
         self.session().p2p()

src/system/condvar.rs

@@ -1,45 +1,84 @@
 use std::{
     future::Future,
     pin::Pin,
-    sync::atomic::{AtomicBool, Ordering},
-    task::{Context, Poll},
+    sync::Mutex,
+    task::{Context, Poll, Waker},
 };
 
 /// Condition variable which allows a task to block until woken up
 pub struct CondVar {
-    is_active: AtomicBool,
+    state: Mutex<CondVarState>,
+}
+
+struct CondVarState {
+    is_awake: bool,
+    waker: Option<Waker>,
 }
 
 impl CondVar {
     pub fn new() -> Self {
-        Self { is_active: AtomicBool::new(false) }
+        Self { state: Mutex::new(CondVarState { is_awake: false, waker: None }) }
     }
 
     /// Wakeup the waiting task. Subsequent calls to this do nothing until `wait()` is called.
-    pub fn notify(&mut self) {
-        self.is_active.store(true, Ordering::Relaxed)
+    pub fn notify(&self) {
+        let mut state = self.state.lock().unwrap();
+        state.is_awake = true;
+        if let Some(waker) = state.waker.take() {
+            waker.wake()
+        }
     }
 
     /// Reset the condition variable and wait for a notification
-    pub async fn wait(&self) -> CondVarWait<'_> {
-        self.is_active.store(false, Ordering::SeqCst);
-        CondVarWait { condvar: self }
+    pub fn wait(&self) -> CondVarWait {
+        CondVarWait { state: &self.state }
     }
 
-    fn is_active(&self) -> bool {
-        self.is_active.load(Ordering::Relaxed)
+    /// Reset self ready to wait() again.
+    /// The reason this is separate from `wait()` is that usually
+    /// on the first `wait()` we want to catch any `notify()` calls that
+    /// happened before we started. For example,
+    /// ```rust
+    /// loop {
+    ///     // Wait for signal
+    ///     cv.wait().await;
+    ///
+    ///     // Do stuff...
+    ///
+    ///     cv.reset();
+    /// }
+    /// ```
+    pub fn reset(&self) {
+        let mut state = self.state.lock().unwrap();
+        state.is_awake = false;
     }
 }
 
 pub struct CondVarWait<'a> {
-    condvar: &'a CondVar,
+    state: &'a Mutex<CondVarState>,
 }
 
 impl<'a> Future for CondVarWait<'a> {
     type Output = ();
 
-    fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
-        match self.condvar.is_active() {
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let mut state = self.state.lock().unwrap();
+
+        // Avoid cloning wherever possible.
+        let new_waker = match state.waker.take() {
+            Some(waker) => {
+                let cx_waker = cx.waker();
+                if cx_waker.will_wake(&waker) {
+                    waker
+                } else {
+                    cx_waker.clone()
+                }
+            }
+            None => cx.waker().clone(),
+        };
+        state.waker = Some(new_waker);
+
+        match state.is_awake {
             true => Poll::Ready(()),
             false => Poll::Pending,
         }