diff --git a/Cargo.lock b/Cargo.lock index 549c4860d4..3bedaaa638 100644 --- a/Cargo.lock +++ b/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]] diff --git a/Cargo.toml b/Cargo.toml index d21e218b12..2f8c8acfdb 100644 --- a/Cargo.toml +++ b/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" diff --git a/lib/api/Cargo.toml b/lib/api/Cargo.toml index d95fb21aee..3b667b0b05 100644 --- a/lib/api/Cargo.toml +++ b/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 } diff --git a/lib/collection/Cargo.toml b/lib/collection/Cargo.toml index 8c7f917763..d265d6ad4e 100644 --- a/lib/collection/Cargo.toml +++ b/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" } diff --git a/lib/common/cancel/Cargo.toml b/lib/common/cancel/Cargo.toml index b433fd189a..b5a97ac683 100644 --- a/lib/common/cancel/Cargo.toml +++ b/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 } diff --git a/lib/common/common/Cargo.toml b/lib/common/common/Cargo.toml index 30fb2d416e..1f23b84b8b 100644 --- a/lib/common/common/Cargo.toml +++ b/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]] diff --git a/lib/common/io/Cargo.toml b/lib/common/io/Cargo.toml index 590d3f9e61..6179c2b2ff 100644 --- a/lib/common/io/Cargo.toml +++ b/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 } diff --git a/lib/common/memory/Cargo.toml b/lib/common/memory/Cargo.toml index 6d1953c26b..db1dc0cec0 100644 --- a/lib/common/memory/Cargo.toml +++ b/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 } \ No newline at end of file diff --git a/lib/common/memory/src/lib.rs b/lib/common/memory/src/lib.rs index b98260e59e..6b5f8dee2b 100644 --- a/lib/common/memory/src/lib.rs +++ b/lib/common/memory/src/lib.rs @@ -1,2 +1,3 @@ pub mod madvise; pub mod mmap_ops; +pub mod mmap_type; diff --git a/lib/segment/src/common/mmap_type.rs b/lib/common/memory/src/mmap_type.rs similarity index 98% rename from lib/segment/src/common/mmap_type.rs rename to lib/common/memory/src/mmap_type.rs index 4ba9793cc6..f010d6defa 100644 --- a/lib/segment/src/common/mmap_type.rs +++ b/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 = std::result::Result; +pub type MmapFlusher = Box 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 MmapSlice { } /// 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() diff --git a/lib/segment/Cargo.toml b/lib/segment/Cargo.toml index f2b8af1809..00a966509f 100644 --- a/lib/segment/Cargo.toml +++ b/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 } diff --git a/lib/segment/src/common/mmap_bitslice_buffered_update_wrapper.rs b/lib/segment/src/common/mmap_bitslice_buffered_update_wrapper.rs index 6913c9fbdc..5b6e04a5f8 100644 --- a/lib/segment/src/common/mmap_bitslice_buffered_update_wrapper.rs +++ b/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()()?) }) } } diff --git a/lib/segment/src/common/mmap_slice_buffered_update_wrapper.rs b/lib/segment/src/common/mmap_slice_buffered_update_wrapper.rs index d50cadefd4..a53de0693e 100644 --- a/lib/segment/src/common/mmap_slice_buffered_update_wrapper.rs +++ b/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()()?) }) } } diff --git a/lib/segment/src/common/mod.rs b/lib/segment/src/common/mod.rs index 6523d1bf24..dc02cdd626 100644 --- a/lib/segment/src/common/mod.rs +++ b/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; diff --git a/lib/segment/src/common/operation_error.rs b/lib/segment/src/common/operation_error.rs index 3bb0c21e88..98a0b3990b 100644 --- a/lib/segment/src/common/operation_error.rs +++ b/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; diff --git a/lib/segment/src/id_tracker/immutable_id_tracker.rs b/lib/segment/src/id_tracker/immutable_id_tracker.rs index 6b045d40bd..c5fd650904 100644 --- a/lib/segment/src/id_tracker/immutable_id_tracker.rs +++ b/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; diff --git a/lib/segment/src/index/field_index/map_index/mmap_map_index.rs b/lib/segment/src/index/field_index/map_index/mmap_map_index.rs index 912084e4a0..de4a0f0813 100644 --- a/lib/segment/src/index/field_index/map_index/mmap_map_index.rs +++ b/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; diff --git a/lib/segment/src/index/field_index/numeric_index/mmap_numeric_index.rs b/lib/segment/src/index/field_index/numeric_index/mmap_numeric_index.rs index ca91893ac4..0a9a286b1a 100644 --- a/lib/segment/src/index/field_index/numeric_index/mmap_numeric_index.rs +++ b/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}; diff --git a/lib/segment/src/vector_storage/chunked_mmap_vectors.rs b/lib/segment/src/vector_storage/chunked_mmap_vectors.rs index 44db72dd23..b2854acb7c 100644 --- a/lib/segment/src/vector_storage/chunked_mmap_vectors.rs +++ b/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}; diff --git a/lib/segment/src/vector_storage/chunked_utils.rs b/lib/segment/src/vector_storage/chunked_utils.rs index 63fe48e651..2adff4f08b 100644 --- a/lib/segment/src/vector_storage/chunked_utils.rs +++ b/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_"; diff --git a/lib/segment/src/vector_storage/dense/dynamic_mmap_flags.rs b/lib/segment/src/vector_storage/dense/dynamic_mmap_flags.rs index 17e57121f7..00c281846a 100644 --- a/lib/segment/src/vector_storage/dense/dynamic_mmap_flags.rs +++ b/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>>, + flags_flusher: Arc>>, status: MmapType, 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)?; diff --git a/lib/segment/src/vector_storage/dense/memmap_dense_vector_storage.rs b/lib/segment/src/vector_storage/dense/memmap_dense_vector_storage.rs index 24a984bdcd..63aeacf60d 100644 --- a/lib/segment/src/vector_storage/dense/memmap_dense_vector_storage.rs +++ b/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 VectorStorage for MemmapDenseVectorStorage { 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(())), } } diff --git a/lib/segment/src/vector_storage/dense/mmap_dense_vectors.rs b/lib/segment/src/vector_storage/dense/mmap_dense_vectors.rs index 7cee0a1683..e7dfaa8702 100644 --- a/lib/segment/src/vector_storage/dense/mmap_dense_vectors.rs +++ b/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 MmapDenseVectors { self.uring_reader.lock().is_some() } - pub fn flusher(&self) -> Flusher { + pub fn flusher(&self) -> MmapFlusher { self.deleted.flusher() } diff --git a/lib/segment/src/vector_storage/quantized/quantized_multivector_storage.rs b/lib/segment/src/vector_storage/quantized/quantized_multivector_storage.rs index e736d5f117..13f6cf6176 100644 --- a/lib/segment/src/vector_storage/quantized/quantized_multivector_storage.rs +++ b/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}; diff --git a/lib/storage/Cargo.toml b/lib/storage/Cargo.toml index b4942aa1a6..70337e9f50 100644 --- a/lib/storage/Cargo.toml +++ b/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 }