fud: add `fud.start_tasks()`

This method starts all the tasks that are necessary for fud to work. Those tasks are stopped in `fud.stop()`.

bin/fud/fud/src/lib.rs

@@ -39,7 +39,8 @@ use url::Url;
 
 use darkfi::{
     dht::{
-        impl_dht_node_defaults, Dht, DhtHandler, DhtNode, DhtRouterItem, DhtRouterPtr, DhtSettings,
+        impl_dht_node_defaults, tasks as dht_tasks, Dht, DhtHandler, DhtNode, DhtRouterItem,
+        DhtRouterPtr, DhtSettings,
     },
     geode::{hash_to_string, ChunkedStorage, FileSequence, Geode, MAX_CHUNK_SIZE},
     net::{ChannelPtr, P2pPtr},
@@ -75,7 +76,7 @@ pub mod rpc;
 
 /// Background tasks
 pub mod tasks;
-use tasks::FetchReply;
+use tasks::{start_task, FetchReply};
 
 /// Bitcoin
 pub mod bitcoin;
@@ -172,8 +173,14 @@ pub struct Fud {
     /// Currently active put tasks (running the `fud.insert_resource()` method)
     put_tasks: Arc<RwLock<HashMap<PathBuf, Arc<StoppableTask>>>>,
 
+    /// Currently active tasks (defined in `tasks`, started with the `start_task` macro)
+    tasks: Arc<RwLock<HashMap<String, Arc<StoppableTask>>>>,
+
     /// Used to send events to fud clients
     event_publisher: PublisherPtr<FudEvent>,
+
+    /// Global multithreaded executor reference
+    pub executor: ExecutorPtr,
 }
 
 #[async_trait]
@@ -328,12 +335,23 @@ impl Fud {
             put_rx,
             fetch_tasks: Arc::new(RwLock::new(HashMap::new())),
             put_tasks: Arc::new(RwLock::new(HashMap::new())),
+            tasks: Arc::new(RwLock::new(HashMap::new())),
             event_publisher,
+            executor,
         };
 
         Ok(fud)
     }
 
