dhtd: Implement routing table and P2P protocols.

Cargo.lock

@@ -1618,6 +1618,23 @@ dependencies = [
  "syn 1.0.109",
 ]
 
+[[package]]
+name = "dhtd"
+version = "0.4.1"
+dependencies = [
+ "async-std",
+ "async-trait",
+ "blake3",
+ "darkfi",
+ "darkfi-serial",
+ "easy-parallel",
+ "log",
+ "rand",
+ "simplelog",
+ "smol",
+ "url",
+]
+
 [[package]]
 name = "digest"
 version = "0.9.0"

Cargo.toml

@@ -37,6 +37,7 @@ members = [
     "bin/tau/tau-cli",
     #"bin/darkwiki/darkwikid",
     #"bin/darkwiki/darkwiki-cli",
+    "bin/dhtd/dhtd",
     "bin/vanityaddr",
     "bin/lilith",
     "bin/zktool",

bin/dhtd/dhtd/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "dhtd"
+version = "0.4.1"
+homepage = "https://dark.fi"
+description = "DHT daemon"
+authors = ["Dyne.org foundation <foundation@dyne.org>"]
+repository = "https://github.com/darkrenaissance/darkfi"
+license = "AGPL-3.0-only"
+edition = "2021"
+
+[dependencies]
+async-std = {version = "1.12.0", features = ["attributes"]}
+async-trait = "0.1.68"
+blake3 = "1.3.3"
+darkfi = {path = "../../../", features = ["dht"]}
+darkfi-serial = {path = "../../../src/serial", features = ["derive", "crypto"]}
+easy-parallel = "3.3.0"
+log = "0.4.17"
+rand = "0.8.5"
+simplelog = "0.12.1"
+smol = "1.3.0"
+url = "2.3.1"

bin/dhtd/dhtd/src/main.rs

@@ -0,0 +1,40 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2023 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 std::collections::{HashMap, HashSet};
+
+use async_std::sync::{Arc, RwLock};
+use darkfi::{dht2::Dht, Result};
+use url::Url;
+
+/// Protocol implementations
+mod proto;
+
+//#[cfg(test)]
+mod tests;
+
+pub type DhtdPtr = Arc<RwLock<Dhtd>>;
+
+pub struct Dhtd {
+    pub dht: Dht,
+    pub routing_table: HashMap<blake3::Hash, HashSet<Url>>,
+}
+
+fn main() -> Result<()> {
+    Ok(())
+}

bin/dhtd/dhtd/src/proto.rs

@@ -0,0 +1,170 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2023 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 std::collections::HashSet;
+
+use async_std::sync::Arc;
+use async_trait::async_trait;
+use darkfi::{
+    dht2::net_hashmap::{NetHashMapInsert, NetHashMapRemove},
+    net::{
+        self, ChannelPtr, MessageSubscription, P2pPtr, ProtocolBase, ProtocolBasePtr,
+        ProtocolJobsManager, ProtocolJobsManagerPtr,
+    },
+    Result,
+};
+use darkfi_serial::{SerialDecodable, SerialEncodable};
+use log::debug;
+use smol::Executor;
+
+use super::DhtdPtr;
+
+pub struct ProtocolDht {
+    jobsman: ProtocolJobsManagerPtr,
+    channel: ChannelPtr,
+    p2p: P2pPtr,
+    state: DhtdPtr,
+    insert_sub: MessageSubscription<NetHashMapInsert<blake3::Hash, Vec<blake3::Hash>>>,
+    remove_sub: MessageSubscription<NetHashMapRemove<blake3::Hash>>,
+    chunk_request_sub: MessageSubscription<ChunkRequest>,
+    chunk_reply_sub: MessageSubscription<ChunkReply>,
+}
+
+#[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
+pub struct ChunkRequest {
+    pub hash: blake3::Hash,
+}
+
+impl net::Message for ChunkRequest {
+    fn name() -> &'static str {
+        "dhtchunkrequest"
+    }
+}
+
+#[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
+pub struct ChunkReply {
+    pub hash: blake3::Hash,
+    pub data: Vec<u8>,
+}
+
+impl net::Message for ChunkReply {
+    fn name() -> &'static str {
+        "dhtchunkreply"
+    }
+}
+
+impl ProtocolDht {
+    pub async fn init(channel: ChannelPtr, p2p: P2pPtr, state: DhtdPtr) -> Result<ProtocolBasePtr> {
+        let msg_subsystem = channel.get_message_subsystem();
+        msg_subsystem.add_dispatch::<NetHashMapInsert<blake3::Hash, Vec<blake3::Hash>>>().await;
+        msg_subsystem.add_dispatch::<NetHashMapRemove<blake3::Hash>>().await;
+        msg_subsystem.add_dispatch::<ChunkRequest>().await;
+        msg_subsystem.add_dispatch::<ChunkReply>().await;
+
+        let insert_sub = channel.subscribe_msg().await?;
+        let remove_sub = channel.subscribe_msg().await?;
+        let chunk_request_sub = channel.subscribe_msg().await?;
+        let chunk_reply_sub = channel.subscribe_msg().await?;
+
+        Ok(Arc::new(Self {
+            jobsman: ProtocolJobsManager::new("DHTProto", channel.clone()),
+            channel,
+            p2p,
+            state,
+            insert_sub,
+            remove_sub,
+            chunk_request_sub,
+            chunk_reply_sub,
+        }))
+    }
+
+    async fn handle_insert(self: Arc<Self>) -> Result<()> {
+        debug!("ProtocolDht::handle_insert START");
+        loop {
+            let Ok(msg) = self.insert_sub.receive().await else {
+                continue
+            };
+
+            let mut state = self.state.write().await;
+
+            if !state.routing_table.contains_key(&msg.k) {
+                state.routing_table.insert(msg.k, HashSet::new());
+            }
+
+            let hashset = state.routing_table.get_mut(&msg.k).unwrap();
+            hashset.insert(self.channel.address());
+        }
+    }
+
+    async fn handle_remove(self: Arc<Self>) -> Result<()> {
+        debug!("ProtocolDht::handle_remove START");
+        loop {
+            let Ok(msg) = self.remove_sub.receive().await else {
+                continue
+            };
+
+            let mut state = self.state.write().await;
+
+            if !state.routing_table.contains_key(&msg.k) {
+                continue
+            }
+
+            let hashset = state.routing_table.get_mut(&msg.k).unwrap();
+            hashset.remove(&self.channel.address());
+        }
+    }
+
+    async fn handle_chunk_request(self: Arc<Self>) -> Result<()> {
+        debug!("ProtocolDht::handle_chunk_request START");
+        loop {
+            let Ok(msg) = self.chunk_request_sub.receive().await else {
+                continue
+            };
+
+            println!("{:?}", msg);
+        }
+    }
+
+    async fn handle_chunk_reply(self: Arc<Self>) -> Result<()> {
+        debug!("ProtocolDht::handle_chunk_reply START");
+        loop {
+            let Ok(msg) = self.chunk_reply_sub.receive().await else {
+                continue
+            };
+
+            println!("{:?}", msg);
+        }
+    }
+}
+
+#[async_trait]
+impl ProtocolBase for ProtocolDht {
+    async fn start(self: Arc<Self>, ex: Arc<Executor<'_>>) -> Result<()> {
+        debug!("ProtocolDht::start()");
+        self.jobsman.clone().start(ex.clone());
+        self.jobsman.clone().spawn(self.clone().handle_insert(), ex.clone()).await;
+        self.jobsman.clone().spawn(self.clone().handle_remove(), ex.clone()).await;
+        self.jobsman.clone().spawn(self.clone().handle_chunk_request(), ex.clone()).await;
+        self.jobsman.clone().spawn(self.clone().handle_chunk_reply(), ex.clone()).await;
+        Ok(())
+    }
+
+    fn name(&self) -> &'static str {
+        "ProtoDHT"
+    }
+}

