refactor(storage): optimize concurrent metadata sync and remove lock contention

Simplify binary metadata synchronization by:
- Remove retry mechanism in sync_to_disk for cleaner error handling
- Minimize lock scope by cloning store before I/O operations
- Add Clone derive to BinaryMetadataStore for easier snapshotting
- Remove implicit sync in constructor to avoid I/O contention with multiple instances

Also bump version to 0.5.2 and update documentation with expanded API reference, concurrent access examples, and performance tips.

Author: birchkwok

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "numpack"
-version = "0.5.1"
+version = "0.5.2"
 edition = "2021"
 description = "A high-performance array storage and manipulation library"
 authors = ["NumPack Contributors"]
@@ -32,7 +32,7 @@ tempfile = "3.8.1"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 rayon = { version = "1.8", optional = true }
-memmap2 = "0.5.10"
+memmap2 = "0.5.20"
 half = { version = "2.6.0", features = ["bytemuck"] }
 bitvec = "1.0.1"
 bincode = "1.3"

README.md

@@ -36,11 +36,12 @@ Add to your `Cargo.toml`:
 ```toml
 [dependencies]
 numpack = "0.5.1"
+ndarray = "0.16"
 ```
 
 **Features:**
 - `rayon` (default) - Parallel processing support
-- `avx512` - AVX-512 SIMD optimizations
+- `avx512` - AVX-512 SIMD optimizations  
 - `io-uring-support` - io_uring on Linux
 
 **Requirements:** Rust ≥ 1.70.0
@@ -57,69 +58,161 @@ maturin develop  # or: maturin build --release
 ```
 </details>
 
-## Quick Start
+#### Basic Usage
 
-### Python
-
-```python
-import numpy as np
-from numpack import NumPack
+```rust
+use numpack::prelude::*;
+use ndarray::{ArrayD, Array2, array};
+use std::path::Path;
 
-with NumPack("data.npk") as npk:
-    # Save
-    npk.save({'embeddings': np.random.rand(10000, 128).astype(np.float32)})
+fn main() -> NpkResult<()> {
+    // Create or open a NumPack storage
+    let io = ParallelIO::new(PathBuf::from("data.npk"))?;
 
-    # Load (normal or lazy)
-    data = npk.load("embeddings")
-    lazy = npk.load("embeddings", lazy=True)
+    // Save arrays with explicit dtype
+    let data: Array2<f32> = Array2::from_shape_fn((1000, 128), |_| rand::random());
+    io.save_arrays(&[("embeddings".to_string(), data.into_dyn(), DataType::Float32)])?;
 
-    # Modify
-    npk.replace({'embeddings': new_rows}, indices=[0, 1, 2])
-    npk.append({'embeddings': more_rows})
-    npk.drop('embeddings', [0, 1, 2])  # drop rows
+    // Metadata is written on drop, or call sync_metadata() explicitly
+    io.sync_metadata()?;
 
-    # Random access
-    subset = npk.getitem('embeddings', [100, 200, 300])
+    Ok(())
+}
 ```
 
