From dbbcba0bbbbea1ce7228572b7b24152fb5b79a39 Mon Sep 17 00:00:00 2001 From: Retype15 Date: Fri, 26 Dec 2025 03:35:45 -0500 Subject: [PATCH 1/3] feat: Implement an slice on reader to move chunks without memcpy. --- TODO.md | 2 +- benches/lazy_bench.rs | 4 - benches/map_access.rs | 10 +- benches/performance.rs | 110 ++++---------- src/reader.rs | 326 ++++++++++++++++++++++++----------------- tests/miri_test.rs | 66 +++++++++ 6 files changed, 292 insertions(+), 226 deletions(-) create mode 100644 tests/miri_test.rs diff --git a/TODO.md b/TODO.md index 1ebafb5..e2a9450 100644 --- a/TODO.md +++ b/TODO.md @@ -2,7 +2,7 @@ This document tracks pending features, architectural optimizations, and API improvements planned for future versions of Parcode. -## 🟢 High Priority (Immediate - v0.4.x) +## 🟢 High Priority (Immediate - v0.5.x) ### 1. Serialization Backend Migration diff --git a/benches/lazy_bench.rs b/benches/lazy_bench.rs index 65fb709..9cd8fac 100644 --- a/benches/lazy_bench.rs +++ b/benches/lazy_bench.rs @@ -44,7 +44,6 @@ fn bench_lazy(c: &mut Criterion) { let mut group = c.benchmark_group("Lazy Access"); - // Caso A: Carga Completa (Estándar) group.bench_function("full_load", |b| { b.iter(|| { let loaded: Root = Parcode::load(&path).expect("Failed to read parcode data"); @@ -52,18 +51,15 @@ fn bench_lazy(c: &mut Criterion) { }); }); - // Caso B: Carga Lazy (Solo metadatos) group.bench_function("lazy_meta_only", |b| { b.iter(|| { let file_handle = Parcode::open(&path).expect("Failed to open file"); let lazy = file_handle.root::().expect("Failed to read lazy"); - // Accedemos a meta profundo A y B let sum = lazy.child_a.meta + lazy.child_b.meta; black_box(sum); }); }); - // Caso C: Carga Parcial (Meta A + Payload A) group.bench_function("lazy_partial_load", |b| { b.iter(|| { let file_handle = Parcode::open(&path).expect("Failed to open file"); diff --git a/benches/map_access.rs b/benches/map_access.rs index 78d44b4..5dab84c 100644 --- a/benches/map_access.rs +++ b/benches/map_access.rs @@ -1,6 +1,3 @@ -// benches/map_access.rs -//! PLACEHOLDER -//! #![allow(missing_docs)] use criterion::{Criterion, criterion_group, criterion_main}; use parcode::{Parcode, ParcodeObject}; @@ -10,13 +7,10 @@ use tempfile::NamedTempFile; #[derive(Serialize, Deserialize, ParcodeObject)] struct MapContainer { - /// PLACEHOLDER #[parcode(map)] // Optimized opt_map: HashMap, - // #[parcode(chunkable)] // Standard Blob (Unoptimized for random access) + // #[parcode(chunkable)] // std_map: HashMap, - // Nota: Para comparar justo, deberíamos usar otro campo o struct, - // ya que HashMap sin 'map' flag se guarda como blob. } fn bench_map(c: &mut Criterion) { @@ -34,14 +28,12 @@ fn bench_map(c: &mut Criterion) { let mut group = c.benchmark_group("Map Random Access"); group.bench_function("optimized_lookup", |b| { - // Setup reader once per batch to simulate persistent app let file_handle = Parcode::open(&path).expect("Failed to open file"); let lazy = file_handle .root::() .expect("Failed to read lazy"); b.iter(|| { - // Lookup key 50,000 (Middle) let val = lazy.opt_map.get(&50_000).expect("Failed to get value"); std::hint::black_box(val); }); diff --git a/benches/performance.rs b/benches/performance.rs index 3a3d451..9cd61ef 100644 --- a/benches/performance.rs +++ b/benches/performance.rs @@ -1,60 +1,24 @@ -// ===== benches\performance.rs ===== #![allow(missing_docs)] use criterion::{Criterion, Throughput, criterion_group, criterion_main}; -use parcode::{Parcode, ParcodeObject, graph::*, visitor::ParcodeVisitor}; +use parcode::{Parcode, ParcodeObject}; use serde::{Deserialize, Serialize}; use std::fs::File; use std::hint::black_box; use std::io::BufWriter; use tempfile::NamedTempFile; -// --- SETUP --- - -#[derive(Clone, Serialize, Deserialize, ParcodeObject)] +#[derive(Clone, Serialize, Deserialize, ParcodeObject, Debug)] struct BenchItem { id: u64, - payload: Vec, // 1KB payload -} - -// Wrapper to satisfy Parcode traits -#[derive(Clone, Serialize, Deserialize)] -struct BenchCollection(Vec); - -// Minimal Manual Implementation for Benchmarking -impl ParcodeVisitor for BenchCollection { - fn visit<'a>( - &'a self, - graph: &mut TaskGraph<'a>, - parent_id: Option, - config_override: Option, - ) { - // Delegate to the internal Vec, propagating the configuration - self.0.visit(graph, parent_id, config_override); - } - - fn create_job<'a>( - &'a self, - config_override: Option, - ) -> Box + 'a> { - let base = Box::new(ContainerJob); - if let Some(cfg) = config_override { - Box::new(parcode::rt::ConfiguredJob::new(base, cfg)) - } else { - base - } - } + #[parcode(chunkable)] + payload: Vec, } -#[derive(Clone)] -struct ContainerJob; -impl SerializationJob<'_> for ContainerJob { - fn execute(&self, _: &[parcode::format::ChildRef]) -> parcode::Result> { - Ok(vec![]) - } - fn estimated_size(&self) -> usize { - 0 - } +#[derive(Clone, Serialize, Deserialize, ParcodeObject, Debug)] +struct BenchCollection { + #[parcode(chunkable)] + data: Vec, } fn generate_data(count: usize) -> BenchCollection { @@ -64,15 +28,15 @@ fn generate_data(count: usize) -> BenchCollection { payload: vec![i as u64; 128], // ~1KB }) .collect(); - BenchCollection(items) + BenchCollection { data: items } } // --- BENCHMARKS --- fn bench_writers(c: &mut Criterion) { - let item_count = 200_000; + let item_count = 100_000; let data = generate_data(item_count); - let raw_data = &data.0; // For bincode + let raw_data = &data.data; // For bincode println!("Writers Item count: {}", item_count); @@ -93,7 +57,7 @@ fn bench_writers(c: &mut Criterion) { }); }); - // 2. Parcode (Parallel Graph Engine) + // 2. Parcode group.bench_function("parcode_save", |b| { b.iter(|| { let file = NamedTempFile::new().expect("Failed to create temp file"); @@ -105,7 +69,7 @@ fn bench_writers(c: &mut Criterion) { } fn bench_readers(c: &mut Criterion) { - let item_count = 200_000; + let item_count = 100_000; println!("Readers Item count: {}", item_count); @@ -114,7 +78,7 @@ fn bench_readers(c: &mut Criterion) { // Setup files let bincode_file = NamedTempFile::new().expect("Failed to create temp file"); bincode::serde::encode_into_std_write( - &data.0, + &data.data, &mut BufWriter::new(&bincode_file), bincode::config::standard(), ) @@ -150,44 +114,34 @@ fn bench_readers(c: &mut Criterion) { group.bench_function("parcode_random_access_10", |b| { b.iter(|| { let file_handle = Parcode::open(&parcode_path).expect("Failed to open file"); - let root = file_handle.root_node().expect("Failed to get root"); + let root = file_handle + .root::() + .expect("Failed to get root"); - for i in (0..10).map(|x| x * (item_count / 10)) { - let _obj: BenchItem = root.get(i).expect("Failed to get item"); + for i in (0..10).map(|x| x * (item_count / 20)) { + let _obj: BenchItem = root.data.get(i).expect("Failed to get item"); } }); }); - // 3. Parcode: Full Scan (Manual Shard Iteration) - group.bench_function("parcode_full_scan_manual", |b| { + // 3. Parcode: Parallel Stitching + group.bench_function("parcode_read_all_parallel", |b| { b.iter(|| { - let file_handle = Parcode::open(&parcode_path).expect("Failed to open file"); - let root = file_handle.root_node().expect("Failed to get root"); - - // Get the Shards (direct children) - let shards = root.children().expect("Failed to get children"); - - for shard_node in shards { - // Deserialize the complete Shard (Vec) - let items: Vec = shard_node.decode().expect("Failed to decode shard"); - - // Iterate items in memory (simulating usage) - for item in items { - black_box(item); - } - } + let _res: BenchCollection = Parcode::load(&parcode_path).expect("Failed to open file"); }); }); - // 4. Parcode: Parallel Stitching (La nueva joya) - // Añadimos esto para probar la velocidad de reconstrucción total - group.bench_function("parcode_read_all_parallel", |b| { + // 4. Parcode: lazy iterator + group.bench_function("parcode_read_all_iter", |b| { b.iter(|| { - let file_handle = Parcode::open(&parcode_path).expect("Failed to open file"); - let root = file_handle.root_node().expect("Failed to get root"); - let _res: Vec = root - .decode_parallel_collection() - .expect("Failed to decode parallel"); + let file = Parcode::open(&parcode_path).expect("Failed to open file"); + let data = file + .load_lazy::() + .expect("Some error was ocurred"); + for item in data.data.iter_lazy().expect("Some error was ocurred") { + let i = item.expect("Some error was ocurred"); + black_box(i); + } }); }); diff --git a/src/reader.rs b/src/reader.rs index ce14ff8..9738eac 100644 --- a/src/reader.rs +++ b/src/reader.rs @@ -212,6 +212,8 @@ use std::io::Cursor; use std::marker::PhantomData; use std::mem::{ManuallyDrop, MaybeUninit}; use std::path::Path; +use std::ptr; +use std::slice; use std::sync::Arc; use crate::compression::CompressorRegistry; @@ -322,47 +324,81 @@ pub trait ParcodeItem: Sized + Send + Sync + 'static { children: &mut std::vec::IntoIter>, ) -> Result; - /// Reads a slice of items from the shard payload and children. - /// - /// This method provides an optimization opportunity for types that can deserialize - /// multiple items more efficiently than calling [`read_from_shard`](Self::read_from_shard) - /// in a loop. - /// - /// ## Default Implementation - /// - /// The default implementation reads the slice length (u64) and then calls - /// `read_from_shard` for each item. Primitive types override this to use bulk - /// deserialization. - /// - /// ## Parameters - /// - /// * `reader`: Cursor over the shard's decompressed payload - /// * `children`: Iterator over child nodes (for chunkable fields) - /// - /// ## Returns - /// - /// A vector containing all deserialized items. - /// - /// ## Errors - /// - /// Returns an error if deserialization fails or the slice length exceeds `usize`. + /// DEPRECATED in favor of read_into_slice internally, but kept for API compat. + /// Default impl now delegates to read_into_slice to reduce code duplication. fn read_slice_from_shard( reader: &mut std::io::Cursor<&[u8]>, children: &mut std::vec::IntoIter>, ) -> Result> { - // Default implementation: Read length, then loop - let len = - bincode::serde::decode_from_std_read::(reader, bincode::config::standard()) - .map_err(|e| ParcodeError::Serialization(e.to_string()))?; + let start_pos = reader.position(); - let mut vec = Vec::with_capacity( - usize::try_from(len) - .map_err(|_| ParcodeError::Serialization("Vector length exceeds usize".into()))?, - ); - for _ in 0..len { - vec.push(Self::read_from_shard(reader, children)?); + // Read length prefix + let len: u64 = bincode::serde::decode_from_std_read(reader, bincode::config::standard()) + .map_err(|e| ParcodeError::Serialization(e.to_string()))?; + + let count = usize::try_from(len) + .map_err(|_| ParcodeError::Serialization("Vector length exceeds usize".into()))?; + + // Allocate uninit vec + let mut uninit_vec: Vec> = Vec::with_capacity(count); + #[allow(unsafe_code)] + unsafe { + uninit_vec.set_len(count); + } + + reader.set_position(start_pos); + + let read_count = Self::read_into_slice(reader, children, &mut uninit_vec)?; + + if read_count != count { + return Err(ParcodeError::Format("Mismatch in read count".into())); + } + + // Transmute to Vec + #[allow(unsafe_code)] + let final_vec = unsafe { + let mut manual = ManuallyDrop::new(uninit_vec); + Vec::from_raw_parts(manual.as_mut_ptr() as *mut Self, count, count) + }; + Ok(final_vec) + } + + /// Reads items directly into a destination slice of uninitialized memory. + /// + /// # Performance + /// This method avoids allocating a temporary `Vec`, writing directly to the final buffer. + /// + /// # Safety + /// The implementation must ensure it does not write past the end of `destination`. + /// It returns the number of items actually written. + fn read_into_slice( + reader: &mut std::io::Cursor<&[u8]>, + children: &mut std::vec::IntoIter>, + destination: &mut [MaybeUninit], + ) -> Result { + // 1. Read Length + let len: u64 = bincode::serde::decode_from_std_read(reader, bincode::config::standard()) + .map_err(|e| ParcodeError::Serialization(e.to_string()))?; + + let count = usize::try_from(len) + .map_err(|_| ParcodeError::Serialization("Shard length exceeds usize".into()))?; + + // 2. Bounds Check + if destination.len() < count { + return Err(ParcodeError::Format(format!( + "Destination buffer too small: expected {}, got {}", + count, + destination.len() + ))); + } + + // 3. Loop and Initialize + for i in 0..count { + let item = Self::read_from_shard(reader, children)?; + destination[i].write(item); } - Ok(vec) + + Ok(count) } } @@ -378,13 +414,47 @@ macro_rules! impl_primitive_parcode_item { .map_err(|e| ParcodeError::Serialization(e.to_string())) } - // Optimize slice reading for primitives (bulk read) - fn read_slice_from_shard( + fn read_into_slice( reader: &mut std::io::Cursor<&[u8]>, _children: &mut std::vec::IntoIter>, - ) -> Result> { - bincode::serde::decode_from_std_read(reader, bincode::config::standard()) - .map_err(|e| ParcodeError::Serialization(e.to_string())) + destination: &mut [MaybeUninit], + ) -> Result { + // Read Length + let len: u64 = bincode::serde::decode_from_std_read(reader, bincode::config::standard()) + .map_err(|e| ParcodeError::Serialization(e.to_string()))?; + let count = usize::try_from(len).map_err(|_| ParcodeError::Format("Len overflow".into()))?; + + if destination.len() < count { + return Err(ParcodeError::Format("Buffer too small".into())); + } + + // If T is u8, we can memcpy directly from the reader slice. + if std::any::TypeId::of::() == std::any::TypeId::of::() { + let pos = reader.position() as usize; + let inner = reader.get_ref(); + if inner.len() < pos + count { + return Err(ParcodeError::Format("Unexpected EOF reading u8 blob".into())); + } + + let src_slice = &inner[pos..pos+count]; + + let dest_ptr = destination.as_mut_ptr() as *mut u8; + #[allow(unsafe_code)] + unsafe { + ptr::copy_nonoverlapping(src_slice.as_ptr(), dest_ptr, count); + } + + reader.set_position((pos + count) as u64); + return Ok(count); + } + + // Default Primitive Loop + for i in 0..count { + let item: Self = bincode::serde::decode_from_std_read(reader, bincode::config::standard()) + .map_err(|e| ParcodeError::Serialization(e.to_string()))?; + destination[i].write(item); + } + Ok(count) } } )* @@ -392,7 +462,7 @@ macro_rules! impl_primitive_parcode_item { } impl_primitive_parcode_item!( - u8, u16, u32, u64, u128, i8, i16, i32, i64, i128, f32, f64, bool, String + u8, u16, u32, u64, u128, usize, i8, i16, i32, i64, i128, isize, f32, f64, bool, String ); macro_rules! impl_primitive_parcode_native { @@ -408,10 +478,10 @@ macro_rules! impl_primitive_parcode_native { } impl_primitive_parcode_native!( - u8, u16, u32, u64, u128, i8, i16, i32, i64, i128, f32, f64, bool, String + u8, u16, u32, u64, u128, usize, i8, i16, i32, i64, i128, isize, f32, f64, bool, String ); -/// Optimized implementation for Vectors: Uses Parallel Stitching. +/// Uses Parallel Stitching. impl ParcodeNative for Vec where T: ParcodeItem, @@ -427,18 +497,16 @@ where V: DeserializeOwned + Send + Sync, { fn from_node(node: &ChunkNode<'_>) -> Result { - // 1. Read container (num shards) + // Standard HashMap reconstruction logic (same as before) let container_payload = node.read_raw()?; if container_payload.len() < 4 { return Ok(Self::new()); } - // If it's a Blob, node.child_count == 0. if node.child_count == 0 { - return node.decode(); // Fallback to normal Bincode + return node.decode(); } - // If it has children, it's a Sharded Map. let shards = node.children()?; let mut map = Self::new(); @@ -448,31 +516,15 @@ where continue; } - let count = u32::from_le_bytes( - payload - .get(0..4) - .ok_or_else(|| ParcodeError::Format("Payload too short for count".into()))? - .try_into() - .map_err(|_| ParcodeError::Format("Failed to read count".into()))?, - ) as usize; + let count = u32::from_le_bytes(payload.get(0..4).unwrap().try_into().unwrap()) as usize; let offsets_start = 8 + (count * 8); let data_start = offsets_start + (count * 4); - let offsets_bytes = payload - .get(offsets_start..data_start) - .ok_or_else(|| ParcodeError::Format("Offsets out of bounds".into()))?; + let offsets_bytes = payload.get(offsets_start..data_start).unwrap(); for i in 0..count { - let off_bytes = offsets_bytes - .get(i * 4..(i + 1) * 4) - .ok_or_else(|| ParcodeError::Format("Offset index out of bounds".into()))?; - let offset = u32::from_le_bytes( - off_bytes - .try_into() - .map_err(|_| ParcodeError::Format("Failed to read offset".into()))?, - ) as usize; - let data_slice = payload - .get(data_start + offset..) - .ok_or_else(|| ParcodeError::Format("Data slice out of bounds".into()))?; + let off_bytes = offsets_bytes.get(i * 4..(i + 1) * 4).unwrap(); + let offset = u32::from_le_bytes(off_bytes.try_into().unwrap()) as usize; + let data_slice = payload.get(data_start + offset..).unwrap(); let (k, v) = bincode::serde::decode_from_slice(data_slice, bincode::config::standard()) .map_err(|e| ParcodeError::Serialization(e.to_string()))? @@ -484,8 +536,6 @@ where } } -// --- CORE READER HANDLE --- - /// Represents an open Parcode file mapped in memory. /// /// This handle allows: @@ -520,7 +570,7 @@ impl ParcodeFile { )); } - // SAFETY: Mmap is fundamentally unsafe in the presence of external modification + // Mmap is fundamentally unsafe in the presence of external modification // (e.g., another process truncating the file). We assume the file is treated // as immutable by the OS while we read it. #[allow(unsafe_code)] @@ -792,45 +842,50 @@ impl<'a> ChunkNode<'a> { /// **Parallel Shard Reconstruction** /// - /// Reconstructs a `Vec` by deserializing shards concurrently. + /// Reconstructs a `Vec` by deserializing shards concurrently directly into the final buffer. /// - /// # Safety - /// This function uses `unsafe` to write directly into an uninitialized buffer. - /// To ensure safety: - /// 1. We pre-calculate correct offsets using RLE metadata. - /// 2. We cast the buffer pointer to `usize` to safely pass it to Rayon threads. - /// 3. We verify that the number of items read from a shard matches - /// the RLE expectation (`expected_count`). If not, we abort before writing, - /// preventing partial initialization UB. + /// # Optimizations + /// - **Destination Passing Style:** Shards write directly to `&mut [MaybeUninit]`. + /// - **Zero Alloc:** No temporary `Vec` created per thread. + /// - **Memory Efficient:** Peak memory usage is reduced by ~50% during load. + #[allow(unsafe_code)] pub fn decode_parallel_collection(&self) -> Result> where T: ParcodeItem, { let payload = self.read_raw()?; + if payload.is_empty() { + return Ok(Vec::new()); + } + + // Fallback for small/inline payloads if payload.len() < 8 { let mut cursor = std::io::Cursor::new(payload.as_ref()); let children = self.children()?; let mut child_iter = children.into_iter(); + // Use legacy read_slice helper (which internally creates a Vec, acceptable for small inline data) return T::read_slice_from_shard(&mut cursor, &mut child_iter); } + // 1. Read Total Items let total_items = usize::try_from(u64::from_le_bytes( payload .get(0..8) - .ok_or_else(|| ParcodeError::Format("Payload too short for header".into()))? + .ok_or_else(|| ParcodeError::Format("Payload too short".into()))? .try_into() - .map_err(|_| ParcodeError::Format("Failed to read total_items".into()))?, + .unwrap(), )) - .map_err(|_| ParcodeError::Format("total_items exceeds usize range".into()))?; + .map_err(|_| ParcodeError::Format("total_items exceeds usize".into()))?; + // 2. Parse RLE Metadata let runs_data = payload.get(8..).unwrap_or(&[]); let shard_runs: Vec = bincode::serde::decode_from_slice(runs_data, bincode::config::standard()) .map(|(obj, _)| obj) .map_err(|e| ParcodeError::Serialization(e.to_string()))?; - // 2. Expand RLE into explicit shard jobs. + // 3. Expand RLE to Jobs let mut shard_jobs = Vec::with_capacity(self.child_count as usize); let mut current_shard_idx = 0; let mut current_global_idx: usize = 0; @@ -839,9 +894,7 @@ impl<'a> ChunkNode<'a> { let items_per_shard = run.item_count as usize; for _ in 0..run.repeat { if current_global_idx.checked_add(items_per_shard).is_none() { - return Err(ParcodeError::Format( - "Integer overflow in RLE calculation".into(), - )); + return Err(ParcodeError::Format("Integer overflow in RLE".into())); } shard_jobs.push((current_shard_idx, current_global_idx, items_per_shard)); current_shard_idx += 1; @@ -851,7 +904,7 @@ impl<'a> ChunkNode<'a> { if current_global_idx != total_items { return Err(ParcodeError::Format(format!( - "Metadata mismatch: Header says {} items, RLE implies {}", + "Metadata mismatch: Header={}, RLE={}", total_items, current_global_idx ))); } @@ -860,73 +913,78 @@ impl<'a> ChunkNode<'a> { return Ok(Vec::new()); } - // 3. Allocate Uninit Buffer + // 4. Allocate Uninitialized Buffer let mut result_buffer: Vec> = Vec::with_capacity(total_items); - - #[allow(unsafe_code)] + // SAFETY: We strictly control initialization via the parallel loop below. unsafe { result_buffer.set_len(total_items); } - // 4. Prepare Pointer as usize to safely pass to Rayon threads (Send + Sync) - let buffer_start_addr = result_buffer.as_mut_ptr().addr(); + // 5. Parallel Execution (Destination Passing) + // We cast the buffer to a raw usize address to allow sending it to threads. + // `MaybeUninit` is not implicitly Sync, but since each thread writes to disjoint regions, + // it is sound. + let buffer_base_addr = result_buffer.as_mut_ptr() as usize; - shard_jobs.into_par_iter().try_for_each( + // Note: try_for_each stops on the first error. + let exec_result = shard_jobs.into_par_iter().try_for_each( move |(shard_idx, start_idx, expected_count)| -> Result<()> { + // A. Load Chunk let shard_node = self.get_child_by_index(shard_idx)?; let payload = shard_node.read_raw()?; let mut cursor = Cursor::new(payload.as_ref()); let children = shard_node.children()?; let mut child_iter = children.into_iter(); - let mut items: Vec = T::read_slice_from_shard(&mut cursor, &mut child_iter)?; - let count = items.len(); + // B. Construct Mutable Slice Window (SAFETY CRITICAL) + // We recreate the slice reference inside the thread. + // Constraint: Only write to [start_idx .. start_idx + expected_count] + let dest_slice = unsafe { + let ptr = (buffer_base_addr as *mut MaybeUninit).add(start_idx); + slice::from_raw_parts_mut(ptr, expected_count) + }; - // Safety Checks - // Verify shard integrity before touching the shared buffer. - if count != expected_count { + // C. Decode Direct-to-Memory + let items_read = T::read_into_slice(&mut cursor, &mut child_iter, dest_slice)?; + + if items_read != expected_count { return Err(ParcodeError::Format(format!( - "Shard integrity error: Expected {} items, found {}", - expected_count, count + "Shard {} items mismatch: expected {}, got {}", + shard_idx, expected_count, items_read ))); } - if start_idx + count > total_items { - return Err(ParcodeError::Format( - "Shard items overflowed allocated buffer".into(), - )); - } - - #[allow(unsafe_code)] - unsafe { - let base_ptr = - std::ptr::with_exposed_provenance_mut::>(buffer_start_addr); - - let dest_uninit = base_ptr.add(start_idx); - - let dest_ptr = dest_uninit.cast::(); - let src_ptr = items.as_ptr(); - - std::ptr::copy_nonoverlapping(src_ptr, dest_ptr, count); - - items.set_len(0); - } Ok(()) }, - )?; - - // 5. Bless the buffer - #[allow(unsafe_code)] - let final_vec = unsafe { - let mut manual_buffer = ManuallyDrop::new(result_buffer); - Vec::from_raw_parts( - manual_buffer.as_mut_ptr() as *mut T, - manual_buffer.len(), - manual_buffer.capacity(), - ) - }; + ); - Ok(final_vec) + // 6. Handle Result + match exec_result { + Ok(_) => { + // SAFETY: All shards reported success and wrote their assigned slots. + // Total items matches the buffer length. We can safely transmute. + let final_vec = unsafe { + let mut manual = ManuallyDrop::new(result_buffer); + Vec::from_raw_parts( + manual.as_mut_ptr() as *mut T, + manual.len(), + manual.capacity(), + ) + }; + Ok(final_vec) + } + Err(e) => { + // MEMORY LEAK WARNING: + // If we error here, `result_buffer` (Vec) is dropped. + // `MaybeUninit` does NOT drop its payload. + // So any items T that were successfully initialized in other threads will NOT be dropped. + // This is safe (no UB), but it is a memory leak of T's resources. + // Fixing this requires tracking exactly which ranges were initialized, which is complex + // in a parallel context. Given that this is an error path (file corruption), + // leaking resources is acceptable vs UB. + Err(e) + } + } } // --- COLLECTION UTILITIES --- diff --git a/tests/miri_test.rs b/tests/miri_test.rs new file mode 100644 index 0000000..e3be131 --- /dev/null +++ b/tests/miri_test.rs @@ -0,0 +1,66 @@ +#![allow(missing_docs)] +#[cfg(miri)] +mod miri_tests { + use parcode::{Parcode, ParcodeObject}; + use serde::{Deserialize, Serialize}; + + // Estructura simple para el test + #[derive(Serialize, Deserialize, ParcodeObject, Debug, PartialEq, Clone)] + struct DataChunk { + id: u32, + payload: Vec, + } + + #[derive(Serialize, Deserialize, ParcodeObject, Debug, PartialEq)] + struct World { + #[parcode(chunkable)] + chunks: Vec, + } + + #[test] + fn test_parallel_reconstruction_safety() { + // 1. Generar datos (Pequeños, para que Miri acabe rápido) + // Usamos suficientes items para forzar la creación de múltiples shards + // si la heurística lo permite, o forzamos shards manualmente si pudiéramos. + // Con 100 items y overhead, debería ser suficiente para ejercitar el loop. + let chunks: Vec = (0..100) + .map(|i| DataChunk { + id: i, + payload: vec![i as u8; 10], // Pequeño payload + }) + .collect(); + + let world = World { + chunks: chunks.clone(), + }; + + // 2. Guardar en memoria (Buffer) + let mut buffer = Vec::new(); + Parcode::builder() + .write_to_writer(&mut buffer, &world) + .expect("Write failed"); + + // 3. Leer desde memoria (Simulando archivo) + // Miri no soporta mmap ni File I/O real fácilmente. + // TRUCO: ParcodeReader::open requiere un Path. + // Para testear la lógica interna de `decode_parallel_collection` con Miri, + // necesitamos un entorno que no use mmap si es posible, o usar archivos temporales. + // Miri soporta File I/O básico con aislamiento. + + let path = "miri_test.par"; + std::fs::write(path, &buffer).expect("File write failed"); + + // 4. Ejecutar la lectura (Aquí Miri vigilará cada puntero) + let reader = Parcode::open(path).expect("Open failed"); + let world_mirror = reader.read_lazy::().expect("Lazy read failed"); + + // Esto dispara `decode_parallel_collection` y el bloque unsafe + let loaded_chunks = world_mirror.chunks.load().expect("Load failed"); + + // 5. Validar datos + assert_eq!(loaded_chunks, chunks); + + // Cleanup + std::fs::remove_file(path).unwrap(); + } +} From 0adeb1c90518e522b30e09134d9a8e25671c1252 Mon Sep 17 00:00:00 2001 From: Retype15 Date: Fri, 26 Dec 2025 14:44:54 -0500 Subject: [PATCH 2/3] feat: Implement an slice on reader to move chunks without memcpy. --- src/reader.rs | 76 +++++++++++++++++++++++++++------------------------ 1 file changed, 40 insertions(+), 36 deletions(-) diff --git a/src/reader.rs b/src/reader.rs index 9738eac..1e29684 100644 --- a/src/reader.rs +++ b/src/reader.rs @@ -324,8 +324,8 @@ pub trait ParcodeItem: Sized + Send + Sync + 'static { children: &mut std::vec::IntoIter>, ) -> Result; - /// DEPRECATED in favor of read_into_slice internally, but kept for API compat. - /// Default impl now delegates to read_into_slice to reduce code duplication. + /// DEPRECATED in favor of `read_into_slice` internally, but kept for API compat. + /// Default impl now delegates to `read_into_slice` to reduce code duplication. fn read_slice_from_shard( reader: &mut std::io::Cursor<&[u8]>, children: &mut std::vec::IntoIter>, @@ -393,9 +393,9 @@ pub trait ParcodeItem: Sized + Send + Sync + 'static { } // 3. Loop and Initialize - for i in 0..count { + for slot in destination.iter_mut().take(count) { let item = Self::read_from_shard(reader, children)?; - destination[i].write(item); + slot.write(item); } Ok(count) @@ -430,13 +430,12 @@ macro_rules! impl_primitive_parcode_item { // If T is u8, we can memcpy directly from the reader slice. if std::any::TypeId::of::() == std::any::TypeId::of::() { - let pos = reader.position() as usize; + let pos = usize::try_from(reader.position()) + .map_err(|_| ParcodeError::Format("Position overflow".into()))?; let inner = reader.get_ref(); - if inner.len() < pos + count { - return Err(ParcodeError::Format("Unexpected EOF reading u8 blob".into())); - } - let src_slice = &inner[pos..pos+count]; + let src_slice = inner.get(pos..pos + count) + .ok_or_else(|| ParcodeError::Format("Unexpected EOF reading u8 blob".into()))?; let dest_ptr = destination.as_mut_ptr() as *mut u8; #[allow(unsafe_code)] @@ -449,10 +448,10 @@ macro_rules! impl_primitive_parcode_item { } // Default Primitive Loop - for i in 0..count { + for slot in destination.iter_mut().take(count) { let item: Self = bincode::serde::decode_from_std_read(reader, bincode::config::standard()) .map_err(|e| ParcodeError::Serialization(e.to_string()))?; - destination[i].write(item); + slot.write(item); } Ok(count) } @@ -516,15 +515,28 @@ where continue; } - let count = u32::from_le_bytes(payload.get(0..4).unwrap().try_into().unwrap()) as usize; + let count = u32::from_le_bytes( + payload + .get(0..4) + .ok_or_else(|| ParcodeError::Format("Payload too short for count".into()))? + .try_into() + .map_err(|_| ParcodeError::Format("Failed to parse count".into()))?, + ) as usize; let offsets_start = 8 + (count * 8); let data_start = offsets_start + (count * 4); - let offsets_bytes = payload.get(offsets_start..data_start).unwrap(); - - for i in 0..count { - let off_bytes = offsets_bytes.get(i * 4..(i + 1) * 4).unwrap(); - let offset = u32::from_le_bytes(off_bytes.try_into().unwrap()) as usize; - let data_slice = payload.get(data_start + offset..).unwrap(); + let offsets_bytes = payload + .get(offsets_start..data_start) + .ok_or_else(|| ParcodeError::Format("Offsets out of bounds".into()))?; + + for off_bytes in offsets_bytes.chunks_exact(4).take(count) { + let offset = u32::from_le_bytes( + off_bytes + .try_into() + .map_err(|_| ParcodeError::Format("Failed to parse offset".into()))?, + ) as usize; + let data_slice = payload + .get(data_start + offset..) + .ok_or_else(|| ParcodeError::Format("Data slice out of bounds".into()))?; let (k, v) = bincode::serde::decode_from_slice(data_slice, bincode::config::standard()) .map_err(|e| ParcodeError::Serialization(e.to_string()))? @@ -864,7 +876,7 @@ impl<'a> ChunkNode<'a> { let mut cursor = std::io::Cursor::new(payload.as_ref()); let children = self.children()?; let mut child_iter = children.into_iter(); - // Use legacy read_slice helper (which internally creates a Vec, acceptable for small inline data) + // Use read_slice helper return T::read_slice_from_shard(&mut cursor, &mut child_iter); } @@ -874,7 +886,7 @@ impl<'a> ChunkNode<'a> { .get(0..8) .ok_or_else(|| ParcodeError::Format("Payload too short".into()))? .try_into() - .unwrap(), + .map_err(|_| ParcodeError::Format("Failed to parse total items".into()))?, )) .map_err(|_| ParcodeError::Format("total_items exceeds usize".into()))?; @@ -915,36 +927,34 @@ impl<'a> ChunkNode<'a> { // 4. Allocate Uninitialized Buffer let mut result_buffer: Vec> = Vec::with_capacity(total_items); - // SAFETY: We strictly control initialization via the parallel loop below. unsafe { result_buffer.set_len(total_items); } - // 5. Parallel Execution (Destination Passing) - // We cast the buffer to a raw usize address to allow sending it to threads. + // 5. Parallel Execution + // Cast the buffer to a raw usize address to allow sending it to threads. // `MaybeUninit` is not implicitly Sync, but since each thread writes to disjoint regions, // it is sound. let buffer_base_addr = result_buffer.as_mut_ptr() as usize; - // Note: try_for_each stops on the first error. let exec_result = shard_jobs.into_par_iter().try_for_each( move |(shard_idx, start_idx, expected_count)| -> Result<()> { - // A. Load Chunk + // 1. Load Chunk let shard_node = self.get_child_by_index(shard_idx)?; let payload = shard_node.read_raw()?; let mut cursor = Cursor::new(payload.as_ref()); let children = shard_node.children()?; let mut child_iter = children.into_iter(); - // B. Construct Mutable Slice Window (SAFETY CRITICAL) + // 2. Construct Mutable Slice Window // We recreate the slice reference inside the thread. - // Constraint: Only write to [start_idx .. start_idx + expected_count] + // start_idx .. start_idx + expected_count let dest_slice = unsafe { let ptr = (buffer_base_addr as *mut MaybeUninit).add(start_idx); slice::from_raw_parts_mut(ptr, expected_count) }; - // C. Decode Direct-to-Memory + // 3. Decode Direct-to-Memory let items_read = T::read_into_slice(&mut cursor, &mut child_iter, dest_slice)?; if items_read != expected_count { @@ -961,8 +971,6 @@ impl<'a> ChunkNode<'a> { // 6. Handle Result match exec_result { Ok(_) => { - // SAFETY: All shards reported success and wrote their assigned slots. - // Total items matches the buffer length. We can safely transmute. let final_vec = unsafe { let mut manual = ManuallyDrop::new(result_buffer); Vec::from_raw_parts( @@ -974,6 +982,7 @@ impl<'a> ChunkNode<'a> { Ok(final_vec) } Err(e) => { + // TODO: \ // MEMORY LEAK WARNING: // If we error here, `result_buffer` (Vec) is dropped. // `MaybeUninit` does NOT drop its payload. @@ -1014,7 +1023,7 @@ impl<'a> ChunkNode<'a> { return Err(ParcodeError::Format("Invalid container payload".into())); } - // Skip total_items (8 bytes) + // Skip total_items let runs_data = payload.get(8..).unwrap_or(&[]); let shard_runs: Vec = bincode::serde::decode_from_slice(runs_data, bincode::config::standard()) @@ -1047,7 +1056,6 @@ impl<'a> ChunkNode<'a> { let shard_node = self.get_child_by_index(target_shard_idx)?; - // New logic: let payload = shard_node.read_raw()?; let mut cursor = std::io::Cursor::new(payload.as_ref()); let children = shard_node.children()?; @@ -1128,7 +1136,6 @@ impl<'a> ChunkNode<'a> { if global_index < current_base + total_run { let offset = global_index - current_base; - // Integer division gives logical shard, modulo gives index inside return Ok((shard_base + (offset / count), offset % count)); } current_base += total_run; @@ -1159,7 +1166,6 @@ impl<'a> ChunkNode<'a> { /// Calculates the size of the payload (excluding metadata/footer). pub fn payload_len(&self) -> u64 { - // payload_end_offset is calculated in get_chunk, usually: // offset + (payload_end - offset) self.payload_end_offset - self.offset } @@ -1226,7 +1232,6 @@ impl<'a, T: ParcodeItem> Iterator for ChunkIterator<'a, T> { .container .get_child_by_index(self.current_shard_idx) .and_then(|node| { - // New logic: let payload = node.read_raw()?; let mut cursor = std::io::Cursor::new(payload.as_ref()); let children = node.children()?; @@ -1238,7 +1243,6 @@ impl<'a, T: ParcodeItem> Iterator for ChunkIterator<'a, T> { Ok(items) => { self.current_items_in_shard = items.into_iter(); self.current_shard_idx += 1; - // Recursively call next to yield the first item of the new shard self.next() } Err(e) => Some(Err(e)), From 037fa01e9a8f5edff7a5c5144e9b857119d25dc5 Mon Sep 17 00:00:00 2001 From: Retype15 Date: Fri, 26 Dec 2025 15:10:38 -0500 Subject: [PATCH 3/3] feat: Implement an slice on reader to move chunks without memcpy. --- .gitignore | 3 ++- tests/integration_tests.rs | 10 ---------- tests/lazy_iterator_tests.rs | 4 ++-- tests/macro_test.rs | 7 ++----- tests/map_test.rs | 14 ++++++-------- tests/miri_test.rs | 24 +++++++----------------- 6 files changed, 19 insertions(+), 43 deletions(-) diff --git a/.gitignore b/.gitignore index 9bcf0e7..140afd3 100644 --- a/.gitignore +++ b/.gitignore @@ -23,4 +23,5 @@ Cargo.lock parcode.txt road.md .gitignore -CONTRIBUTING.md \ No newline at end of file +CONTRIBUTING.md +internal_todo.md diff --git a/tests/integration_tests.rs b/tests/integration_tests.rs index 6fc47d9..3305b3a 100644 --- a/tests/integration_tests.rs +++ b/tests/integration_tests.rs @@ -1,5 +1,3 @@ -//! Integration test suite for Parcode. - #![allow(missing_docs)] use parcode::{ @@ -13,8 +11,6 @@ use std::fs::{File, OpenOptions}; use std::io::Write; use tempfile::NamedTempFile; -// --- TEST INFRASTRUCTURE --- - #[derive(Clone, Serialize, Deserialize, Debug, PartialEq, ParcodeObject)] struct TestUser { id: u64, @@ -69,8 +65,6 @@ impl ParcodeVisitor for UserDirectory { } // 2. DELEGATE TO VEC - // Here we propagate None, but we could propagate config_override if we wanted the parent's - // config to affect the child users. self.users.visit(graph, Some(my_id), None); } @@ -79,8 +73,6 @@ impl ParcodeVisitor for UserDirectory { } } -// --- TESTS --- - #[test] fn test_primitive_lifecycle() -> Result<()> { let user = TestUser { @@ -91,10 +83,8 @@ fn test_primitive_lifecycle() -> Result<()> { let file = NamedTempFile::new()?; - // WRITE Parcode::save(file.path(), &user)?; - // READ let loaded_user: TestUser = Parcode::load(file.path())?; assert_eq!(user, loaded_user); diff --git a/tests/lazy_iterator_tests.rs b/tests/lazy_iterator_tests.rs index c5ece27..c6cdbf1 100644 --- a/tests/lazy_iterator_tests.rs +++ b/tests/lazy_iterator_tests.rs @@ -52,12 +52,12 @@ fn test_lazy_iterator_robust() { .first() .expect("first() failed") .expect("first() returned None"); - assert_eq!(first_lazy.id, 0); // Inlined + assert_eq!(first_lazy.id, 0); assert_eq!( first_lazy.name.load().expect("Failed to load name"), "Item 0" ); // Chunkable - assert_eq!(first_lazy.data.len(), 10); // Chunkable collection + assert_eq!(first_lazy.data.len(), 10); let last_lazy = items_promise .last() diff --git a/tests/macro_test.rs b/tests/macro_test.rs index 68edb0e..7e5912f 100644 --- a/tests/macro_test.rs +++ b/tests/macro_test.rs @@ -4,13 +4,12 @@ use parcode::{Parcode, ParcodeObject}; use serde::{Deserialize, Serialize}; use tempfile::NamedTempFile; -// Ergonomic Definition V3 #[derive(Clone, Debug, PartialEq, Serialize, Deserialize, ParcodeObject)] struct LevelState { id: u32, name: String, - #[parcode(chunkable)] // Automatic settings + #[parcode(chunkable)] indices: Vec, #[cfg(feature = "lz4_flex")] #[parcode(chunkable, compression = "lz4")] // Explicit LZ4 @@ -24,15 +23,13 @@ fn test_macro_ergonomics() { name: "Dungeon_01".into(), indices: (0..10_000).map(|i| i * 2).collect(), #[cfg(feature = "lz4_flex")] - assets: vec![0xAA; 200_000], // 200KB -> This will enable sharding + LZ4 + assets: vec![0xAA; 200_000], // 200KB }; let file = NamedTempFile::new().expect("Failed to create temp file"); - // 1. Save (The macro handles the entire graph) Parcode::save(file.path(), &level).expect("Failed to save parcode data"); - // 2. Load (The macro handles reconstruction) let loaded: LevelState = Parcode::load(file.path()).expect("Failed to read parcode data"); assert_eq!(level, loaded); diff --git a/tests/map_test.rs b/tests/map_test.rs index 5f00bd1..f9e7d3a 100644 --- a/tests/map_test.rs +++ b/tests/map_test.rs @@ -5,11 +5,10 @@ use serde::{Deserialize, Serialize}; use std::collections::HashMap; use tempfile::NamedTempFile; -// Estructura que usa el modo mapa optimizado #[derive(Serialize, Deserialize, ParcodeObject)] struct UserDatabase { id: u32, - #[parcode(map, compression = "lz4")] // Activamos modo Mapa y LZ4 + #[parcode(map, compression = "lz4")] users: HashMap, } @@ -21,7 +20,7 @@ struct UserProfile { #[test] fn test_optimized_map_access() { - // 1. Generar Datos (Suficientes para provocar sharding real) + // 1. Generate data let mut users = HashMap::new(); for i in 0..5000 { users.insert( @@ -41,13 +40,13 @@ fn test_optimized_map_access() { let file = NamedTempFile::new().expect("Failed to create temp file"); Parcode::save(file.path(), &db).expect("Failed to save parcode data"); - // 2. Lectura Lazy con Acceso Aleatorio O(1) + // 2. Lazy reading with random access O(1) let file_handle = Parcode::open(file.path()).expect("Failed to open file"); let lazy_db = file_handle .root::() .expect("Failed to read lazy"); - // A. Búsqueda Exitosa (Random Access) + // 3. Successful search (Random Access) let target_key = "user_4242".to_string(); let profile = lazy_db .users @@ -58,15 +57,14 @@ fn test_optimized_map_access() { assert_eq!(profile.level, 4242 % 100); assert_eq!(profile.score, 42420); - // B. Búsqueda Fallida (No existe) + // 4. Failed search let missing = lazy_db .users .get(&"admin_root".to_string()) .expect("Failed to get missing user"); assert!(missing.is_none()); - // C. Carga Completa (Fallback a Vec<(K,V)> -> HashMap) - // El método .load() devuelve el HashMap completo reconstruido + // 5. Complete loading (Fallback to Vec<(K,V)> -> HashMap) let loaded_map = lazy_db.users.load().expect("Failed to load map"); assert_eq!(loaded_map.len(), 5000); assert_eq!( diff --git a/tests/miri_test.rs b/tests/miri_test.rs index e3be131..2a8e786 100644 --- a/tests/miri_test.rs +++ b/tests/miri_test.rs @@ -1,10 +1,11 @@ +//! MIRIFLAGS="-Zmiri-permissive-provenance -Zmiri-disable-stacked-borrows -Zmiri-disable-isolation" + #![allow(missing_docs)] #[cfg(miri)] mod miri_tests { use parcode::{Parcode, ParcodeObject}; use serde::{Deserialize, Serialize}; - // Estructura simple para el test #[derive(Serialize, Deserialize, ParcodeObject, Debug, PartialEq, Clone)] struct DataChunk { id: u32, @@ -19,14 +20,10 @@ mod miri_tests { #[test] fn test_parallel_reconstruction_safety() { - // 1. Generar datos (Pequeños, para que Miri acabe rápido) - // Usamos suficientes items para forzar la creación de múltiples shards - // si la heurística lo permite, o forzamos shards manualmente si pudiéramos. - // Con 100 items y overhead, debería ser suficiente para ejercitar el loop. let chunks: Vec = (0..100) .map(|i| DataChunk { id: i, - payload: vec![i as u8; 10], // Pequeño payload + payload: vec![i as u8; 10], }) .collect(); @@ -34,30 +31,23 @@ mod miri_tests { chunks: chunks.clone(), }; - // 2. Guardar en memoria (Buffer) let mut buffer = Vec::new(); Parcode::builder() .write_to_writer(&mut buffer, &world) .expect("Write failed"); - // 3. Leer desde memoria (Simulando archivo) - // Miri no soporta mmap ni File I/O real fácilmente. - // TRUCO: ParcodeReader::open requiere un Path. - // Para testear la lógica interna de `decode_parallel_collection` con Miri, - // necesitamos un entorno que no use mmap si es posible, o usar archivos temporales. - // Miri soporta File I/O básico con aislamiento. - let path = "miri_test.par"; std::fs::write(path, &buffer).expect("File write failed"); - // 4. Ejecutar la lectura (Aquí Miri vigilará cada puntero) + let reader: World = Parcode::load(path).expect("Load failed"); + + assert_eq!(reader, world); + let reader = Parcode::open(path).expect("Open failed"); let world_mirror = reader.read_lazy::().expect("Lazy read failed"); - // Esto dispara `decode_parallel_collection` y el bloque unsafe let loaded_chunks = world_mirror.chunks.load().expect("Load failed"); - // 5. Validar datos assert_eq!(loaded_chunks, chunks); // Cleanup