bin/dhtd/dhtd/src/tests.rs

@@ -0,0 +1,158 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2023 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 std::collections::HashMap;
+
+use async_std::{
+    fs,
+    net::TcpListener,
+    sync::{Arc, RwLock},
+};
+use darkfi::{
+    dht2::{Dht, MAX_CHUNK_SIZE},
+    net::{self, transport::TransportName, P2p},
+    util::async_util::sleep,
+    Result,
+};
+use rand::{rngs::OsRng, RngCore};
+use smol::Executor;
+use url::Url;
+
+use super::{proto::ProtocolDht, Dhtd};
+
+async fn dht_remote_get_insert_real(ex: Arc<Executor<'_>>) -> Result<()> {
+    const NET_SIZE: usize = 5;
+
+    let mut dhtds = vec![];
+    let mut base_path = std::env::temp_dir();
+    base_path.push("dht");
+
+    let mut addrs = vec![];
+    for i in 0..NET_SIZE {
+        // Find an available port
+        let listener = TcpListener::bind("127.0.0.1:0").await?;
+        let sockaddr = listener.local_addr()?;
+        let url = Url::parse(&format!("tcp://127.0.0.1:{}", sockaddr.port()))?;
+        drop(listener);
+
+        let settings = net::Settings {
+            inbound: vec![url.clone()],
+            peers: addrs.clone(),
+            outbound_transports: vec![TransportName::Tcp(None)],
+            localnet: true,
+            ..Default::default()
+        };
+
+        addrs.push(url);
+
+        let p2p = P2p::new(settings).await;
+        let mut node_path = base_path.clone();
+        node_path.push(format!("node_{}", i));
+        let dht = Dht::new(&node_path.into(), p2p.clone()).await?;
+        let dhtd = Arc::new(RwLock::new(Dhtd { dht, routing_table: HashMap::new() }));
+
+        // Register P2P protocol
+        let registry = p2p.protocol_registry();
+
+        let _dhtd = dhtd.clone();
+        registry
+            .register(net::SESSION_ALL, move |channel, p2p| {
+                let dhtd = _dhtd.clone();
+                async move { ProtocolDht::init(channel, p2p, dhtd).await.unwrap() }
+            })
+            .await;
+
+        p2p.clone().start(ex.clone()).await?;
+        let _p2p = p2p.clone();
+        let _ex = ex.clone();
+        ex.spawn(async move {
+            assert!(_p2p.run(_ex).await.is_ok());
+        })
+        .detach();
+
+        dhtds.push(dhtd);
+
+        p2p.wait_for_outbound(ex.clone()).await?;
+        sleep(1).await;
+    }
+
+    // Now the P2P network is set up. Try some stuff.
+    for dhtd in dhtds.iter_mut() {
+        dhtd.write().await.dht.garbage_collect().await?;
+    }
+
+    let dhtd = &mut dhtds[NET_SIZE - 1];
+    let rng = &mut OsRng;
+    let mut data = vec![0u8; MAX_CHUNK_SIZE];
+    rng.fill_bytes(&mut data);
+    let (file_hash, chunk_hashes) = dhtd.write().await.dht.insert(&data).await?;
+
+    for (i, node) in dhtds.iter().enumerate() {
+        if i == NET_SIZE - 1 {
+            continue
+        }
+        assert!(node.read().await.routing_table.contains_key(&file_hash));
+    }
+
+    let dhtd = &mut dhtds[NET_SIZE - 1];
+    let mut chunk_path = dhtd.read().await.dht.chunks_path();
+    chunk_path.push(chunk_hashes[0].to_hex().as_str());
+    fs::remove_file(chunk_path).await?;
+    dhtd.write().await.dht.garbage_collect().await?;
+
+    for (i, node) in dhtds.iter().enumerate() {
+        if i == NET_SIZE - 1 {
+            continue
+        }
+
+        let peers = node.read().await.routing_table.get(&file_hash).unwrap().clone();
+        assert!(peers.is_empty());
+    }
+
+    fs::remove_dir_all(base_path).await?;
+
+    Ok(())
+}
+
+#[test]
+fn dht_remote_get_insert() -> Result<()> {
+    let mut cfg = simplelog::ConfigBuilder::new();
+    cfg.add_filter_ignore("net::protocol_version".to_string());
+    cfg.add_filter_ignore("net::protocol_ping".to_string());
+
+    simplelog::TermLogger::init(
+        simplelog::LevelFilter::Info,
+        cfg.build(),
+        simplelog::TerminalMode::Mixed,
+        simplelog::ColorChoice::Auto,
+    )?;
+
+    let ex = Arc::new(Executor::new());
+    let (signal, shutdown) = async_std::channel::unbounded::<()>();
+
+    easy_parallel::Parallel::new().each(0..4, |_| smol::block_on(ex.run(shutdown.recv()))).finish(
+        || {
+            smol::block_on(async {
+                dht_remote_get_insert_real(ex.clone()).await.unwrap();
+                drop(signal);
+            })
+        },
+    );
+
+    Ok(())
+}