-### Rust
+#### API Reference
+
+**Storage Operations:**
+
+| Method | Description | Example |
+|--------|-------------|---------|
+| `ParallelIO::new(path)` | Create/open storage | `ParallelIO::new(PathBuf::from("data"))?` |
+| `save_arrays(&[(name, array, dtype)])` | Save multiple arrays | See below |
+| `sync_metadata()` | Persist metadata to disk | `io.sync_metadata()?` |
+| `reset()` | Delete all arrays | `io.reset()?` |
+
+**Array Operations:**
 
 ```rust
-use numpack::{ParallelIO, DataType};
-use ndarray::{ArrayD, Array2};
-use std::path::Path;
+// Check if array exists
+if io.has_array("embeddings") {
+    println!("Array exists!");
+}
 
-fn main() -> Result<(), Box<dyn std::error::Error>> {
-    // Create or open a NumPack storage
-    let npk = ParallelIO::new(Path::new("data.npk"));
-    
-    // Save arrays
-    let data: Array2<f32> = Array2::from_shape_fn((1000, 128), |_| rand::random());
-    npk.save_arrays(&[("embeddings", data.view().into_dyn())])?;
-    
-    // Load arrays
-    let loaded: ArrayD<f32> = npk.load_array("embeddings")?;
+// List all arrays
+let names = io.list_arrays();
+println!("Arrays: {:?}", names);
+
+// Get array metadata
+let meta = io.get_array_metadata("embeddings")?;
+println!("Shape: {:?}, dtype: {:?}", meta.shape, meta.get_dtype());
+
+// Read specific rows (returns raw bytes)
+let row_data = io.read_rows("embeddings", &[0, 10, 20])?;
+
+// Replace rows in-place (fastest for existing arrays)
+let new_rows = Array2::<f32>::zeros((3, 128)).into_dyn();
+io.replace_rows("embeddings", &new_rows, &[0, 1, 2])?;
+
+// Logical delete (mark rows as deleted)
+io.drop_arrays("embeddings", Some(&[5, 6, 7]))?;
+
+// Physical compact (remove deleted rows, reclaim space)
+io.compact_array("embeddings")?;
+
+// Delete entire array
+io.drop_arrays("embeddings", None)?;
+```
+
+**Data Type Mapping:**
+
+| NumPack Type | Rust Type | Size |
+|--------------|-----------|------|
+| `DataType::Bool` | `bool` | 1 byte |
+| `DataType::Int8` | `i8` | 1 byte |
+| `DataType::Int16` | `i16` | 2 bytes |
+| `DataType::Int32` | `i32` | 4 bytes |
+| `DataType::Int64` | `i64` | 8 bytes |
+| `DataType::Uint8` | `u8` | 1 byte |
+| `DataType::Uint16` | `u16` | 2 bytes |
+| `DataType::Uint32` | `u32` | 4 bytes |
+| `DataType::Uint64` | `u64` | 8 bytes |
+| `DataType::Float16` | `half::f16` | 2 bytes |
+| `DataType::Float32` | `f32` | 4 bytes |
+| `DataType::Float64` | `f64` | 8 bytes |
+| `DataType::Complex64` | `num_complex::Complex32` | 8 bytes |
+| `DataType::Complex128` | `num_complex::Complex64` | 16 bytes |
+
+#### Concurrent Access
+
+Multiple threads can safely write to the same storage concurrently (since v0.5.1+):
+
+```rust
+use std::thread;
+
+fn concurrent_write() -> NpkResult<()> {
+    let dir = "/tmp/numpack_data";
+    std::fs::create_dir_all(dir)?;
 
-    // Random access - get specific rows
-    let rows = npk.get_rows("embeddings", &[0, 10, 20])?;
+    let handles: Vec<_> = (0..10)
+        .map(|i| {
+            let dir = dir.to_string();
+            thread::spawn(move || {
+                let io = ParallelIO::new(PathBuf::from(dir))?;
+                let data = Array2::<f32>::ones((100, 128)).into_dyn();
+                io.save_arrays(&[(format!("chunk_{}", i), data, DataType::Float32)])?;
+                io.sync_metadata()?;  // Ensure metadata is written
+                println!("Thread {} done", i);
+                Ok::<_, NpkError>(())
+            })
+        })
+        .collect();
 
-    // Get metadata
-    let meta = npk.get_metadata("embeddings")?;
-    println!("Shape: {:?}, dtype: {:?}", meta.shape, meta.dtype);
+    for h in handles {
+        h.join().unwrap()?;
+    }
 
     Ok(())
 }
 ```
 
-**Core Types:**
-- `ParallelIO` - Main interface for storage operations
-- `DataType` - Supported data types (Bool, Int8-64, Uint8-64, Float16/32/64, Complex64/128)
-- `ArrayMetadata` - Array shape, dtype, and file info
+**Best Practices for Concurrent Access:**
+- Each thread creates its own `ParallelIO` instance
+- Call `sync_metadata()` before dropping the instance
+- For read-heavy workloads, use separate read instances
 
