dht: Implement generic networked hashmap.

src/dht2/mod.rs

@@ -18,7 +18,7 @@
 
 //! Filesystem-based Distributed Hash-Table (DHT) implementation
 
-use std::collections::{HashMap, HashSet};
+use std::collections::HashSet;
 
 use async_std::{
     fs,
@@ -29,7 +29,11 @@ use async_std::{
 };
 use log::{debug, warn};
 
-use crate::Result;
+use crate::{net::P2pPtr, Result};
+
+/// Networked HashMap
+mod net_hashmap;
+use net_hashmap::NetHashMap;
 
 /// Maximum size of a stored chunk (2 MiB)
 pub const MAX_CHUNK_SIZE: usize = 2_097_152;
@@ -44,9 +48,7 @@ const CHUNKS_PATH: &str = "chunks";
 /// Files distributed on the DHT
 pub struct Dht {
     /// Map of hashed files and their (ordered) chunks
-    // TODO: This HashMap should be wrapped into an interface providing the
-    //       same API, but also broadcasts changes over P2P
-    hash_map: HashMap<blake3::Hash, Vec<blake3::Hash>>,
+    hash_map: NetHashMap<blake3::Hash, Vec<blake3::Hash>>,
     /// Path to the filesystem directory where file metadata is stored
     files_path: PathBuf,
     /// Path to the filesystem directory where the file chunks are stored
@@ -60,7 +62,7 @@ impl Dht {
     ///
     /// After the object is instantiated, the caller should also run
     /// the [`Dht::garbage_collect()`] function to ensure consistency.
-    pub async fn new(base_path: &PathBuf) -> Result<Self> {
+    pub async fn new(base_path: &PathBuf, p2p: P2pPtr) -> Result<Self> {
         let mut tmp_path: PathBuf = base_path.into();
         let mut files_path: PathBuf = base_path.into();
         let mut chunks_path: PathBuf = base_path.into();
@@ -73,7 +75,7 @@ impl Dht {
         create_dir_all(&files_path).await?;
         create_dir_all(&chunks_path).await?;
 
-        Ok(Self { hash_map: HashMap::new(), files_path, chunks_path, tmp_path })
+        Ok(Self { hash_map: NetHashMap::new(p2p), files_path, chunks_path, tmp_path })
     }
 
     /// Return the `PathBuf` where the file metadata is stored
@@ -186,7 +188,7 @@ impl Dht {
                 continue
             }
 
-            self.hash_map.insert(file_hash, chunk_hashes);
+            self.hash_map.insert(file_hash, chunk_hashes).await?;
         }
 
         // At this point we scanned through our hierarchy.
@@ -224,7 +226,7 @@ impl Dht {
             let hash_str = file_path.file_name().unwrap().to_str().unwrap();
             let file_hash = blake3::Hash::from_hex(hash_str).unwrap();
 
-            self.hash_map.remove(&file_hash);
+            self.hash_map.remove(file_hash).await?;
 
             if let Err(e) = fs::remove_file(file_path).await {
                 warn!(target: "dht", "DHT::garbage_collect(): Failed to remove corrupted file: {}", e);
@@ -279,7 +281,7 @@ impl Dht {
             file_fd.write(format!("{}\n", ch.to_hex().as_str()).as_bytes()).await?;
         }
 
-        self.hash_map.insert(file_hash, chunk_hashes.clone());
+        self.hash_map.insert(file_hash, chunk_hashes.clone()).await?;
 
         Ok((file_hash, chunk_hashes))
     }
@@ -385,13 +387,18 @@ impl Dht {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::{net, net::P2p};
     use rand::{rngs::OsRng, RngCore};
 
     #[async_std::test]
     async fn dht_local_get_insert() -> Result<()> {
         let mut base_path = std::env::temp_dir();
         base_path.push("dht");
-        let mut dht = Dht::new(&base_path.clone().into()).await?;
+
+        let settings = net::Settings::default();
+        let p2p = P2p::new(settings).await;
+
+        let mut dht = Dht::new(&base_path.clone().into(), p2p).await?;
         dht.garbage_collect().await?;
 
         let rng = &mut OsRng;

src/dht2/net_hashmap.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::{
+    borrow::Borrow,
+    collections::{
+        hash_map::{Iter, Keys, Values},
+        HashMap,
+    },
+    hash::Hash,
+};
+
+use darkfi_serial::{Decodable, Encodable, SerialDecodable, SerialEncodable};
+
+use crate::{net, net::P2pPtr, Result};
+
+/// A general networked hashmap. Propagates changes over P2P.
+#[derive(Clone)]
+pub struct NetHashMap<K, V> {
+    /// The internal [`HashMap`] that represents the actual state
+    hashmap: HashMap<K, V>,
+    /// Pointer to the P2P network
+    p2p: P2pPtr,
+}
+
+impl<K, V> NetHashMap<K, V> {
+    /// Instantiate a new [`NetHashMap`] with the given [`P2pPtr`]
+    pub fn new(p2p: P2pPtr) -> Self {
+        let hashmap = HashMap::new();
+
+        Self { hashmap, p2p }
+    }
+}
+
+impl<K, V> NetHashMap<K, V>
+where
+    K: Eq + Hash + Send + Sync + Encodable + Decodable + Clone + 'static,
+    V: Send + Sync + Encodable + Decodable + Clone + 'static,
+{
+    /// Returns `true` if the map contains a value for the specified key.
+    #[allow(dead_code)]
+    pub fn contains_key<Q: ?Sized>(&self, k: &Q) -> bool
+    where
+        K: Borrow<Q>,
+        Q: Hash + Eq,
+    {
+        self.hashmap.contains_key(k)
+    }
+
+    /// Returns `true` if the map contains no elements.
+    #[allow(dead_code)]
+    pub fn is_empty(&self) -> bool {
+        self.hashmap.is_empty()
+    }
+
+    /// Returns the number of elements in the map.
+    #[allow(dead_code)]
+    pub fn len(&self) -> usize {
+        self.hashmap.len()
+    }
+
+    /// Insert a key-value pair into the map.
+    ///
+    /// If the map did not have this key present, `None` is returned.
+    ///
+    /// If the map did have this key present, the value is updated, and
+    /// the old value is returned.
+    ///
+    /// Additionally, this change will be broadcasted to the P2P network.
+    pub async fn insert(&mut self, k: K, v: V) -> Result<Option<V>> {
+        let message = NetHashMapInsert { k: k.clone(), v: v.clone() };
+        self.p2p.broadcast(message).await?;
+        Ok(self.hashmap.insert(k, v))
+    }
+
+    /// Removes a key from the map, returning the value at the key if the key
+    /// was previously in the map.
+    ///
+    /// Additionally, this change will be broadcasted to the P2P network.
+    pub async fn remove<Q: Encodable + Decodable + ?Sized + Clone>(
+        &mut self,
+        k: Q,
+    ) -> Result<Option<V>>
+    where
+        K: Borrow<Q>,
+        Q: Hash + Eq + Send + Sync + Encodable + Decodable + 'static,
+    {
+        let message = NetHashMapRemove { k: k.clone() };
+        self.p2p.broadcast(message).await?;
+        Ok(self.hashmap.remove(&k))
+    }
+
+    /// An iterator visiting all key-value pairs in arbitrary order.
+    /// The iterator element type is `(&'a K, &'a V)`.
+    #[allow(dead_code)]
+    pub fn iter(&self) -> Iter<'_, K, V> {
+        self.hashmap.iter()
+    }
+
+    /// An iterator visiting all keys in arbitrary order.
+    /// The iterator element type is `&'a K`.
+    #[allow(dead_code)]
+    pub fn keys(&self) -> Keys<'_, K, V> {
+        self.hashmap.keys()
+    }
+
+    /// An iterator visiting all values in arbitrary order.
+    /// The iterator element type is `&'a V`.
+    #[allow(dead_code)]
+    pub fn values(&self) -> Values<'_, K, V> {
+        self.hashmap.values()
+    }
+}
+
+#[derive(Debug, Clone, SerialDecodable, SerialEncodable)]
+pub struct NetHashMapInsert<K, V> {
+    pub k: K,
+    pub v: V,
+}
+
+impl<K, V> net::Message for NetHashMapInsert<K, V>
+where
+    K: Encodable + Decodable + Send + Sync + 'static,
+    V: Encodable + Decodable + Send + Sync + 'static,
+{
+    fn name() -> &'static str {
+        "nethashmap_insert"
+    }
+}
+
+#[derive(Debug, Clone, SerialDecodable, SerialEncodable)]
+pub struct NetHashMapRemove<K> {
+    pub k: K,
+}
+
+impl<K> net::Message for NetHashMapRemove<K>
+where
+    K: Encodable + Decodable + Send + Sync + 'static,
+{
+    fn name() -> &'static str {
+        "nethashmap_remove"
+    }
+}