src/dht2/mod.rs

@@ -32,7 +32,7 @@ use log::{debug, warn};
 use crate::{net::P2pPtr, Result};
 
 /// Networked HashMap
-mod net_hashmap;
+pub mod net_hashmap;
 use net_hashmap::NetHashMap;
 
 /// Maximum size of a stored chunk (2 MiB)
@@ -388,130 +388,9 @@ impl Dht {
 
 #[cfg(test)]
 mod tests {
-    use super::{
-        net_hashmap::{NetHashMapInsert, NetHashMapRemove},
-        *,
-    };
-    use crate::{
-        net,
-        net::{
-            transport::TransportName, ChannelPtr, MessageSubscription, P2p, ProtocolBase,
-            ProtocolBasePtr, ProtocolJobsManager,
-        },
-        util::async_util::sleep,
-    };
-    use async_std::{net::TcpListener, sync::Arc};
-    use async_trait::async_trait;
-    use log::error;
+    use super::*;
+    use crate::{net, net::P2p};
     use rand::{rngs::OsRng, RngCore};
-    use smol::Executor;
-    use url::Url;
-
-    async fn create_p2p_net(n_peers: usize) -> Result<Vec<P2pPtr>> {
-        let mut ret = vec![];
-        let mut addrs = vec![];
-
-        for i in 0..n_peers {
-            // Find an available port
-            let listener = TcpListener::bind("127.0.0.1:0").await?;
-            let sockaddr = listener.local_addr()?;
-            let url = Url::parse(&format!("tcp://127.0.0.1:{}", sockaddr.port()))?;
-            drop(listener);
-
-            let mut settings = net::Settings::default();
-            settings.inbound = vec![url.clone()];
-            settings.peers = addrs.clone();
-            settings.outbound_transports = vec![TransportName::try_from("tcp").unwrap()];
-            settings.localnet = true;
-            settings.channel_log = true;
-
-            addrs.push(url);
-
-            let p2p = P2p::new(settings).await;
-            let registry = p2p.protocol_registry();
-            registry
-                .register(net::SESSION_ALL, move |channel, p2p| async move {
-                    ProtoDht::init(i, channel, p2p).await.unwrap()
-                })
-                .await;
-
-            ret.push(p2p);
-        }
-
-        Ok(ret)
-    }
-
-    struct ProtoDht {
-        jobsman: net::ProtocolJobsManagerPtr,
-        node_id: usize,
-        _channel: ChannelPtr,
-        _p2p: P2pPtr,
-        insert_sub: MessageSubscription<NetHashMapInsert<blake3::Hash, Vec<blake3::Hash>>>,
-        remove_sub: MessageSubscription<NetHashMapRemove<blake3::Hash>>,
-    }
-
-    impl ProtoDht {
-        pub async fn init(
-            node_id: usize,
-            channel: ChannelPtr,
-            p2p: P2pPtr,
-        ) -> Result<ProtocolBasePtr> {
-            let msg_subsystem = channel.get_message_subsystem();
-            msg_subsystem.add_dispatch::<NetHashMapInsert<blake3::Hash, Vec<blake3::Hash>>>().await;
-            msg_subsystem.add_dispatch::<NetHashMapRemove<blake3::Hash>>().await;
-
-            let insert_sub = channel.subscribe_msg().await?;
-            let remove_sub = channel.subscribe_msg().await?;
-
-            Ok(Arc::new(Self {
-                jobsman: ProtocolJobsManager::new("DHTProto", channel.clone()),
-                node_id,
-                _channel: channel,
-                _p2p: p2p,
-                insert_sub,
-                remove_sub,
-            }))
-        }
-
-        async fn handle_insert(self: Arc<Self>) -> Result<()> {
-            debug!("[Node {}] ProtoDht::handle_insert START", self.node_id);
-            loop {
-                let insert_message = match self.insert_sub.receive().await {
-                    Ok(v) => v,
-                    Err(_) => continue,
-                };
-
-                println!("[Node {}] {:?}", self.node_id, insert_message);
-            }
-        }
-
-        async fn handle_remove(self: Arc<Self>) -> Result<()> {
-            debug!("[Node {}] ProtoDht::handle_remove START", self.node_id);
-            loop {
-                let remove_message = match self.remove_sub.receive().await {
-                    Ok(v) => v,
-                    Err(_) => continue,
-                };
-
-                println!("[Node {}] {:?}", self.node_id, remove_message);
-            }
-        }
-    }
-
-    #[async_trait]
-    impl ProtocolBase for ProtoDht {
-        async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
-            debug!("ProtoDht::start()");
-            self.jobsman.clone().start(executor.clone());
-            self.jobsman.clone().spawn(self.clone().handle_insert(), executor.clone()).await;
-            self.jobsman.clone().spawn(self.clone().handle_remove(), executor.clone()).await;
-            Ok(())
-        }
-
-        fn name(&self) -> &'static str {
-            "ProtoDHT"
-        }
-    }
 
     #[async_std::test]
     async fn dht_local_get_insert() -> Result<()> {
@@ -559,82 +438,4 @@ mod tests {
         fs::remove_dir_all(base_path).await?;
         Ok(())
     }
-
-    async fn dht_remote_get_insert_real(executor: Arc<Executor<'_>>) -> Result<()> {
-        const NET_SIZE: usize = 5;
-
-        let peers = create_p2p_net(NET_SIZE).await?;
-
-        for p2p in &peers {
-            p2p.clone().start(executor.clone()).await?;
-
-            let _p2p = p2p.clone();
-            let _ex = executor.clone();
-            executor
-                .spawn(async move {
-                    if let Err(e) = _p2p.run(_ex).await {
-                        error!("Failed starting P2P network: {}", e);
-                        assert!(false);
-                    }
-                })
-                .detach();
-
-            p2p.clone().wait_for_outbound(executor.clone()).await?;
-        }
-        sleep(2).await;
-
-        let mut dhts = vec![];
-        let mut base_path = std::env::temp_dir();
-        base_path.push("dht");
-
-        for i in 0..NET_SIZE {
-            let mut node_path = base_path.clone();
-            node_path.push(format!("node_{}", i));
-
-            let mut dht = Dht::new(&node_path.into(), peers[i].clone()).await?;
-            dht.garbage_collect().await?;
-            dhts.push(dht);
-        }
-
-        let dht = &mut dhts[2];
-
-        let rng = &mut OsRng;
-        let mut data = vec![0u8; MAX_CHUNK_SIZE];
-        rng.fill_bytes(&mut data);
-        let (file_hash, chunk_hashes) = dht.insert(&data).await?;
-
-        fs::remove_dir_all(base_path).await?;
-
-        Ok(())
-    }
-
-    #[test]
-    fn dht_remote_get_insert() -> Result<()> {
-        // Logging
-        let mut cfg = simplelog::ConfigBuilder::new();
-        cfg.add_filter_ignore("net::protocol_version".to_string());
-        cfg.add_filter_ignore("net::protocol_ping".to_string());
-
-        simplelog::TermLogger::init(
-            //simplelog::LevelFilter::Debug,
-            simplelog::LevelFilter::Info,
-            cfg.build(),
-            simplelog::TerminalMode::Mixed,
-            simplelog::ColorChoice::Auto,
-        )?;
-
-        let ex = Arc::new(Executor::new());
-        let (signal, shutdown) = async_std::channel::unbounded::<()>();
-
-        easy_parallel::Parallel::new()
-            .each(0..4, |_| smol::block_on(ex.run(shutdown.recv())))
-            .finish(|| {
-                smol::block_on(async {
-                    dht_remote_get_insert_real(ex.clone()).await.unwrap();
-                    drop(signal);
-                })
-            });
-
-        Ok(())
-    }
 }