Merge remote-tracking branch 'origin/main' into sync-no-deadlocks

iroh-bytes/Cargo.toml

@@ -30,13 +30,14 @@ postcard = { version = "1", default-features = false, features = ["alloc", "use-
 quinn = "0.10"
 rand = "0.8"
 range-collections = "0.4.0"
+reflink-copy = { version = "0.1.8", optional = true }
 self_cell = "1.0.1"
 serde = { version = "1", features = ["derive"] }
 serde-error = "0.1.2"
 smallvec = { version = "1.10.0", features = ["serde", "const_new"] }
 subtle = "2.4"
 thiserror = "1"
-tokio = { version = "1", features = [] }
+tokio = { version = "1" }
 tokio-util = { version = "0.7", features = ["io-util", "io", "rt"] }
 tracing = "0.1"
 tracing-futures = "0.2.5"
@@ -49,4 +50,5 @@ serde_test = "1.0.176"
 tokio = { version = "1", features = ["macros", "test-util"] }
 
 [features]
-default = []
+default = ["flat-db"]
+flat-db = ["reflink-copy", "tokio/fs"]

iroh-bytes/src/get.rs

@@ -14,7 +14,7 @@ use std::error::Error;
 use std::fmt::{self, Debug};
 use std::time::{Duration, Instant};
 
-use crate::util::Hash;
+use crate::Hash;
 use anyhow::Result;
 use bao_tree::io::fsm::BaoContentItem;
 use bao_tree::ChunkNum;

iroh-bytes/src/hashseq.rs

@@ -1,5 +1,5 @@
 //! traits related to collections of blobs
-use crate::util::Hash;
+use crate::Hash;
 use bytes::Bytes;
 use iroh_io::{AsyncSliceReader, AsyncSliceReaderExt};
 use std::{fmt::Debug, io};

iroh-bytes/src/lib.rs

@@ -3,14 +3,14 @@
 #![deny(missing_docs, rustdoc::broken_intra_doc_links)]
 #![recursion_limit = "256"]
 
-pub mod baomap;
 pub mod get;
 pub mod hashseq;
 pub mod protocol;
 pub mod provider;
+pub mod store;
 pub mod util;
 
-pub use crate::util::Hash;
+pub use crate::util::{BlobFormat, Hash, HashAndFormat, Tag, TempTag};
 use bao_tree::BlockSize;
 
 /// Block size used by iroh, 2^4*1024 = 16KiB

iroh-bytes/src/protocol.rs

@@ -356,7 +356,7 @@ use serde::{Deserialize, Serialize};
 mod range_spec;
 pub use range_spec::{NonEmptyRequestRangeSpecIter, RangeSpec, RangeSpecSeq};
 
-use crate::util::Hash;
+use crate::Hash;
 
 /// Maximum message size is limited to 100MiB for now.
 pub const MAX_MESSAGE_SIZE: usize = 1024 * 1024 * 100;

iroh-bytes/src/provider.rs

@@ -14,9 +14,9 @@ use serde::{Deserialize, Serialize};
 use tracing::{debug, debug_span, info, trace, warn};
 use tracing_futures::Instrument;
 
-use crate::baomap::*;
 use crate::hashseq::parse_hash_seq;
 use crate::protocol::{GetRequest, RangeSpec, Request, RequestToken};
+use crate::store::*;
 use crate::util::{BlobFormat, RpcError, Tag};
 use crate::Hash;
 
@@ -532,7 +532,7 @@ impl<E: EventSender> ResponseWriter<E> {
         let other_duration = total_duration
             .saturating_sub(send_duration)
             .saturating_sub(read_duration);
-        let avg_send_size = total_sent_bytes / send.stats.count;
+        let avg_send_size = total_sent_bytes.checked_div(send.stats.count).unwrap_or(0);
         info!(
             "sent {} bytes in {}s",
             total_sent_bytes,

iroh/src/baomap.rs → iroh-bytes/src/store.rs

@@ -1,9 +1,16 @@
-//! Various database implementations for storing blob data
+//! Implementations of blob stores
+use crate::{
+    util::{BlobFormat, HashAndFormat},
+    Hash,
+};
+pub mod mem;
+pub mod readonly_mem;
+
 #[cfg(feature = "flat-db")]
 pub mod flat;
-pub mod mem;
 
-pub mod readonly_mem;
+mod traits;
+pub use traits::*;
 
 fn flatten_to_io<T>(
     e: std::result::Result<std::io::Result<T>, tokio::task::JoinError>,
@@ -34,19 +41,19 @@ struct TempCounters {
 }
 
 impl TempCounters {
-    fn counter(&mut self, format: iroh_bytes::util::BlobFormat) -> &mut u64 {
+    fn counter(&mut self, format: BlobFormat) -> &mut u64 {
         match format {
-            iroh_bytes::util::BlobFormat::Raw => &mut self.raw,
-            iroh_bytes::util::BlobFormat::HashSeq => &mut self.hash_seq,
+            BlobFormat::Raw => &mut self.raw,
+            BlobFormat::HashSeq => &mut self.hash_seq,
         }
     }
 
-    fn inc(&mut self, format: iroh_bytes::util::BlobFormat) {
+    fn inc(&mut self, format: BlobFormat) {
         let counter = self.counter(format);
         *counter = counter.checked_add(1).unwrap();
     }
 
-    fn dec(&mut self, format: iroh_bytes::util::BlobFormat) {
+    fn dec(&mut self, format: BlobFormat) {
         let counter = self.counter(format);
         *counter = counter.saturating_sub(1);
     }
@@ -57,35 +64,35 @@ impl TempCounters {
 }
 
 #[derive(Debug, Clone, Default)]
-struct TempCounterMap(std::collections::BTreeMap<iroh_bytes::Hash, TempCounters>);
+struct TempCounterMap(std::collections::BTreeMap<Hash, TempCounters>);
 
 impl TempCounterMap {
-    fn inc(&mut self, value: &iroh_bytes::util::HashAndFormat) {
-        let iroh_bytes::util::HashAndFormat { hash, format } = value;
+    fn inc(&mut self, value: &HashAndFormat) {
+        let HashAndFormat { hash, format } = value;
         self.0.entry(*hash).or_default().inc(*format)
     }
 
-    fn dec(&mut self, value: &iroh_bytes::util::HashAndFormat) {
-        let iroh_bytes::util::HashAndFormat { hash, format } = value;
+    fn dec(&mut self, value: &HashAndFormat) {
+        let HashAndFormat { hash, format } = value;
         let counters = self.0.get_mut(hash).unwrap();
         counters.dec(*format);
         if counters.is_empty() {
             self.0.remove(hash);
         }
     }
 
-    fn contains(&self, hash: &iroh_bytes::Hash) -> bool {
+    fn contains(&self, hash: &Hash) -> bool {
         self.0.contains_key(hash)
     }
 
-    fn keys(&self) -> impl Iterator<Item = iroh_bytes::util::HashAndFormat> {
+    fn keys(&self) -> impl Iterator<Item = HashAndFormat> {
         let mut res = Vec::new();
         for (k, v) in self.0.iter() {
             if v.raw > 0 {
-                res.push(iroh_bytes::util::HashAndFormat::raw(*k));
+                res.push(HashAndFormat::raw(*k));
             }
             if v.hash_seq > 0 {
-                res.push(iroh_bytes::util::HashAndFormat::hash_seq(*k));
+                res.push(HashAndFormat::hash_seq(*k));
             }
         }
         res.into_iter()

iroh/src/baomap/flat.rs → iroh-bytes/src/store/flat.rs

@@ -130,6 +130,13 @@ use std::str::FromStr;
 use std::sync::{Arc, Mutex, RwLock};
 use std::time::SystemTime;
 
+use super::{
+    EntryStatus, ExportMode, ImportMode, ImportProgress, Map, MapEntry, PartialMap,
+    PartialMapEntry, ReadableStore, ValidateProgress,
+};
+use crate::util::progress::{IdGenerator, IgnoreProgressSender, ProgressSender};
+use crate::util::{BlobFormat, HashAndFormat, LivenessTracker, Tag};
+use crate::{Hash, TempTag, IROH_BLOCK_SIZE};
 use bao_tree::io::outboard::{PostOrderMemOutboard, PreOrderOutboard};
 use bao_tree::io::sync::ReadAt;
 use bao_tree::{blake3, ChunkRanges};
@@ -138,13 +145,6 @@ use bytes::Bytes;
 use futures::future::BoxFuture;
 use futures::future::Either;
 use futures::{Future, FutureExt, Stream, StreamExt};
-use iroh_bytes::baomap::{
-    self, EntryStatus, ExportMode, ImportMode, ImportProgress, LivenessTracker, Map, MapEntry,
-    PartialMap, PartialMapEntry, ReadableStore, TempTag, ValidateProgress,
-};
-use iroh_bytes::util::progress::{IdGenerator, IgnoreProgressSender, ProgressSender};
-use iroh_bytes::util::{BlobFormat, HashAndFormat, Tag};
-use iroh_bytes::{Hash, IROH_BLOCK_SIZE};
 use iroh_io::{AsyncSliceReader, AsyncSliceWriter, File};
 use tokio::io::AsyncWriteExt;
 use tokio::sync::mpsc;
@@ -685,7 +685,7 @@ impl ReadableStore for Store {
     }
 }
 
-impl baomap::Store for Store {
+impl super::Store for Store {
     fn import_file(
         &self,
         path: PathBuf,
@@ -916,7 +916,7 @@ impl Store {
             Ok(progress2.try_send(ImportProgress::OutboardProgress { id, offset })?)
         })?;
         progress.blocking_send(ImportProgress::OutboardDone { id, hash })?;
-        use baomap::Store;
+        use super::Store;
         // from here on, everything related to the hash is protected by the temp tag
         let tag = self.temp_tag(HashAndFormat { hash, format });
         let hash = *tag.hash();
@@ -1195,7 +1195,7 @@ impl Store {
         complete_path: PathBuf,
         partial_path: PathBuf,
         meta_path: PathBuf,
-        rt: iroh_bytes::util::runtime::Handle,
+        rt: crate::util::runtime::Handle,
     ) -> anyhow::Result<Self> {
         tracing::debug!(
             "loading database from {} {}",
@@ -1465,7 +1465,7 @@ impl Store {
         complete_path: impl AsRef<Path>,
         partial_path: impl AsRef<Path>,
         meta_path: impl AsRef<Path>,
-        rt: &iroh_bytes::util::runtime::Handle,
+        rt: &crate::util::runtime::Handle,
     ) -> anyhow::Result<Self> {
         let complete_path = complete_path.as_ref().to_path_buf();
         let partial_path = partial_path.as_ref().to_path_buf();
@@ -1480,7 +1480,7 @@ impl Store {
         complete_path: impl AsRef<Path>,
         partial_path: impl AsRef<Path>,
         meta_path: impl AsRef<Path>,
-        rt: &iroh_bytes::util::runtime::Handle,
+        rt: &crate::util::runtime::Handle,
     ) -> anyhow::Result<Self> {
         let complete_path = complete_path.as_ref().to_path_buf();
         let partial_path = partial_path.as_ref().to_path_buf();

iroh/src/baomap/mem.rs → iroh-bytes/src/store/mem.rs

@@ -15,6 +15,17 @@ use std::time::SystemTime;
 use super::flatten_to_io;
 use super::temp_name;
 use super::TempCounterMap;
+use crate::{
+    store::{
+        EntryStatus, ExportMode, ImportMode, ImportProgress, Map, MapEntry, PartialMap,
+        PartialMapEntry, ReadableStore, ValidateProgress,
+    },
+    util::{
+        progress::{IdGenerator, IgnoreProgressSender, ProgressSender},
+        runtime, BlobFormat, HashAndFormat, LivenessTracker,
+    },
+    Hash, Tag, TempTag, IROH_BLOCK_SIZE,
+};
 use bao_tree::blake3;
 use bao_tree::io::fsm::Outboard;
 use bao_tree::io::outboard::PreOrderOutboard;
@@ -28,27 +39,7 @@ use derive_more::From;
 use futures::future::BoxFuture;
 use futures::FutureExt;
 use futures::{Stream, StreamExt};
-use iroh_bytes::baomap;
-use iroh_bytes::baomap::EntryStatus;
-use iroh_bytes::baomap::ExportMode;
-use iroh_bytes::baomap::ImportMode;
-use iroh_bytes::baomap::ImportProgress;
-use iroh_bytes::baomap::LivenessTracker;
-use iroh_bytes::baomap::PartialMap;
-use iroh_bytes::baomap::PartialMapEntry;
-use iroh_bytes::baomap::TempTag;
-use iroh_bytes::baomap::ValidateProgress;
-use iroh_bytes::baomap::{Map, MapEntry, ReadableStore};
-use iroh_bytes::util::progress::IdGenerator;
-use iroh_bytes::util::progress::IgnoreProgressSender;
-use iroh_bytes::util::progress::ProgressSender;
-use iroh_bytes::util::runtime;
-use iroh_bytes::util::BlobFormat;
-use iroh_bytes::util::HashAndFormat;
-use iroh_bytes::util::Tag;
-use iroh_bytes::{Hash, IROH_BLOCK_SIZE};
-use iroh_io::AsyncSliceReader;
-use iroh_io::AsyncSliceWriter;
+use iroh_io::{AsyncSliceReader, AsyncSliceWriter};
 use tokio::sync::mpsc;
 
 /// A mutable file like object that can be used for partial entries.
@@ -458,7 +449,7 @@ impl PartialMap for Store {
     }
 }
 
-impl baomap::Store for Store {
+impl super::Store for Store {
     fn import_file(
         &self,
         path: std::path::PathBuf,
@@ -619,7 +610,7 @@ impl Store {
             data: outboard.into(),
         };
         let hash = hash.into();
-        use baomap::Store;
+        use super::Store;
         let tag = self.temp_tag(HashAndFormat { hash, format });
         self.0
             .state

iroh/src/baomap/readonly_mem.rs → iroh-bytes/src/store/readonly_mem.rs

@@ -8,6 +8,17 @@ use std::{
     sync::Arc,
 };
 
+use crate::{
+    store::{
+        EntryStatus, ExportMode, ImportMode, ImportProgress, Map, MapEntry, PartialMap,
+        PartialMapEntry, ReadableStore, ValidateProgress,
+    },
+    util::{
+        progress::{IdGenerator, ProgressSender},
+        BlobFormat, HashAndFormat, Tag,
+    },
+    Hash, TempTag, IROH_BLOCK_SIZE,
+};
 use bao_tree::{
     blake3,
     io::{
@@ -21,17 +32,6 @@ use futures::{
     future::{self, BoxFuture},
     FutureExt, Stream,
 };
-use iroh_bytes::{
-    baomap::{
-        self, EntryStatus, ExportMode, ImportMode, ImportProgress, Map, MapEntry, PartialMap,
-        PartialMapEntry, ReadableStore, TempTag, ValidateProgress,
-    },
-    util::{
-        progress::{IdGenerator, ProgressSender},
-        BlobFormat, HashAndFormat, Tag,
-    },
-    Hash, IROH_BLOCK_SIZE,
-};
 use tokio::{io::AsyncWriteExt, sync::mpsc};
 
 /// A readonly in memory database for iroh-bytes.
@@ -320,7 +320,7 @@ impl PartialMapEntry<Store> for PartialEntry {
     }
 }
 
-impl baomap::Store for Store {
+impl super::Store for Store {
     fn import_file(
         &self,
         data: PathBuf,