+    pub async fn start_tasks(self: &Arc<Self>) {
+        let mut tasks = self.tasks.write().await;
+        start_task!(self, "get", tasks::get_task, tasks);
+        start_task!(self, "put", tasks::put_task, tasks);
+        start_task!(self, "DHT channel", dht_tasks::channel_task::<Fud, FudNode>, tasks);
+        start_task!(self, "announce", tasks::announce_seed_task, tasks);
+        start_task!(self, "node ID", tasks::node_id_task, tasks);
+    }
+
     /// Bootstrap the DHT, verify our resources, add ourselves to
     /// `seeders_router` for the resources we already have, announce our files.
     async fn init(&self) -> Result<()> {
@@ -1680,8 +1698,9 @@ impl Fud {
         notify_event!(self, ResourceRemoved, { hash: *hash });
     }
 
-    /// Stop all tasks in `fetch_tasks` and `put_tasks.
+    /// Stop all tasks.
     pub async fn stop(&self) {
+        info!("Stopping fetch tasks...");
         // Create a clone of fetch_tasks because `task.stop()` needs a write lock
         let fetch_tasks = self.fetch_tasks.read().await;
         let cloned_fetch_tasks: HashMap<blake3::Hash, Arc<StoppableTask>> =
@@ -1693,6 +1712,7 @@ impl Fud {
             task.stop().await;
         }
 
+        info!("Stopping put tasks...");
         // Create a clone of put_tasks because `task.stop()` needs a write lock
         let put_tasks = self.put_tasks.read().await;
         let cloned_put_tasks: HashMap<PathBuf, Arc<StoppableTask>> =
@@ -1703,5 +1723,13 @@ impl Fud {
         for task in cloned_put_tasks.values() {
             task.stop().await;
         }
+
+        // Stop all other tasks
+        let mut tasks = self.tasks.write().await;
+        for (name, task) in tasks.clone() {
+            info!("Stopping {name} task...");
+            task.stop().await;
+        }
+        *tasks = HashMap::new();
     }
 }

bin/fud/fud/src/main.rs

@@ -24,7 +24,6 @@ use structopt_toml::StructOptToml;
 
 use darkfi::{
     async_daemonize,
-    dht::DhtHandler,
     net::{session::SESSION_DEFAULT, P2p, Settings as NetSettings},
     rpc::{
         jsonrpc::JsonSubscriber,
@@ -36,10 +35,9 @@ use darkfi::{
     Error, Result,
 };
 use fud::{
-    proto::{FudFindNodesReply, ProtocolFud},
+    proto::ProtocolFud,
     rpc::JsonRpcInterface,
     settings::{Args, CONFIG_FILE, CONFIG_FILE_CONTENTS},
-    tasks::{announce_seed_task, get_task, node_id_task, put_task},
     Fud,
 };
 
@@ -89,7 +87,9 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
     let fud: Arc<Fud> =
         Arc::new(Fud::new(args_, p2p.clone(), &sled_db, event_pub.clone(), ex.clone()).await?);
 
-    info!(target: "fud", "Starting download subs task");
+    fud.start_tasks().await;
+
+    info!(target: "fud", "Starting event subs task");
     let event_sub = JsonSubscriber::new("event");
     let event_sub_ = event_sub.clone();
     let event_task = StoppableTask::new();
@@ -112,34 +112,6 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
         ex.clone(),
     );
 
-    info!(target: "fud", "Starting get task");
-    let get_task_ = StoppableTask::new();
-    get_task_.clone().start(
-        get_task(fud.clone(), ex.clone()),
-        |res| async {
-            match res {
-                Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
-                Err(e) => error!(target: "fud", "Failed starting get task: {e}"),
-            }
-        },
-        Error::DetachedTaskStopped,
-        ex.clone(),
-    );
-
-    info!(target: "fud", "Starting put task");
-    let put_task_ = StoppableTask::new();
-    put_task_.clone().start(
-        put_task(fud.clone(), ex.clone()),
-        |res| async {
-            match res {
-                Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
-                Err(e) => error!(target: "fud", "Failed starting put task: {e}"),
-            }
-        },
-        Error::DetachedTaskStopped,
-        ex.clone(),
-    );
-
     let rpc_settings: RpcSettings = args.rpc.into();
     info!(target: "fud", "Starting JSON-RPC server on {}", rpc_settings.listen);
     let rpc_interface = Arc::new(JsonRpcInterface::new(fud.clone(), dnet_sub, event_sub));
@@ -175,76 +147,19 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
     }
     drop(p2p_settings);
 
-    info!(target: "fud", "Starting DHT tasks");
-    let dht_channel_task = StoppableTask::new();
-    let fud_ = fud.clone();
-    dht_channel_task.clone().start(
-        async move { fud_.channel_task::<FudFindNodesReply>().await },
-        |res| async {
-            match res {
-                Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
-                Err(e) => error!(target: "fud", "Failed starting dht channel task: {e}"),
-            }
-        },
-        Error::DetachedTaskStopped,
-        ex.clone(),
-    );
-    let announce_task = StoppableTask::new();
-    let fud_ = fud.clone();
-    announce_task.clone().start(
-        async move { announce_seed_task(fud_.clone()).await },
-        |res| async {
-            match res {
-                Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
-                Err(e) => error!(target: "fud", "Failed starting announce task: {e}"),
-            }
-        },
-        Error::DetachedTaskStopped,
-        ex.clone(),
-    );
-
-    info!(target: "fud", "Starting node ID task");
-    let node_task = StoppableTask::new();
-    let fud_ = fud.clone();
-    node_task.clone().start(
-        async move { node_id_task(fud_.clone()).await },
-        |res| async {
-            match res {
-                Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
-                Err(e) => error!(target: "fud", "Failed starting node ID task: {e}"),
-            }
-        },
-        Error::DetachedTaskStopped,
-        ex.clone(),
-    );
-
     // Signal handling for graceful termination.
     let (signals_handler, signals_task) = SignalHandler::new(ex)?;
     signals_handler.wait_termination(signals_task).await?;
     info!(target: "fud", "Caught termination signal, cleaning up and exiting...");
 
-    info!(target: "fud", "Stopping fetch tasks...");
     fud.stop().await;
 
-    info!(target: "fud", "Stopping get task...");
-    get_task_.stop().await;
-
-    info!(target: "fud", "Stopping put task...");
-    put_task_.stop().await;
-
     info!(target: "fud", "Stopping JSON-RPC server...");
     rpc_task.stop().await;
 
     info!(target: "fud", "Stopping P2P network...");
     p2p.stop().await;
 
-    info!(target: "fud", "Stopping DHT tasks...");
-    dht_channel_task.stop().await;
-    announce_task.stop().await;
-
-    info!(target: "fud", "Stopping node ID task...");
-    node_task.stop().await;
-
     info!(target: "fud", "Flushing sled database...");
     let flushed_bytes = sled_db.flush_async().await?;
     info!(target: "fud", "Flushed {flushed_bytes} bytes");

bin/fud/fud/src/tasks.rs

@@ -23,7 +23,7 @@ use log::{error, info, warn};
 use darkfi::{
     dht::{DhtHandler, DhtNode},
     geode::hash_to_string,
-    system::{sleep, ExecutorPtr, StoppableTask},
+    system::{sleep, StoppableTask},
     Error, Result,
 };
 
@@ -44,7 +44,7 @@ pub enum FetchReply {
 /// It creates a new StoppableTask (running `fud.fetch_resource()`) and inserts
 /// it into the `fud.fetch_tasks` hashmap. When the task is stopped it's
 /// removed from the hashmap.
-pub async fn get_task(fud: Arc<Fud>, executor: ExecutorPtr) -> Result<()> {
+pub async fn get_task(fud: Arc<Fud>) -> Result<()> {
     loop {
         let (hash, path, files) = fud.get_rx.recv().await.unwrap();
 
@@ -78,13 +78,13 @@ pub async fn get_task(fud: Arc<Fud>, executor: ExecutorPtr) -> Result<()> {
                 }
             },
             Error::DetachedTaskStopped,
-            executor.clone(),
+            fud.executor.clone(),
         );
     }
 }
 
 /// Triggered when calling the `fud.put()` method.
-pub async fn put_task(fud: Arc<Fud>, executor: ExecutorPtr) -> Result<()> {
+pub async fn put_task(fud: Arc<Fud>) -> Result<()> {
     loop {
         let path = fud.put_rx.recv().await.unwrap();
 
@@ -119,7 +119,7 @@ pub async fn put_task(fud: Arc<Fud>, executor: ExecutorPtr) -> Result<()> {
                 }
             },
             Error::DetachedTaskStopped,
-            executor.clone(),
+            fud.executor.clone(),
         );
     }
 }
