fud: create `DhtSettings`

bin/fud/fud/fud_config.toml

@@ -9,11 +9,22 @@
 # Path to the contents directory
 base_dir = "~/.local/share/darkfi/fud"
 
-# Chunk transfer timeout in seconds
-chunk_timeout = 60
+## Chunk transfer timeout in seconds
+#chunk_timeout = 60
 
-# DHT requests timeout in seconds
-dht_timeout = 5
+# DHT settings
+[dht]
+## Number of nodes in a bucket
+#dht_k = 16
+
+## Number of lookup requests in a burst
+#dht_alpha = 4
+
+## Maximum number of parallel lookup requests
+#dht_concurrency = 10
+
+## Timeout in seconds
+#dht_timeout = 5
 
 # JSON-RPC settings
 [rpc]

bin/fud/fud/src/dht.rs

@@ -35,6 +35,7 @@ use futures::future::join_all;
 use log::{debug, error, warn};
 use num_bigint::BigUint;
 use smol::lock::RwLock;
+use structopt::StructOpt;
 use url::Url;
 
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable, Eq)]
@@ -86,7 +87,58 @@ impl From<DhtNode> for DhtRouterItem {
     }
 }
 
-// TODO: Add a DhtSettings
+#[derive(Clone, Debug)]
+pub struct DhtSettings {
+    /// Number of nodes in a bucket
+    pub k: usize,
+    /// Number of lookup requests in a burst
+    pub alpha: usize,
+    /// Maximum number of parallel lookup requests
+    pub concurrency: usize,
+    /// Timeout in seconds
+    pub timeout: u64,
+}
+
+impl Default for DhtSettings {
+    fn default() -> Self {
+        Self { k: 16, alpha: 4, concurrency: 10, timeout: 5 }
+    }
+}
+
+#[derive(Clone, Debug, serde::Deserialize, structopt::StructOpt, structopt_toml::StructOptToml)]
+#[structopt()]
+#[serde(rename = "dht")]
+pub struct DhtSettingsOpt {
+    /// Number of nodes in a DHT bucket
+    #[structopt(long)]
+    pub dht_k: Option<usize>,
+
+    /// Number of DHT lookup requests in a burst
+    #[structopt(long)]
+    pub dht_alpha: Option<usize>,
+
+    /// Maximum number of parallel DHT lookup requests
+    #[structopt(long)]
+    pub dht_concurrency: Option<usize>,
+
+    /// Timeout in seconds
+    #[structopt(long)]
+    pub dht_timeout: Option<u64>,
+}
+
+impl From<DhtSettingsOpt> for DhtSettings {
+    fn from(opt: DhtSettingsOpt) -> Self {
+        let def = DhtSettings::default();
+
+        Self {
+            k: opt.dht_k.unwrap_or(def.k),
+            alpha: opt.dht_alpha.unwrap_or(def.alpha),
+            concurrency: opt.dht_concurrency.unwrap_or(def.concurrency),
+            timeout: opt.dht_timeout.unwrap_or(def.timeout),
+        }
+    }
+}
+
 pub struct Dht {
     /// Our own node id
     pub node_id: blake3::Hash,
@@ -94,10 +146,6 @@ pub struct Dht {
     pub bootstrapped: Arc<RwLock<bool>>,
     /// Vec of buckets
     pub buckets: Arc<RwLock<Vec<DhtBucket>>>,
-    /// Number of parallel lookup requests
-    pub alpha: usize,
-    /// Number of nodes in a bucket
-    pub k: usize,
     /// Number of buckets
     pub n_buckets: usize,
     /// Channel ID -> Node ID
@@ -106,8 +154,8 @@ pub struct Dht {
     pub channel_cache: Arc<RwLock<HashMap<blake3::Hash, u32>>>,
     /// Node ID -> Set of keys
     pub router_cache: Arc<RwLock<HashMap<blake3::Hash, HashSet<blake3::Hash>>>>,
-    /// Seconds
-    pub timeout: u64,
+
+    pub settings: DhtSettings,
 
     pub p2p: P2pPtr,
     pub executor: ExecutorPtr,
@@ -115,9 +163,7 @@ pub struct Dht {
 impl Dht {
     pub async fn new(
         node_id: &blake3::Hash,
-        a: usize,
-        k: usize,
-        timeout: u64,
+        settings: &DhtSettings,
         p2p: P2pPtr,
         ex: ExecutorPtr,
     ) -> Self {
@@ -130,14 +176,13 @@ impl Dht {
         Self {
             node_id: *node_id,
             buckets: Arc::new(RwLock::new(buckets)),
-            bootstrapped: Arc::new(RwLock::new(false)),
-            alpha: a,
-            k,
             n_buckets: 256,
+            bootstrapped: Arc::new(RwLock::new(false)),
             node_cache: Arc::new(RwLock::new(HashMap::new())),
             channel_cache: Arc::new(RwLock::new(HashMap::new())),
             router_cache: Arc::new(RwLock::new(HashMap::new())),
-            timeout,
+
+            settings: settings.clone(),
 
             p2p: p2p.clone(),
             executor: ex,
@@ -371,7 +416,7 @@ pub trait DhtHandler {
         }
 
         // Bucket is full
-        if bucket.nodes.len() >= self.dht().k {
+        if bucket.nodes.len() >= self.dht().settings.k {
             // Ping the least recently seen node
             let channel = self.get_channel(&bucket.nodes[0]).await;
             if channel.is_ok() {
@@ -514,7 +559,7 @@ pub trait DhtHandler {
             let session_weak = Arc::downgrade(&self.dht().p2p.session_outbound());
 
             let connector = Connector::new(self.dht().p2p.settings(), session_weak);
-            let dur = Duration::from_secs(self.dht().timeout);
+            let dur = Duration::from_secs(self.dht().settings.timeout);
             let Ok(connect_res) = timeout(dur, connector.connect(&addr)).await else {
                 warn!(target: "dht::DhtHandler::get_channel()", "Timeout trying to connect to {}", addr);
                 return Err(Error::ConnectTimeout);

bin/fud/fud/src/main.rs

@@ -38,7 +38,10 @@ use smol::{
 };
 use structopt_toml::{structopt::StructOpt, StructOptToml};
 
-use crate::rpc::FudEvent;
+use crate::{
+    dht::{DhtSettings, DhtSettingsOpt},
+    rpc::FudEvent,
+};
 use darkfi::{
     async_daemonize, cli_desc,
     geode::{hash_to_string, ChunkedFile, Geode},
@@ -103,13 +106,9 @@ struct Args {
     /// Default path to store downloaded files (defaults to <base_dir>/downloads)
     downloads_path: Option<String>,
 
-    #[structopt(short, long)]
+    #[structopt(short, long, default_value = "60")]
     /// Chunk transfer timeout in seconds
-    chunk_timeout: Option<u64>,
-
-    #[structopt(short, long)]
-    /// DHT requests timeout in seconds
-    dht_timeout: Option<u64>,
+    chunk_timeout: u64,
 
     #[structopt(flatten)]
     /// Network settings
@@ -118,6 +117,10 @@ struct Args {
     #[structopt(flatten)]
     /// JSON-RPC settings
     rpc: RpcSettingsOpt,
+
+    #[structopt(flatten)]
+    /// DHT settings
+    dht: DhtSettingsOpt,
 }
 
 pub struct Fud {
@@ -181,7 +184,7 @@ impl DhtHandler for Fud {
 
         channel.send(&request).await?;
 
-        let reply = msg_subscriber.receive_with_timeout(self.dht().timeout).await?;
+        let reply = msg_subscriber.receive_with_timeout(self.dht().settings.timeout).await?;
 
         msg_subscriber.unsubscribe().await;
 
@@ -231,7 +234,7 @@ impl DhtHandler for Fud {
         let request = FudFindNodesRequest { key: *key };
         channel.send(&request).await?;
 
-        let reply = msg_subscriber_nodes.receive_with_timeout(self.dht().timeout).await?;
+        let reply = msg_subscriber_nodes.receive_with_timeout(self.dht().settings.timeout).await?;
 
         msg_subscriber_nodes.unsubscribe().await;
 
@@ -368,7 +371,8 @@ impl Fud {
                 continue;
             }
 
-            let reply = match msg_subscriber.receive_with_timeout(self.dht().timeout).await {
+            let reply = match msg_subscriber.receive_with_timeout(self.dht().settings.timeout).await
+            {
                 Ok(reply) => reply,
                 Err(e) => {
                     warn!(target: "fud::fetch_seeders()", "Error waiting for reply: {}", e);
@@ -559,7 +563,8 @@ impl Fud {
                 continue;
             }
 
-            let reply = match msg_subscriber.receive_with_timeout(self.dht().timeout).await {
+            let reply = match msg_subscriber.receive_with_timeout(self.dht().settings.timeout).await
+            {
                 Ok(reply) => reply,
                 Err(e) => {
                     warn!(target: "fud::fetch_file_metadata()", "Error waiting for reply: {}", e);
@@ -768,16 +773,14 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
     let (get_tx, get_rx) = smol::channel::unbounded();
     let (file_fetch_tx, file_fetch_rx) = smol::channel::unbounded();
     let (file_fetch_end_tx, file_fetch_end_rx) = smol::channel::unbounded();
-    // TODO: Add DHT settings in the config file
-    let dht = Arc::new(
-        Dht::new(&node_id_, 4, 16, args.dht_timeout.unwrap_or(5), p2p.clone(), ex.clone()).await,
-    );
+    let dht_settings: DhtSettings = args.dht.into();
+    let dht: Arc<Dht> = Arc::new(Dht::new(&node_id_, &dht_settings, p2p.clone(), ex.clone()).await);
     let fud = Arc::new(Fud {
         seeders_router,
         p2p: p2p.clone(),
         geode,
         downloads_path,
-        chunk_timeout: args.chunk_timeout.unwrap_or(60),
+        chunk_timeout: args.chunk_timeout,
         dht: dht.clone(),
         resources: Arc::new(RwLock::new(HashMap::new())),
         get_tx,

bin/fud/fud/src/proto.rs

@@ -279,7 +279,7 @@ impl ProtocolFud {
             }
 
             let reply = FudFindNodesReply {
-                nodes: self.fud.dht().find_neighbors(&request.key, self.fud.dht().k).await,
+                nodes: self.fud.dht().find_neighbors(&request.key, self.fud.dht().settings.k).await,
             };
             match self.channel.send(&reply).await {
                 Ok(()) => continue,