net: remove p2p.run(), now we simply call p2p.start() and p2p.stop()

bin/darkirc/src/main.rs

@@ -259,17 +259,6 @@ async fn realmain(settings: Args, executor: Arc<smol::Executor<'static>>) -> Res
     ////////////////////
     info!(target: "darkirc", "Starting P2P network");
     p2p.clone().start().await?;
-    StoppableTask::new().start(
-        p2p.clone().run(),
-        |res| async {
-            match res {
-                Ok(()) | Err(Error::P2PNetworkStopped) => { /* Do nothing */ }
-                Err(e) => error!(target: "darkirc", "Failed starting P2P network: {}", e),
-            }
-        },
-        Error::P2PNetworkStopped,
-        executor.clone(),
-    );
 
     ////////////////////
     // IRC server

src/net/p2p.rs

@@ -40,8 +40,10 @@ use super::{
     settings::{Settings, SettingsPtr},
 };
 use crate::{
-    system::{Subscriber, SubscriberPtr, Subscription},
-    Result,
+    system::{
+        sleep_forever, StoppableTask, StoppableTaskPtr, Subscriber, SubscriberPtr, Subscription,
+    },
+    Error, Result,
 };
 
 /// Set of channels that are awaiting connection
@@ -61,8 +63,6 @@ pub struct P2p {
     channels: ConnectedChannels,
     /// Subscriber for notifications of new channels
     channel_subscriber: SubscriberPtr<Result<ChannelPtr>>,
-    /// Subscriber for stop notifications
-    stop_subscriber: SubscriberPtr<()>,
     /// Known hosts (peers)
     hosts: HostsPtr,
     /// Protocol registry
@@ -79,6 +79,9 @@ pub struct P2p {
     /// Reference to configured [`OutboundSession`]
     session_outbound: Mutex<Option<Arc<OutboundSession>>>,
 
+    /// Main process that starts and stops all the sessions
+    process: StoppableTaskPtr,
+
     /// Enable network debugging
     pub dnet_enabled: Mutex<bool>,
     /// The subscriber for which we can give dnet info over
@@ -102,7 +105,6 @@ impl P2p {
             pending: Mutex::new(HashSet::new()),
             channels: Mutex::new(HashMap::new()),
             channel_subscriber: Subscriber::new(),
-            stop_subscriber: Subscriber::new(),
             hosts: Hosts::new(settings.clone()),
             protocol_registry: ProtocolRegistry::new(),
             settings,
@@ -112,6 +114,8 @@ impl P2p {
             session_inbound: Mutex::new(None),
             session_outbound: Mutex::new(None),
 
+            process: StoppableTask::new(),
+
             dnet_enabled: Mutex::new(false),
             dnet_subscriber: Subscriber::new(),
         });
@@ -129,35 +133,24 @@ impl P2p {
 
     /// Invoke startup and seeding sequence. Call from constructing thread.
     pub async fn start(self: Arc<Self>) -> Result<()> {
-        debug!(target: "net::p2p::start()", "P2P::start() [BEGIN]");
-        info!(target: "net::p2p::start()", "[P2P] Seeding P2P subsystem");
-
-        // Start seed session
-        let seed = SeedSyncSession::new(Arc::downgrade(&self));
-        // This will block until all seed queries have finished
-        seed.start().await?;
-
-        debug!(target: "net::p2p::start()", "P2P::start() [END]");
+        let self_ = self.clone();
+        self.process.clone().start(
+            self.clone()._run(),
+            |result| self_.handle_stop(result),
+            Error::NetworkServiceStopped,
+            self.executor(),
+        );
         Ok(())
     }
 
-    /// Reseed the P2P network.
-    pub async fn reseed(self: Arc<Self>) -> Result<()> {
-        debug!(target: "net::p2p::reseed()", "P2P::reseed() [BEGIN]");
-        info!(target: "net::p2p::reseed()", "[P2P] Reseeding P2P subsystem");
-
-        // Start seed session
-        let seed = SeedSyncSession::new(Arc::downgrade(&self));
-        // This will block until all seed queries have finished
-        seed.start().await?;
-
-        debug!(target: "net::p2p::reseed()", "P2P::reseed() [END]");
+    // TODO: fix this, see outbound session also
+    async fn _run(self: Arc<Self>) -> Result<()> {
+        self.run().await;
         Ok(())
     }
-
     /// Runs the network. Starts inbound, outbound, and manual sessions.
     /// Waits for a stop signal and stops the network if received.
-    pub async fn run(self: Arc<Self>) -> Result<()> {
+    async fn run(self: Arc<Self>) {
         debug!(target: "net::p2p::run()", "P2P::run() [BEGIN]");
         info!(target: "net::p2p::run()", "[P2P] Running P2P subsystem");
 
@@ -169,37 +162,55 @@ impl P2p {
 
         // Start the inbound session
         let inbound = self.session_inbound().await;
-        inbound.clone().start().await?;
+        if let Err(err) = inbound.clone().start().await {
+            error!(target: "net::p2p::run()", "Failed to start inbound session!: {}", err);
+            manual.stop().await;
+            return
+        }
 
         // Start the outbound session
         let outbound = self.session_outbound().await;
-        outbound.clone().start().await?;
+        if let Err(err) = outbound.clone().start().await {
+            error!(target: "net::p2p::run()", "Failed to start outbound session!: {}", err);
+            manual.stop().await;
+            inbound.stop().await;
+            return
+        }
 
         info!(target: "net::p2p::run()", "[P2P] P2P subsystem started");
 
         // Wait for stop signal
-        let stop_sub = self.subscribe_stop().await;
-        stop_sub.receive().await;
-
-        info!(target: "net::p2p::run()", "[P2P] Received P2P subsystem stop signal. Shutting down.");
+        sleep_forever().await;
+        unreachable!();
+    }
 
+    async fn handle_stop(self: Arc<Self>, result: Result<()>) {
+        assert!(result.is_err());
+        //assert_eq!(result.unwrap_err(), Error::NetworkServiceStopped);
+        info!(target: "net::p2p::handle_stop()", "[P2P] Received stop signal. Shutting down.");
         // Stop the sessions
-        manual.stop().await;
-        inbound.stop().await;
-        outbound.stop().await;
-
-        debug!(target: "net::p2p::run()", "P2P::run() [END]");
-        Ok(())
+        self.session_manual().await.stop().await;
+        self.session_inbound().await.stop().await;
+        self.session_outbound().await.stop().await;
     }
 
-    /// Subscribe to a stop signal.
-    pub async fn subscribe_stop(&self) -> Subscription<()> {
-        self.stop_subscriber.clone().subscribe().await
+    /// Reseed the P2P network.
+    pub async fn seed(self: Arc<Self>) -> Result<()> {
+        debug!(target: "net::p2p::seed()", "P2P::seed() [BEGIN]");
+        info!(target: "net::p2p::seed()", "[P2P] Seeding P2P subsystem");
+
+        // Start seed session
+        let seed = SeedSyncSession::new(Arc::downgrade(&self));
+        // This will block until all seed queries have finished
+        seed.start().await?;
+
+        debug!(target: "net::p2p::seed()", "P2P::seed() [END]");
+        Ok(())
     }
 
     /// Stop the running P2P subsystem
     pub async fn stop(&self) {
-        self.stop_subscriber.notify(()).await
+        self.process.stop().await;
     }
 
     /// Broadcasts a message concurrently across all active channels.
@@ -246,7 +257,7 @@ impl P2p {
     }
 
     /// Add a channel to the set of connected channels
-    pub(crate) async fn store(&self, channel: ChannelPtr) {
+    pub(super) async fn store(&self, channel: ChannelPtr) {
         // TODO: Check the code path for this, and potentially also insert the remote
         // into the hosts list?
         self.channels.lock().await.insert(channel.address().clone(), channel.clone());
@@ -254,17 +265,17 @@ impl P2p {
     }
 
     /// Remove a channel from the set of connected channels
-    pub(crate) async fn remove(&self, channel: ChannelPtr) {
+    pub(super) async fn remove(&self, channel: ChannelPtr) {
         self.channels.lock().await.remove(channel.address());
     }
 
     /// Add an address to the list of pending channels.
-    pub(crate) async fn add_pending(&self, addr: &Url) -> bool {
+    pub(super) async fn add_pending(&self, addr: &Url) -> bool {
         self.pending.lock().await.insert(addr.clone())
     }
 
     /// Remove a channel from the list of pending channels.
-    pub(crate) async fn remove_pending(&self, addr: &Url) {
+    pub(super) async fn remove_pending(&self, addr: &Url) {
         self.pending.lock().await.remove(addr);
     }
 
@@ -290,7 +301,7 @@ impl P2p {
     }
 
     /// Reference the global executor
-    pub(super) fn executor(&self) -> Arc<Executor<'static>> {
+    pub fn executor(&self) -> Arc<Executor<'static>> {
         self.executor.clone()
     }
 

src/net/session/outbound_session.rs

@@ -131,7 +131,7 @@ pub enum SlotState {
 
 pub struct Slot {
     slot: u32,
-    process: Mutex<StoppableTaskPtr>,
+    process: StoppableTaskPtr,
     state: Mutex<SlotState>,
     session: Weak<OutboundSession>,
 }
@@ -140,7 +140,7 @@ impl Slot {
     pub fn new(session: Weak<OutboundSession>, slot: u32) -> Arc<Slot> {
         Arc::new(Self {
             slot,
-            process: Mutex::new(StoppableTask::new()),
+            process: StoppableTask::new(),
             state: Mutex::new(SlotState::Inactive),
             session,
         })
@@ -148,7 +148,7 @@ impl Slot {
 
     async fn start(self: Arc<Self>) {
         // TODO: way too many clones, look into making this implicit. See implicit-clone crate
-        self.process.lock().await.clone().start(
+        self.process.clone().start(
             self.clone()._run(),
             // Ignore stop handler
             |_| async {},
@@ -157,7 +157,7 @@ impl Slot {
         );
     }
     async fn stop(self: Arc<Self>) {
-        self.process.lock().await.stop().await
+        self.process.stop().await
     }
 
     // TODO: need to fix StoppableTask so it accepts arbitrary function signatures
@@ -455,7 +455,7 @@ impl Slot {
                 "[P2P] No connected channels found for peer discovery. Reseeding.",
             );
 
-            if let Err(e) = p2p.clone().reseed().await {
+            if let Err(e) = p2p.clone().seed().await {
                 error!(
                     target: "net::outbound_session::peer_discovery()",
                     "[P2P] Network reseed failed: {}", e,

src/system/mod.rs

@@ -38,6 +38,12 @@ pub async fn sleep(seconds: u64) {
     Timer::after(Duration::from_secs(seconds)).await;
 }
 
+pub async fn sleep_forever() {
+    loop {
+        sleep(100000000).await
+    }
+}
+
 /// Sleep for any number of milliseconds.
 pub async fn msleep(millis: u64) {
     Timer::after(Duration::from_millis(millis)).await;