-**Features:**
-- Zero-copy memory mapping via `memmap2`
-- Parallel IO with `rayon`
-- SIMD vector operations
+#### Performance Tips
+
+```rust
+// 1. Batch saves for multiple arrays
+let arrays: Vec<(String, ArrayD<f32>, DataType)> = vec![
+    ("a".to_string(), data_a, DataType::Float32),
+    ("b".to_string(), data_b, DataType::Float32),
+];
+io.save_arrays(&arrays)?;  // Parallel processing for large data
+
+// 2. Use replace_rows for updating existing arrays (fastest)
+// Avoids file recreation when shape and dtype match
+
+// 3. Call sync_metadata() once after all operations
+// Not needed after every save_arrays()
+
+// 4. Use compact_array() periodically after many deletions
+io.drop_arrays("data", Some(&[0, 1, 2]))?;  // Logical delete
+io.compact_array("data")?;  // Physical cleanup when convenient
+```
+
+#### Error Handling
+
+```rust
+use numpack::core::error::{NpkError, NpkResult};
+
+match io.get_array_metadata("nonexistent") {
+    Ok(meta) => println!("Found: {:?}", meta.shape),
+    Err(NpkError::ArrayNotFound(name)) => println!("Array {} not found", name),
+    Err(e) => eprintln!("Error: {:?}", e),
+}
+```
 
 ### Batch Modes
 

docs/api_reference/README.md

@@ -83,4 +83,4 @@ from numpack import get_backend_info
 
 ## Version
 
-This documentation is for NumPack version **0.5.1**.
+This documentation is for NumPack version **0.5.2**.

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "maturin"
 
 [project]
 name = "numpack"
-version = "0.5.1"
+version = "0.5.2"
 description = "A high-performance array storage and manipulation library"
 authors = [{ name = "NumPack Contributors" }]
 requires-python = ">=3.9"

python/numpack/__init__.py

@@ -39,7 +39,7 @@ def _numpack_issubdtype(arg1, arg2):
 
         np.issubdtype = _numpack_issubdtype  # type: ignore[assignment]
 
-__version__ = "0.5.1"
+__version__ = "0.5.2"
 
 # Platform detection
 def _is_windows():

src/storage/binary_metadata.rs

@@ -396,7 +396,7 @@ impl BinaryArrayMetadata {
 }
 
 /// 二进制格式的元数据存储
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct BinaryMetadataStore {
     pub version: u32,
     pub arrays: HashMap<String, BinaryArrayMetadata>,
@@ -574,8 +574,8 @@ impl BinaryCachedStore {
             sync_interval: std::time::Duration::from_secs(1),
         };
 
-        // 保存初始存储
-        cached_store.sync_to_disk()?;
+        // 不在构造函数中调用 sync_to_disk，避免多实例并发创建时的 I/O 争用
+        // 元数据会在显式调用 force_sync() 时写入磁盘
         Ok(cached_store)
     }
 
@@ -593,34 +593,14 @@ impl BinaryCachedStore {
     }
 
     fn sync_to_disk(&self) -> NpkResult<()> {
-        // 多线程环境下可能出现临时的文件访问冲突，添加重试机制
-        const MAX_RETRIES: usize = 3;
-        const RETRY_DELAY_MS: u64 = 10;
-
-        let mut last_error = None;
-        for attempt in 0..MAX_RETRIES {
-            let store = self.store.read().unwrap();
-            match store.save(&self.path) {
-                Ok(_) => {
-                    drop(store);
-                    let mut last_sync = self.last_sync.lock().unwrap();
-                    *last_sync = SystemTime::now();
-                    return Ok(());
-                }
-                Err(e) => {
-                    last_error = Some(e);
-                    drop(store);
-
-                    // 最后一次尝试不需要等待
-                    if attempt < MAX_RETRIES - 1 {
-                        std::thread::sleep(std::time::Duration::from_millis(RETRY_DELAY_MS));
-                    }
-                }
-            }
-        }
+        // 最小化锁范围：仅在 clone 期间持有读锁，I/O 操作不持有任何锁
+        let snapshot = self.store.read().unwrap().clone();
+        // 读锁已在上一行末尾自动释放
 
-        // 所有重试都失败，返回最后一个错误
-        Err(last_error.unwrap())
+        snapshot.save(&self.path)?;
+
+        *self.last_sync.lock().unwrap() = SystemTime::now();
+        Ok(())
     }
 
     pub fn add_array(&self, meta: BinaryArrayMetadata) -> NpkResult<()> {