Mmap subcrate refactoring (#4886)

* make mmap_type independent from segment structures

* make bitvec and thiserror workspace dependencies

* move mmap_type into common/memory subcrate

* fmt

Cargo.lock

@@ -3529,10 +3529,14 @@ dependencies = [
 name = "memory"
 version = "0.0.0"
 dependencies = [
+ "bitvec",
  "log",
  "memmap2 0.9.4",
  "parking_lot",
+ "rand 0.8.5",
  "serde",
+ "tempfile",
+ "thiserror",
 ]
 
 [[package]]

Cargo.toml

@@ -49,7 +49,7 @@ rusty-hook = "^0.11.2"
 [dependencies]
 parking_lot = { workspace = true, optional = true }
 
-thiserror = "1.0"
+thiserror = { workspace = true }
 log = { workspace = true }
 colored = "2"
 serde = { workspace = true }
@@ -196,6 +196,8 @@ wal = { git = "https://github.com/qdrant/wal.git", rev = "a7870900f29811a24e2088
 zerocopy = { version = "0.7.34", features = ["derive"] }
 atomic_refcell = "0.1.13"
 byteorder = "1.5.0"
+thiserror = "1.0.63"
+bitvec = "1.0.1"
 
 [[bin]]
 name = "schema_generator"

lib/api/Cargo.toml

@@ -25,7 +25,7 @@ uuid = { workspace = true }
 tokio = { workspace = true }
 rand = { workspace = true }
 chrono = { workspace = true }
-thiserror = "1.0"
+thiserror = { workspace = true }
 parking_lot = { workspace = true }
 validator = { workspace = true }
 itertools = { workspace = true }

lib/collection/Cargo.toml

@@ -32,7 +32,7 @@ pprof = { workspace = true }
 parking_lot = { workspace = true }
 
 rand = { workspace = true }
-thiserror = "1.0"
+thiserror = { workspace = true }
 serde = { workspace = true }
 serde_cbor = { workspace = true }
 serde_json = { workspace = true }
@@ -42,7 +42,7 @@ wal = { workspace = true }
 ordered-float = "4.2"
 hashring = "0.3.6"
 tinyvec = { version = "1.8.0", features = ["alloc"] }
-bitvec = "1.0.1"
+bitvec = { workspace = true }
 lazy_static = "1.5.0"
 smallvec = "1.13.2"
 
@@ -60,7 +60,7 @@ uuid = { workspace = true }
 url = { version = "2", features = ["serde"] }
 validator = { workspace = true }
 actix-web-validator = "5.0.1"
-actix-web = {version = "4.9.0"}
+actix-web = { version = "4.9.0" }
 actix-files = "0.6.6"
 
 common = { path = "../common/common" }

lib/common/cancel/Cargo.toml

@@ -13,6 +13,6 @@ publish = false
 workspace = true
 
 [dependencies]
-thiserror = "1.0"
+thiserror = { workspace = true }
 tokio = { workspace = true }
 tokio-util = { workspace = true }

lib/common/common/Cargo.toml

@@ -34,7 +34,7 @@ common = { path = ".", features = ["testing"] }
 criterion = "0.5"
 
 [target.'cfg(target_os = "linux")'.dependencies]
-thiserror = "1.0"
+thiserror = { workspace = true }
 thread-priority = "1.1"
 
 [[bench]]

lib/common/io/Cargo.toml

@@ -18,4 +18,4 @@ bincode = "1.3.3"
 semver = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
-thiserror = "1.0.63"
+thiserror = { workspace = true }

lib/common/memory/Cargo.toml

@@ -17,3 +17,9 @@ memmap2 = { workspace = true }
 log = { workspace = true }
 parking_lot = { workspace = true }
 serde = { workspace = true }
+bitvec = { workspace = true }
+thiserror = { workspace = true }
+
+[dev-dependencies]
+rand = { workspace = true }
+tempfile = { workspace = true }

lib/common/memory/src/lib.rs

@@ -1,2 +1,3 @@
 pub mod madvise;
 pub mod mmap_ops;
+pub mod mmap_type;

lib/common/memory/src/mmap_type.rs

@@ -29,11 +29,11 @@ use std::{fmt, mem, slice};
 use bitvec::slice::BitSlice;
 use memmap2::MmapMut;
 
-use crate::common::Flusher;
-
 /// Result for mmap errors.
 type Result<T> = std::result::Result<T, Error>;
 
+pub type MmapFlusher = Box<dyn FnOnce() -> Result<()> + Send>;
+
 /// Type `T` on a memory mapped file
 ///
 /// Functions as if it is `T` because this implements [`Deref`] and [`DerefMut`].
@@ -155,7 +155,7 @@ where
     T: ?Sized + 'static,
 {
     /// Get flusher to explicitly flush mmap at a later time
-    pub fn flusher(&self) -> Flusher {
+    pub fn flusher(&self) -> MmapFlusher {
         // TODO: if we explicitly flush when dropping this type, we can switch to a weak reference
         // here to only flush if it hasn't been done already
         Box::new({
@@ -253,7 +253,7 @@ impl<T> MmapSlice<T> {
     }
 
     /// Get flusher to explicitly flush mmap at a later time
-    pub fn flusher(&self) -> Flusher {
+    pub fn flusher(&self) -> MmapFlusher {
         self.mmap.flusher()
     }
 }
@@ -320,7 +320,7 @@ impl MmapBitSlice {
     }
 
     /// Get flusher to explicitly flush mmap at a later time
-    pub fn flusher(&self) -> Flusher {
+    pub fn flusher(&self) -> MmapFlusher {
         self.mmap.flusher()
     }
 }
@@ -340,12 +340,14 @@ impl DerefMut for MmapBitSlice {
 }
 
 /// Typed mmap errors.
-#[derive(thiserror::Error, Clone, Debug)]
+#[derive(thiserror::Error, Debug)]
 pub enum Error {
     #[error("Mmap length must be {0} to match the size of type, but it is {1}")]
     SizeExact(usize, usize),
     #[error("Mmap length must be multiple of {0} to match the size of type, but it is {1}")]
     SizeMultiple(usize, usize),
+    #[error("{0}")]
+    Io(#[from] std::io::Error),
 }
 
 /// Get a second mutable reference for type `T` from the given mmap
@@ -468,12 +470,12 @@ mod tests {
     use std::fmt::Debug;
     use std::iter;
 
-    use memory::mmap_ops;
     use rand::rngs::StdRng;
     use rand::{Rng, SeedableRng};
     use tempfile::{Builder, NamedTempFile};
 
     use super::*;
+    use crate::mmap_ops;
 
     fn create_temp_mmap_file(len: usize) -> NamedTempFile {
         let tempfile = Builder::new()

lib/segment/Cargo.toml

@@ -52,7 +52,7 @@ serde-value = "0.7"
 serde_variant = { workspace = true }
 serde-untagged = "0.1.6"
 ordered-float = "4.2"
-thiserror = "1.0"
+thiserror = { workspace = true }
 atomic_refcell = { workspace = true }
 atomicwrites = "0.4.3"
 memmap2 = { workspace = true }
@@ -64,7 +64,7 @@ num-traits = { workspace = true }
 num-derive = "0.4.2"
 num-cmp = "0.1.0"
 rand = { workspace = true }
-bitvec = "1.0.1"
+bitvec = { workspace = true }
 seahash = "4.1.0"
 semver = { workspace = true }
 tar = { workspace = true }

lib/segment/src/common/mmap_bitslice_buffered_update_wrapper.rs

@@ -2,9 +2,9 @@ use std::collections::HashMap;
 use std::mem;
 use std::sync::Arc;
 
+use memory::mmap_type::MmapBitSlice;
 use parking_lot::{Mutex, RwLock};
 
-use crate::common::mmap_type::MmapBitSlice;
 use crate::common::Flusher;
 
 /// A wrapper around `MmapBitSlice` that delays writing changes to the underlying file until they get
@@ -63,7 +63,7 @@ impl MmapBitSliceBufferedUpdateWrapper {
             for (index, value) in pending_updates {
                 mmap_slice_write.set(index, value);
             }
-            mmap_slice_write.flusher()()
+            Ok(mmap_slice_write.flusher()()?)
         })
     }
 }

lib/segment/src/common/mmap_slice_buffered_update_wrapper.rs

@@ -2,9 +2,9 @@ use std::collections::HashMap;
 use std::mem;
 use std::sync::Arc;
 
+use memory::mmap_type::MmapSlice;
 use parking_lot::{Mutex, RwLock};
 
-use crate::common::mmap_type::MmapSlice;
 use crate::common::Flusher;
 
 /// A wrapper around `MmapSlice` that delays writing changes to the underlying file until they get
@@ -55,7 +55,7 @@ where
             for (index, value) in pending_updates {
                 mmap_slice_write[index] = value;
             }
-            mmap_slice_write.flusher()()
+            Ok(mmap_slice_write.flusher()()?)
         })
     }
 }

lib/segment/src/common/mod.rs

@@ -3,7 +3,6 @@ pub mod error_logging;
 pub mod macros;
 pub mod mmap_bitslice_buffered_update_wrapper;
 pub mod mmap_slice_buffered_update_wrapper;
-pub mod mmap_type;
 pub mod operation_error;
 pub mod operation_time_statistics;
 pub mod reciprocal_rank_fusion;

lib/segment/src/common/operation_error.rs

@@ -5,10 +5,10 @@ use std::sync::atomic::{AtomicBool, Ordering};
 
 use atomicwrites::Error as AtomicIoError;
 use io::file_operations::FileStorageError;
+use memory::mmap_type::Error as MmapError;
 use rayon::ThreadPoolBuildError;
 use thiserror::Error;
 
-use crate::common::mmap_type::Error as MmapError;
 use crate::types::{PayloadKeyType, PointIdType, SeqNumberType};
 use crate::utils::mem::Mem;
 

lib/segment/src/id_tracker/immutable_id_tracker.rs

@@ -9,11 +9,11 @@ use bitvec::vec::BitVec;
 use byteorder::{ReadBytesExt, WriteBytesExt};
 use common::types::PointOffsetType;
 use memory::mmap_ops::{create_and_ensure_length, open_write_mmap};
+use memory::mmap_type::{MmapBitSlice, MmapSlice};
 use uuid::Uuid;
 
 use crate::common::mmap_bitslice_buffered_update_wrapper::MmapBitSliceBufferedUpdateWrapper;
 use crate::common::mmap_slice_buffered_update_wrapper::MmapSliceBufferedUpdateWrapper;
-use crate::common::mmap_type::{MmapBitSlice, MmapSlice};
 use crate::common::operation_error::{OperationError, OperationResult};
 use crate::common::Flusher;
 use crate::id_tracker::in_memory_id_tracker::InMemoryIdTracker;

lib/segment/src/index/field_index/map_index/mmap_map_index.rs

@@ -10,11 +10,11 @@ use common::types::PointOffsetType;
 use io::file_operations::{atomic_save_json, read_json};
 use memmap2::MmapMut;
 use memory::mmap_ops::{self, create_and_ensure_length};
+use memory::mmap_type::MmapBitSlice;
 use serde::{Deserialize, Serialize};
 
 use super::{IdRefIter, MapIndexKey};
 use crate::common::mmap_bitslice_buffered_update_wrapper::MmapBitSliceBufferedUpdateWrapper;
-use crate::common::mmap_type::MmapBitSlice;
 use crate::common::operation_error::OperationResult;
 use crate::common::Flusher;
 use crate::index::field_index::mmap_point_to_values::MmapPointToValues;

lib/segment/src/index/field_index/numeric_index/mmap_numeric_index.rs

@@ -6,12 +6,12 @@ use common::types::PointOffsetType;
 use io::file_operations::{atomic_save_json, read_json};
 use memmap2::MmapMut;
 use memory::mmap_ops::{self, create_and_ensure_length};
+use memory::mmap_type::{MmapBitSlice, MmapSlice};
 use serde::{Deserialize, Serialize};
 
 use super::mutable_numeric_index::DynamicNumericIndex;
 use super::Encodable;
 use crate::common::mmap_bitslice_buffered_update_wrapper::MmapBitSliceBufferedUpdateWrapper;
-use crate::common::mmap_type::{MmapBitSlice, MmapSlice};
 use crate::common::operation_error::OperationResult;
 use crate::common::Flusher;
 use crate::index::field_index::histogram::{Histogram, Numericable, Point};

lib/segment/src/vector_storage/chunked_mmap_vectors.rs

@@ -5,10 +5,10 @@ use std::path::{Path, PathBuf};
 
 use memmap2::MmapMut;
 use memory::mmap_ops::{create_and_ensure_length, open_write_mmap};
+use memory::mmap_type::MmapType;
 use num_traits::AsPrimitive;
 use serde::{Deserialize, Serialize};
 
-use crate::common::mmap_type::MmapType;
 use crate::common::operation_error::{OperationError, OperationResult};
 use crate::common::Flusher;
 use crate::vector_storage::chunked_utils::{chunk_name, create_chunk, read_mmaps, MmapChunk};

lib/segment/src/vector_storage/chunked_utils.rs

@@ -2,8 +2,8 @@ use std::collections::HashMap;
 use std::path::{Path, PathBuf};
 
 use memory::mmap_ops::{create_and_ensure_length, open_write_mmap};
+use memory::mmap_type::MmapSlice;
 
-use crate::common::mmap_type::MmapSlice;
 use crate::common::operation_error::{OperationError, OperationResult};
 
 const MMAP_CHUNKS_PATTERN_START: &str = "chunk_";

lib/segment/src/vector_storage/dense/dynamic_mmap_flags.rs

@@ -6,10 +6,10 @@ use std::{fmt, fs};
 use bitvec::prelude::BitSlice;
 use memmap2::MmapMut;
 use memory::mmap_ops::{create_and_ensure_length, open_write_mmap};
+use memory::mmap_type::{MmapBitSlice, MmapFlusher, MmapType};
 use parking_lot::Mutex;
 
 use crate::common::error_logging::LogError;
-use crate::common::mmap_type::{MmapBitSlice, MmapType};
 use crate::common::operation_error::{OperationError, OperationResult};
 use crate::common::Flusher;
 
@@ -80,7 +80,7 @@ pub struct DynamicMmapFlags {
     /// Current mmap'ed BitSlice for flags
     flags: MmapBitSlice,
     /// Flusher to flush current flags mmap
-    flags_flusher: Arc<Mutex<Option<Flusher>>>,
+    flags_flusher: Arc<Mutex<Option<MmapFlusher>>>,
     status: MmapType<DynamicMmapStatus>,
     directory: PathBuf,
 }
@@ -141,7 +141,7 @@ impl DynamicMmapFlags {
         num_flags: usize,
         directory: &Path,
         new_file_id: FileId,
-    ) -> OperationResult<(MmapBitSlice, Flusher)> {
+    ) -> OperationResult<(MmapBitSlice, MmapFlusher)> {
         let capacity_bytes = mmap_capacity_bytes(num_flags);
         let mmap_path = Self::file_id_to_file(directory, new_file_id);
         create_and_ensure_length(&mmap_path, capacity_bytes)?;

lib/segment/src/vector_storage/dense/memmap_dense_vector_storage.rs

@@ -9,7 +9,7 @@ use bitvec::prelude::BitSlice;
 use common::types::PointOffsetType;
 use memory::mmap_ops;
 
-use crate::common::operation_error::{check_process_stopped, OperationResult};
+use crate::common::operation_error::{check_process_stopped, OperationError, OperationResult};
 use crate::common::Flusher;
 use crate::data_types::named_vectors::CowVector;
 use crate::data_types::primitive::PrimitiveVectorElement;
@@ -229,7 +229,10 @@ impl<T: PrimitiveVectorElement> VectorStorage for MemmapDenseVectorStorage<T> {
 
     fn flusher(&self) -> Flusher {
         match &self.mmap_store {
-            Some(mmap_store) => mmap_store.flusher(),
+            Some(mmap_store) => {
+                let mmap_flusher = mmap_store.flusher();
+                Box::new(move || mmap_flusher().map_err(OperationError::from))
+            }
             None => Box::new(|| Ok(())),
         }
     }

lib/segment/src/vector_storage/dense/mmap_dense_vectors.rs

@@ -8,12 +8,11 @@ use bitvec::prelude::BitSlice;
 use common::types::PointOffsetType;
 use memmap2::Mmap;
 use memory::mmap_ops;
+use memory::mmap_type::{MmapBitSlice, MmapFlusher};
 use parking_lot::Mutex;
 
 use crate::common::error_logging::LogError;
-use crate::common::mmap_type::MmapBitSlice;
 use crate::common::operation_error::OperationResult;
-use crate::common::Flusher;
 use crate::data_types::primitive::PrimitiveVectorElement;
 #[cfg(target_os = "linux")]
 use crate::vector_storage::async_io::UringReader;
@@ -95,7 +94,7 @@ impl<T: PrimitiveVectorElement> MmapDenseVectors<T> {
         self.uring_reader.lock().is_some()
     }
 
-    pub fn flusher(&self) -> Flusher {
+    pub fn flusher(&self) -> MmapFlusher {
         self.deleted.flusher()
     }
 

lib/segment/src/vector_storage/quantized/quantized_multivector_storage.rs

@@ -5,10 +5,10 @@ use std::path::{Path, PathBuf};
 
 use common::types::{PointOffsetType, ScoreType};
 use memmap2::MmapMut;
+use memory::mmap_type::MmapSlice;
 use quantization::{EncodedVectors, VectorParameters};
 use serde::{Deserialize, Serialize};
 
-use crate::common::mmap_type::MmapSlice;
 use crate::common::operation_error::OperationResult;
 use crate::data_types::vectors::{TypedMultiDenseVectorRef, VectorElementType};
 use crate::types::{MultiVectorComparator, MultiVectorConfig};

lib/storage/Cargo.toml

@@ -19,7 +19,7 @@ proptest = "1.5.0"
 env_logger = "0.11"
 
 [dependencies]
-thiserror = "1.0"
+thiserror = { workspace = true }
 rand = { workspace = true }
 wal = { workspace = true }
 tokio = { workspace = true }