@@ -245,3 +245,24 @@ pub async fn node_id_task(fud: Arc<Fud>) -> Result<()> {
         // DHT will be bootstrapped on the next channel connection
     }
 }
+
+macro_rules! start_task {
+    ($fud:expr, $task_name:expr, $task_fn:expr, $tasks:expr) => {{
+        info!(target: "fud", "Starting {} task", $task_name);
+        let task = StoppableTask::new();
+        let fud_ = $fud.clone();
+        task.clone().start(
+            async move { $task_fn(fud_).await },
+            |res| async {
+                match res {
+                    Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
+                    Err(e) => error!(target: "fud", "Failed starting {} task: {e}", $task_name),
+                }
+            },
+            Error::DetachedTaskStopped,
+            $fud.executor.clone(),
+        );
+        $tasks.insert($task_name.to_string(), task);
+    }};
+}
+pub(crate) use start_task;

src/dht/handler.rs

@@ -23,6 +23,7 @@ use num_bigint::BigUint;
 use smol::{lock::Semaphore, stream::StreamExt};
 use std::{
     collections::{HashMap, HashSet},
+    marker::Sync,
     sync::Arc,
     time::Duration,
 };
@@ -40,7 +41,7 @@ use crate::{
 };
 
 #[async_trait]
-pub trait DhtHandler<N: DhtNode> {
+pub trait DhtHandler<N: DhtNode>: Sync {
     fn dht(&self) -> Arc<Dht<N>>;
 
     /// Get our own node
@@ -98,53 +99,6 @@ pub trait DhtHandler<N: DhtNode> {
         }
     }
 
-    /// Send a DHT ping request when there is a new channel, to know the node id of the new peer,
-    /// Then fill the channel cache and the buckets
-    async fn channel_task<M: Message>(&self) -> Result<()> {
-        loop {
-            let channel_sub = self.dht().p2p.hosts().subscribe_channel().await;
-            let res = channel_sub.receive().await;
-            channel_sub.unsubscribe().await;
-            if res.is_err() {
-                continue;
-            }
-            let channel = res.unwrap();
-            let channel_cache_lock = self.dht().channel_cache.clone();
-            let mut channel_cache = channel_cache_lock.write().await;
-
-            // Skip this channel if it's stopped or not new.
-            if channel.is_stopped() || channel_cache.keys().any(|&k| k == channel.info.id) {
-                continue;
-            }
-            // Skip this channel if it's a seed or refine session.
-            if channel.session_type_id() & (SESSION_SEED | SESSION_REFINE) != 0 {
-                continue;
-            }
-
-            let ping_res = self.ping(channel.clone()).await;
-
-            if let Err(e) = ping_res {
-                warn!(target: "dht::DhtHandler::channel_task()", "Error while pinging (requesting node id) {}: {e}", channel.address());
-                // channel.stop().await;
-                continue;
-            }
-
-            let node = ping_res.unwrap();
-
-            channel_cache.entry(channel.info.id).or_insert_with(|| ChannelCacheItem {
-                node: node.clone(),
-                topic: None,
-                usage_count: 0,
-            });
-            drop(channel_cache);
-
-            if !node.addresses().is_empty() {
-                self.add_node(node.clone()).await;
-                let _ = self.on_new_node(&node.clone()).await;
-            }
-        }
-    }
-
     /// Add a node in the correct bucket
     async fn add_node(&self, node: N)
     where

src/dht/mod.rs

@@ -40,6 +40,8 @@ pub use settings::{DhtSettings, DhtSettingsOpt};
 pub mod handler;
 pub use handler::DhtHandler;
 
+pub mod tasks;
+
 pub trait DhtNode: Debug + Clone + Send + Sync + PartialEq + Eq + Hash {
     fn id(&self) -> blake3::Hash;
     fn addresses(&self) -> Vec<Url>;

src/dht/tasks.rs

@@ -0,0 +1,73 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2025 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * 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/>.
+ */
+
+use log::warn;
+use std::sync::Arc;
+
+use crate::{
+    dht::{ChannelCacheItem, DhtHandler, DhtNode},
+    net::session::{SESSION_REFINE, SESSION_SEED},
+    Result,
+};
+
+/// Send a DHT ping request when there is a new channel, to know the node id of the new peer,
+/// Then fill the channel cache and the buckets
+pub async fn channel_task<H: DhtHandler<N>, N: DhtNode>(handler: Arc<H>) -> Result<()> {
+    loop {
+        let channel_sub = handler.dht().p2p.hosts().subscribe_channel().await;
+        let res = channel_sub.receive().await;
+        channel_sub.unsubscribe().await;
+        if res.is_err() {
+            continue;
+        }
+        let channel = res.unwrap();
+        let channel_cache_lock = handler.dht().channel_cache.clone();
+        let mut channel_cache = channel_cache_lock.write().await;
+
+        // Skip this channel if it's stopped or not new.
+        if channel.is_stopped() || channel_cache.keys().any(|&k| k == channel.info.id) {
+            continue;
+        }
+        // Skip this channel if it's a seed or refine session.
+        if channel.session_type_id() & (SESSION_SEED | SESSION_REFINE) != 0 {
+            continue;
+        }
+
+        let ping_res = handler.ping(channel.clone()).await;
+
+        if let Err(e) = ping_res {
+            warn!(target: "dht::DhtHandler::channel_task()", "Error while pinging (requesting node id) {}: {e}", channel.address());
+            // channel.stop().await;
+            continue;
+        }
+
+        let node = ping_res.unwrap();
+
+        channel_cache.entry(channel.info.id).or_insert_with(|| ChannelCacheItem {
+            node: node.clone(),
+            topic: None,
+            usage_count: 0,
+        });
+        drop(channel_cache);
+
+        if !node.addresses().is_empty() {
+            handler.add_node(node.clone()).await;
+            let _ = handler.on_new_node(&node.clone()).await;
+        }
+    }
+}