From a782e90001c18f7b963b5c9025bd6d18dd7a80cc Mon Sep 17 00:00:00 2001 From: Retype15 Date: Mon, 29 Dec 2025 16:40:05 -0500 Subject: [PATCH 1/4] feat: Implement features "mmap", "parallel" and "common"(Default) - **async:** Implement rayon and make loading and saving multithreaded. - **mmap:** Includes mmap and uses it in open as in <=v0.5.3: `unsafe { Mmap::map(&file)? };`. - **common:** Enables the two previous features as one. Added alternative methods to use a buffer directly. --- Cargo.lock | 2 +- Cargo.toml | 11 +- src/api.rs | 35 +++++- src/executor.rs | 113 +++++++++++++++--- src/io.rs | 10 +- src/reader.rs | 300 +++++++++++++++++++++++++++--------------------- 6 files changed, 316 insertions(+), 155 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 852628d..6f9f8da 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -355,7 +355,7 @@ dependencies = [ [[package]] name = "parcode" -version = "0.5.3" +version = "0.6.0" dependencies = [ "bincode", "criterion", diff --git a/Cargo.toml b/Cargo.toml index db3056b..fee090d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "parcode" -version = "0.5.3" +version = "0.6.0" edition = "2024" authors = ["RetypeOS"] description = "A high-performance, lazy load and parallelized caching library for complex Rust data structures." @@ -26,15 +26,18 @@ exclude = [ parcode-derive = { version = "0.4.0" } # Third bincode = { version = "2.0.1", features = ["serde"] } -memmap2 = "0.9.9" -rayon = "1.11.0" serde = { version = "1.0.228", features = ["derive", "rc"] } twox-hash = "2.1.2" # Optional +rayon = { version = "1.11.0", optional = true } +memmap2 = { version = "0.9.9", optional = true } lz4_flex = { version = "0.12.0", optional = true } [features] -default = [] +default = ["common"] +common = ["parallel", "mmap"] +parallel = ["dep:rayon"] +mmap = ["dep:memmap2"] lz4_flex = ["dep:lz4_flex"] [dev-dependencies] diff --git a/src/api.rs b/src/api.rs index d565ef5..ca70cd3 100644 --- a/src/api.rs +++ b/src/api.rs @@ -99,17 +99,39 @@ impl Parcode { T: ParcodeNative, P: AsRef, { - // One-liner: Open -> Load ParcodeFile::open(path)?.load() } - /// Opens a Parcode file for advanced usage (Lazy loading, Inspection). + /// PLACEHOLDER + pub fn load_bytes(data: Vec) -> Result { + ParcodeFile::from_bytes(data)?.load() + } + + /// Opens a Parcode resource using the configured storage backend. + /// + /// # Arguments + /// * `input`: Can be: + /// - `Vec`: Direct memory (WASM compatible). + /// - `Path / String`: File system path (Requires `mmap` or fallback read). /// - /// Returns a [`ParcodeFile`] handle. + /// # Example + /// ```rust,ignore + /// // Desktop (Mmap) + /// Parcode::open("data.par")?; + /// + /// // WASM (Memory) + /// let bytes = fetch(...).await; + /// Parcode::open(bytes)?; + /// ``` pub fn open>(path: P) -> Result { ParcodeFile::open(path) } + /// PLACEHOLDER + pub fn open_bytes(data: Vec) -> Result { + ParcodeFile::from_bytes(data) + } + /// Serializes an object synchronously (single-threaded) with default settings. pub fn save_sync(path: P, root_object: &T) -> Result<()> where @@ -125,6 +147,12 @@ impl Parcode { pub fn inspect>(path: P) -> Result { ParcodeInspector::inspect(path) } + + /// PLACEHOLDER + pub fn inspect_bytes(data: Vec) -> Result { + let file = ParcodeFile::from_bytes(data)?; + ParcodeInspector::inspect_file(&file) + } } /// Builder for configuring serialization options (compression, etc.). @@ -305,6 +333,7 @@ impl ParcodeOptions { /// /// - `path`: The file path to write to. If the file exists, it will be truncated. /// - `root_object`: A reference to the object to serialize. + #[cfg(not(target_arch = "wasm32"))] pub fn write(&self, path: P, root_object: &T) -> Result<()> where T: ParcodeVisitor + Sync, diff --git a/src/executor.rs b/src/executor.rs index 401212f..1296256 100644 --- a/src/executor.rs +++ b/src/executor.rs @@ -57,6 +57,20 @@ pub fn execute_graph<'a, W: Write + Send>( writer: &SeqWriter, registry: &CompressorRegistry, use_compression: bool, +) -> Result { + #[cfg(all(feature = "parallel", not(target_arch = "wasm32")))] + return execute_graph_parallel(graph, writer, registry, use_compression); + + #[cfg(not(feature = "parallel"))] + return execute_graph_serial_fallback(graph, writer, registry, use_compression); +} +// ------------ +#[cfg(all(feature = "parallel", not(target_arch = "wasm32")))] +fn execute_graph_parallel<'a, W: Write + Send>( + graph: &TaskGraph<'a>, + writer: &SeqWriter, + registry: &CompressorRegistry, + use_compression: bool, ) -> Result { // 1. Setup the shared context. let ctx = ExecutionContext { @@ -110,6 +124,88 @@ pub fn execute_graph<'a, W: Write + Send>( (*root_guard).ok_or_else(|| ParcodeError::Internal("Graph execution incomplete".into())) } +#[cfg(not(feature = "parallel"))] +fn execute_graph_serial_fallback<'a, W: Write + Send>( + graph: &TaskGraph<'a>, + writer: &SeqWriter, + registry: &CompressorRegistry, + use_compression: bool, +) -> Result { + let mut shared_buffer = Vec::with_capacity(128 * 1024); + let mut root_result: Option = None; + + // TODO: maybe i will refactor this + for node in graph.nodes().iter().rev() { + // 1. Collect children results + let completed_children_raw = { + let mut guard = node.completed_children.lock().expect("Mutex poisoned"); + std::mem::take(&mut *guard) + }; + + let children_refs: Vec = completed_children_raw + .into_iter() + .map(|opt| { + opt.ok_or_else(|| crate::ParcodeError::Internal("Missing child result".into())) + }) + .collect::>()?; + + let is_chunkable = !children_refs.is_empty(); + + // 2. Execute Job + let raw_payload = node.job.execute(&children_refs)?; + + // 3. Compress + shared_buffer.clear(); + let config = node.job.config(); + let compression_id = if use_compression && config.compression_id == 0 { + 1 + } else { + config.compression_id + }; + let compressor = registry.get(compression_id)?; + + let footer_size = if is_chunkable { + (children_refs.len() * ChildRef::SIZE) + 4 + } else { + 0 + }; + shared_buffer.reserve(raw_payload.len() + footer_size + 1); + compressor.compress_append(&raw_payload, &mut shared_buffer)?; + + // 4. Footer & Meta + if is_chunkable { + for child in &children_refs { + shared_buffer.extend_from_slice(&child.to_bytes()); + } + let count = u32::try_from(children_refs.len()).unwrap_or(u32::MAX); + shared_buffer.extend_from_slice(&count.to_le_bytes()); + } + let meta = crate::format::MetaByte::new(is_chunkable, compressor.id()); + shared_buffer.push(meta.as_u8()); + + // 5. Write to SeqWriter + let offset = writer.write_all(&shared_buffer)?; + + let my_ref = ChildRef { + offset, + length: shared_buffer.len() as u64, + }; + + // 6. Propagate to parent + if let Some(parent_id) = node.parent { + let parent_node = graph.get_node(parent_id); + let slot = node.parent_slot.expect("Parent slot missing"); + parent_node.register_child_result(slot, my_ref)?; + } else { + root_result = Some(my_ref); + } + } + + writer.flush()?; + + root_result.ok_or_else(|| crate::ParcodeError::Internal("Graph execution incomplete".into())) +} + /// Executes the graph synchronously with aggressive optimizations. /// /// # Optimizations @@ -133,21 +229,8 @@ pub fn execute_graph_sync<'a, W: Write + Send>( ) -> Result { // 1. Bypass the Mutex lock. We own the writer now. let mut raw_writer = writer.into_inner()?; - // NOTE: We assume we are writing from offset 0 relative to this execution if stream_position fails or isn't used. - // Since we are generic W, we can't rely on Seek. - // We will track offset manually starting from 0. - // If the user appended to a file, they should be aware that the returned offsets are relative to where we started writing, - // UNLESS we require Seek. But requiring Seek limits us (no TcpStream). - // For Parcode file format, absolute offsets are needed for the header. - // If writing to a stream, the header must be written at the end pointing to absolute offsets? - // Actually, Parcode format expects random access for reading, so writing to a non-seekable stream is mostly for - // "streaming out" a file that will be saved to disk later or read linearly? - // But the format relies on offsets. - // If W is a File, we can get the current offset. - // If W is Vec, we can get len. - // If W is TcpStream, we can't. - // However, for the purpose of generating the file, we just need to know the offset relative to the start of the file. - // Let's assume start offset is 0 for now for simplicity in generic context, or we could add a `base_offset` param. + // TODO: we revise this logic in another time. + // At the moment, i assume start offset is 0 for now for simplicity in generic context, or we could add a `base_offset` param. let mut current_offset = 0u64; // 2. Reusable buffer. Start with 128KB to minimize initial reallocs. diff --git a/src/io.rs b/src/io.rs index 6f7282d..c67d58f 100644 --- a/src/io.rs +++ b/src/io.rs @@ -9,11 +9,14 @@ //! - **Stability:** Eliminates channel backpressure "stop-and-go" behavior. use crate::error::{ParcodeError, Result}; -use std::fs::File; use std::io::{BufWriter, Write}; -use std::path::Path; use std::sync::Mutex; +#[cfg(not(target_arch = "wasm32"))] +use std::fs::File; +#[cfg(not(target_arch = "wasm32"))] +use std::path::Path; + /// We use a 16MB buffer. /// This allows ~128 chunks of 128KB to be written purely in memory /// before triggering a syscall. @@ -95,10 +98,13 @@ impl SeqWriter { } } +#[cfg(not(target_arch = "wasm32"))] impl SeqWriter { /// Opens the file with an optimized buffer configuration. /// /// The file is created (truncated if it exists) and wrapped in a large `BufWriter`. + /// + /// **Note:** This method is not available on WASM targets. pub fn create(path: &Path) -> Result { let file = File::create(path)?; Ok(Self::new(file)) diff --git a/src/reader.rs b/src/reader.rs index 1e29684..9b8775e 100644 --- a/src/reader.rs +++ b/src/reader.rs @@ -201,27 +201,56 @@ //! initialization overhead. All unsafe operations are carefully encapsulated and //! documented with safety invariants. -use memmap2::Mmap; -use rayon::prelude::*; use serde::{Deserialize, de::DeserializeOwned}; use std::borrow::Cow; use std::collections::HashMap; -use std::fs::File; use std::hash::Hash; use std::io::Cursor; use std::marker::PhantomData; use std::mem::{ManuallyDrop, MaybeUninit}; -use std::path::Path; +use std::ops::Deref; use std::ptr; use std::slice; use std::sync::Arc; +#[cfg(all(feature = "mmap", not(target_arch = "wasm32")))] +use memmap2::Mmap; +#[cfg(all(feature = "parallel", not(target_arch = "wasm32")))] +use rayon::prelude::*; +#[cfg(not(target_arch = "wasm32"))] +use std::fs::File; +#[cfg(not(target_arch = "wasm32"))] +use std::path::Path; + use crate::compression::CompressorRegistry; use crate::error::{ParcodeError, Result}; use crate::format::{ChildRef, GLOBAL_HEADER_SIZE, GlobalHeader, MAGIC_BYTES, MetaByte}; use crate::rt::ParcodeLazyRef; -// --- TRAIT SYSTEM FOR AUTOMATIC STRATEGY SELECTION --- +// --- + +/// PLACEHOLDER +#[derive(Debug, Clone)] +pub enum DataSource { + #[cfg(all(feature = "mmap", not(target_arch = "wasm32")))] + /// PLACEHOLDER + Mmap(Arc), + /// PLACEHOLDER + Memory(Arc>), +} + +impl Deref for DataSource { + type Target = [u8]; + + #[inline] + fn deref(&self) -> &Self::Target { + match self { + #[cfg(all(feature = "mmap", not(target_arch = "wasm32")))] + DataSource::Mmap(mmap) => mmap.as_ref(), + DataSource::Memory(vec) => vec.as_slice(), + } + } +} /// A trait for types that know how to reconstruct themselves from a [`ChunkNode`]. /// @@ -556,8 +585,8 @@ where /// - Structural inspection (`inspect`). #[derive(Debug)] pub struct ParcodeFile { - /// Memory-mapped file content. - mmap: Arc, + /// Memory file content. + source: DataSource, /// Parsed global footer/header information. header: GlobalHeader, /// Total size of the file in bytes. @@ -567,39 +596,64 @@ pub struct ParcodeFile { } impl ParcodeFile { - /// Opens a Parcode file, maps it into memory, and validates integrity. + /// Opens a Parcode file from disk. + /// + /// # Platform Behavior + /// * **Desktop/Server:** + /// - If `feature = "mmap"` is enabled (default), uses memory mapping for Zero-Copy. + /// - If `feature = "mmap"` is disabled, reads the entire file into RAM. + /// * **WASM:** This method is not available. Use `from_bytes` instead. /// /// # Errors - /// Returns error if the file does not exist, is smaller than the header, - /// or contains invalid magic bytes/version. + /// Returns error if file doesn't exist, lacks permissions, or is invalid. + #[cfg(not(target_arch = "wasm32"))] pub fn open>(path: P) -> Result { - let file = File::open(path)?; - let file_size = file.metadata()?.len(); + let path = path.as_ref(); + + // Memory mapping + #[cfg(feature = "mmap")] + { + let file = File::open(path)?; + #[allow(unsafe_code)] + let mmap = unsafe { Mmap::map(&file)? }; + + Self::init(DataSource::Mmap(Arc::new(mmap)), file.metadata()?.len()) + } + + // Read from RAM + #[cfg(not(feature = "mmap"))] + { + use std::io::Read; + + let mut file = File::open(path)?; + let file_size = file.metadata()?.len(); + let mut buffer = Vec::with_capacity(file_size as usize); + + file.read_to_end(&mut buffer)?; + + Self::init(DataSource::Memory(Arc::new(buffer)), file_size) + } + } + + /// Opens a Parcode file from an in-memory buffer. + /// + /// This is the primary entry point for tests and architectures without direct file access. + /// The `ParcodeFile` takes ownership of the data (wrapped in Arc). + pub fn from_bytes(data: Vec) -> Result { + let size = data.len() as u64; + Self::init(DataSource::Memory(Arc::new(data)), size) + } + fn init(source: DataSource, file_size: u64) -> Result { if file_size < GLOBAL_HEADER_SIZE as u64 { - return Err(ParcodeError::Format( - "File is smaller than the global header".into(), - )); + return Err(ParcodeError::Format("File smaller than header".into())); } - // 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)] - let mmap = unsafe { Mmap::map(&file)? }; - - // Read Global Header (Located at the very end of the file) - let header_start = usize::try_from(file_size) - .map_err(|_| ParcodeError::Format("File too large for address space".into()))? - - GLOBAL_HEADER_SIZE; - let header_bytes = mmap - .get(header_start..) - .ok_or_else(|| ParcodeError::Format("Header start out of bounds".into()))?; - - if header_bytes.get(0..4) != Some(&MAGIC_BYTES) { - return Err(ParcodeError::Format( - "Invalid Magic Bytes. Not a Parcode file.".into(), - )); + let header_start = (file_size - GLOBAL_HEADER_SIZE as u64) as usize; + let header_bytes = &source[header_start..]; + + if header_bytes[0..4] != MAGIC_BYTES { + return Err(ParcodeError::Format("Invalid Magic Bytes".into())); } let version = u16::from_le_bytes( @@ -620,14 +674,14 @@ impl ParcodeFile { .get(6..14) .ok_or_else(|| ParcodeError::Format("Root offset out of bounds".into()))? .try_into() - .map_err(|_| ParcodeError::Format("Failed to read root_offset".into()))?, + .map_err(|_| ParcodeError::Format("Failed to read root offset".into()))?, ); let root_length = u64::from_le_bytes( header_bytes .get(14..22) .ok_or_else(|| ParcodeError::Format("Root length out of bounds".into()))? .try_into() - .map_err(|_| ParcodeError::Format("Failed to read root_length".into()))?, + .map_err(|_| ParcodeError::Format("Failed to read root length".into()))?, ); let checksum = u32::from_le_bytes( header_bytes @@ -637,17 +691,18 @@ impl ParcodeFile { .map_err(|_| ParcodeError::Format("Failed to read checksum".into()))?, ); + let header = GlobalHeader { + magic: MAGIC_BYTES, + version, + root_offset, + root_length, + checksum, + }; + Ok(Self { - mmap: Arc::new(mmap), - header: GlobalHeader { - magic: MAGIC_BYTES, - version, - root_offset, - root_length, - checksum, - }, + source, + header, file_size, - // Initialize registry with default algorithms (NoCompression, Lz4 if enabled) registry: CompressorRegistry::new(), }) } @@ -729,29 +784,19 @@ impl ParcodeFile { let chunk_end = usize::try_from(offset + length) .map_err(|_| ParcodeError::Format("Chunk end exceeds address space".into()))?; - // Read the MetaByte (Last byte of the chunk) - let meta = MetaByte::from_byte( - *self - .mmap - .get(chunk_end - 1) - .ok_or_else(|| ParcodeError::Format("MetaByte out of bounds".into()))?, - ); + let meta = MetaByte::from_byte(self.source[chunk_end - 1]); let mut child_count = 0; - let mut payload_end = chunk_end - 1; // Default: payload ends just before MetaByte + let mut payload_end = chunk_end - 1; if meta.is_chunkable() { - // Layout: [Payload] ... [ChildRefs] [ChildCount (4 bytes)] [MetaByte (1 byte)] if length < 5 { return Err(ParcodeError::Format("Chunk too small for metadata".into())); } let count_start = chunk_end - 5; - let child_count_bytes = self - .mmap - .get(count_start..count_start + 4) - .ok_or_else(|| ParcodeError::Format("Child count out of bounds".into()))?; - child_count = Self::read_u32(child_count_bytes)?; + let count_bytes = &self.source[count_start..count_start + 4]; + child_count = Self::read_u32(count_bytes)?; let footer_size = child_count as usize * ChildRef::SIZE; let total_meta_size = 1 + 4 + footer_size; @@ -813,21 +858,10 @@ impl<'a> ChunkNode<'a> { let end = usize::try_from(self.payload_end_offset) .map_err(|_| ParcodeError::Format("End offset exceeds address space".into()))?; - if end > self.reader.mmap.len() { - return Err(ParcodeError::Format( - "Payload offset out of mmap bounds".into(), - )); - } - - let raw = self - .reader - .mmap - .get(start..end) - .ok_or_else(|| ParcodeError::Format("Payload out of bounds".into()))?; + // Access via self.reader.source (Deref to &[u8]) + let raw = &self.reader.source[start..end]; let method_id = self.meta.compression_method(); - // Delegate decompression to the registry. - // This supports pluggable algorithms (e.g., Lz4). self.reader.registry.get(method_id)?.decompress(raw) } @@ -876,7 +910,6 @@ 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 read_slice helper return T::read_slice_from_shard(&mut cursor, &mut child_iter); } @@ -937,63 +970,73 @@ impl<'a> ChunkNode<'a> { // it is sound. let buffer_base_addr = result_buffer.as_mut_ptr() as usize; - let exec_result = shard_jobs.into_par_iter().try_for_each( - move |(shard_idx, start_idx, expected_count)| -> Result<()> { - // 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(); + #[cfg(all(feature = "parallel", not(target_arch = "wasm32")))] + { + // Parallel Execution + shard_jobs.into_par_iter().try_for_each( + |(shard_idx, start_idx, expected_count)| -> Result<()> { + self.decode_shard_into_buffer::( + shard_idx, + start_idx, + expected_count, + buffer_base_addr, + ) + }, + )?; + } - // 2. Construct Mutable Slice Window - // We recreate the slice reference inside the thread. - // 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) - }; - - // 3. 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 {} items mismatch: expected {}, got {}", - shard_idx, expected_count, items_read - ))); - } + #[cfg(not(feature = "parallel"))] + { + // Serial Execution + for (shard_idx, start_idx, expected_count) in shard_jobs { + self.decode_shard_into_buffer::( + shard_idx, + start_idx, + expected_count, + buffer_base_addr, + )?; + } + } - Ok(()) - }, - ); + unsafe { + let mut manual = ManuallyDrop::new(result_buffer); + Ok(Vec::from_raw_parts( + manual.as_mut_ptr() as *mut T, + manual.len(), + manual.capacity(), + )) + } + } - // 6. Handle Result - match exec_result { - Ok(_) => { - 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) => { - // TODO: \ - // 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) - } + fn decode_shard_into_buffer( + &self, + shard_idx: usize, + start_idx: usize, + expected_count: usize, + buffer_base_addr: usize, + ) -> Result<()> { + 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(); + + #[allow(unsafe_code)] + let dest_slice = unsafe { + let ptr = (buffer_base_addr as *mut MaybeUninit).add(start_idx); + slice::from_raw_parts_mut(ptr, expected_count) + }; + + // Uses Destination Passing Style to avoid intermediate allocs + 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 {} items mismatch: expected {}, got {}", + shard_idx, expected_count, items_read + ))); } + Ok(()) } // --- COLLECTION UTILITIES --- @@ -1115,11 +1158,8 @@ impl<'a> ChunkNode<'a> { let footer_start = usize::try_from(self.payload_end_offset) .map_err(|_| ParcodeError::Format("Offset exceeds usize range".into()))?; let entry_start = footer_start + (index * ChildRef::SIZE); - let bytes = self - .reader - .mmap - .get(entry_start..entry_start + ChildRef::SIZE) - .ok_or_else(|| ParcodeError::Format("ChildRef index out of bounds".into()))?; + + let bytes = &self.reader.source[entry_start..entry_start + ChildRef::SIZE]; let r = ChildRef::from_bytes(bytes)?; self.reader.get_chunk(r.offset, r.length) From 6661c77353d07d1d5aac67c34b7a33c7fdf4aab2 Mon Sep 17 00:00:00 2001 From: Retype15 Date: Mon, 29 Dec 2025 18:52:28 -0500 Subject: [PATCH 2/4] feat: Implement features "mmap", "parallel" and "common"(Default) - **async:** Implement rayon and make loading and saving multithreaded. - **mmap:** Includes mmap and uses it in open as in <=v0.5.3: `unsafe { Mmap::map(&file)? };`. - **common:** Enables the two previous features as one. Added alternative methods to use a buffer directly (write, write_sync). - Changed: writer_to_write -> write --- .axes/axes.toml | 12 +- README.md | 2 +- examples/memory_buffer.rs | 6 +- src/api.rs | 96 +++++++++++----- src/executor.rs | 231 ++++++++------------------------------ tests/miri_test.rs | 2 +- 6 files changed, 128 insertions(+), 221 deletions(-) diff --git a/.axes/axes.toml b/.axes/axes.toml index 987b328..e9d27ae 100644 --- a/.axes/axes.toml +++ b/.axes/axes.toml @@ -2,12 +2,12 @@ name = "parcode" version = "0.5.3" description = "Blazing fast parallel caching for Rust. Serializes complex data to a chunked file format to get CPU maximum power on data loads, but as a simple binary loader." -[scripts.test] +[scripts.check] run = [ "# === Running all standard quality tests... ===", - "@> cargo fmt -- --check", - "@> cargo clippy --all-features --all-targets -- -D warnings", - "@> cargo test --all-features --all-targets", + "> cargo fmt -- --check", + "> cargo clippy -- -D warnings", + "> cargo test ", "# <#green>=== Completed! now you can make a PR. ===<#reset>", ] desc = "Run a standard quality test required from make a PR." @@ -23,6 +23,10 @@ run = "# echo 'Exiting session...'" [options.open_with] [vars] +_cargo_all = " " +_cargo_features = "" +_cargo_targets = "" +_cargo_all_feat_targets = "" GREETING = "Hello from an axes variable!" [env] diff --git a/README.md b/README.md index e3af94f..d1a124a 100644 --- a/README.md +++ b/README.md @@ -203,7 +203,7 @@ let mut buffer = Vec::new(); // Serialize directly to RAM Parcode::builder() .compression(true) - .write_to_writer(&mut buffer, &my_data)?; + .write(&mut buffer, &my_data)?; // 'buffer' now contains the full Parcode file structure ``` diff --git a/examples/memory_buffer.rs b/examples/memory_buffer.rs index 2b95497..4098214 100644 --- a/examples/memory_buffer.rs +++ b/examples/memory_buffer.rs @@ -1,6 +1,6 @@ //! Example: Writing Parcode to a Memory Buffer //! -//! This example demonstrates how to use the generic `write_to_writer` API +//! This example demonstrates how to use the generic `write` API //! to serialize a Parcode object directly into a `Vec` in memory, //! bypassing the file system. @@ -36,12 +36,12 @@ fn main() -> Result<(), Box> { let mut buffer: Vec = Vec::new(); // 3. Serialize to the buffer - // We use `write_to_writer` which accepts any W: Write + Send + // We use `write` which accepts any W: Write + Send // Vec implements Write and Send. println!("Serializing to memory buffer..."); Parcode::builder() .compression(true) // Optional: enable compression - .write_to_writer(&mut buffer, &user)?; + .write(&mut buffer, &user)?; println!( "Serialization complete. Buffer size: {} bytes", diff --git a/src/api.rs b/src/api.rs index ca70cd3..c94d42c 100644 --- a/src/api.rs +++ b/src/api.rs @@ -85,15 +85,38 @@ impl Parcode { } /// Saves an object to a file with default settings. + /// + /// **Note:** This method is not available on WASM targets. + #[cfg(not(target_arch = "wasm32"))] pub fn save(path: P, root_object: &T) -> Result<()> where T: ParcodeVisitor + Sync, P: AsRef, { - ParcodeOptions::default().write(path, root_object) + ParcodeOptions::default().save(path, root_object) } - /// Loads an object fully into memory (Eager load). + /// PLACEHOLDER + pub fn write(writer: W, root_object: &T) -> Result<()> + where + T: ParcodeVisitor + Sync, + W: Write + Send, + { + ParcodeOptions::default().write(writer, root_object) + } + + /// PLACEHOLDER + pub fn serialize(root_object: &T) -> Result> + where + T: ParcodeVisitor + Sync, + { + let mut buffer = Vec::with_capacity(4096); + Self::write(&mut buffer, root_object)?; + Ok(buffer) + } + + /// Loads an object fully into memory from a file. + #[cfg(not(target_arch = "wasm32"))] pub fn load(path: P) -> Result where T: ParcodeNative, @@ -123,6 +146,7 @@ impl Parcode { /// let bytes = fetch(...).await; /// Parcode::open(bytes)?; /// ``` + #[cfg(not(target_arch = "wasm32"))] pub fn open>(path: P) -> Result { ParcodeFile::open(path) } @@ -132,7 +156,17 @@ impl Parcode { ParcodeFile::from_bytes(data) } - /// Serializes an object synchronously (single-threaded) with default settings. + /// PLACEHOLDER + pub fn write_sync(writer: W, root_object: &T) -> Result<()> + where + T: ParcodeVisitor, + W: Write + Send, + { + ParcodeOptions::default().write_sync(writer, root_object) + } + + /// PLACEHOLDER + #[cfg(not(target_arch = "wasm32"))] pub fn save_sync(path: P, root_object: &T) -> Result<()> where T: ParcodeVisitor, @@ -144,6 +178,13 @@ impl Parcode { /// Generates a structural inspection report of a file without full initialization. /// /// This is a convenience wrapper equivalent to `ParcodeInspector::inspect`. + /// + /// # Example + /// ```rust,ignore + /// let report = Parcode::inspect("data.par")?; + /// println!("{}", report); + /// ``` + #[cfg(not(target_arch = "wasm32"))] pub fn inspect>(path: P) -> Result { ParcodeInspector::inspect(path) } @@ -321,7 +362,7 @@ impl ParcodeOptions { /// /// Serializes an object graph to disk with the configured settings. /// - /// This is a convenience wrapper around `write_to_writer` that handles file creation. + /// This is a convenience wrapper around `write` that handles file creation. /// /// ## Type Parameters /// @@ -334,13 +375,13 @@ impl ParcodeOptions { /// - `path`: The file path to write to. If the file exists, it will be truncated. /// - `root_object`: A reference to the object to serialize. #[cfg(not(target_arch = "wasm32"))] - pub fn write(&self, path: P, root_object: &T) -> Result<()> + pub fn save(&self, path: P, root_object: &T) -> Result<()> where T: ParcodeVisitor + Sync, P: AsRef, { let file = std::fs::File::create(path)?; - self.write_to_writer(file, root_object) + self.write(file, root_object) } /// Serializes the object graph to a generic writer (File, `Vec`, `TcpStream`, etc). @@ -389,15 +430,15 @@ impl ParcodeOptions { /// let mut buffer = Vec::new(); /// /// Parcode::builder() - /// .write_to_writer(&mut buffer, &data)?; + /// .write(&mut buffer, &data)?; /// # Ok::<(), parcode::ParcodeError>(()) /// ``` - pub fn write_to_writer<'a, T, W>(&self, writer: W, root_object: &'a T) -> Result<()> + pub fn write<'a, T, W>(&self, writer: W, root_object: &'a T) -> Result<()> where T: ParcodeVisitor + Sync, W: Write + Send, { - // 1. Build the Task Graph (Virtual) + // 1. Build the Task Graph let mut graph = TaskGraph::<'a>::new(); // The root has no parent (None) and no slot (None). root_object.visit(&mut graph, None, None); @@ -425,41 +466,42 @@ impl ParcodeOptions { /// - Benchmarking vs Parallel implementation. /// /// It uses less memory than `write` because it reuses a single compression buffer. + #[cfg(not(target_arch = "wasm32"))] pub fn save_sync(&self, path: P, root_object: &T) -> Result<()> where T: ParcodeVisitor, P: AsRef, { - self.write_sync(path, root_object) + let file = std::fs::File::create(path)?; + self.write_sync(file, root_object) } - /// Internal synchronous write implementation. - /// - /// Currently only supports file paths because `execute_graph_sync` consumes the writer, - /// and we need to re-open the file to append the header (simplest approach for now). - /// A future refactor could support generic writers for sync mode if needed. - pub fn write_sync<'a, T, P>(&self, path: P, root_object: &'a T) -> Result<()> + /// PLACEHOLDER + pub fn write_sync<'a, T, W>(&self, writer: W, root_object: &'a T) -> Result<()> where T: ParcodeVisitor, - P: AsRef, + W: Write + Send, { - let path = path.as_ref(); let mut graph = TaskGraph::<'a>::new(); root_object.visit(&mut graph, None, None); - // CREATE WRITER (Not borrowing it later) - let writer = SeqWriter::create(path)?; + let seq_writer = SeqWriter::new(writer); let registry = crate::compression::CompressorRegistry::new(); - // Pass writer by VALUE (move) - let root_child_ref = - crate::executor::execute_graph_sync(&graph, writer, ®istry, self.use_compression)?; + #[cfg(feature = "parallel")] + let root_child_ref = crate::executor::execute_graph_sync( + &graph, + &seq_writer, + ®istry, + self.use_compression, + )?; + + #[cfg(not(feature = "parallel"))] + let root_child_ref = execute_graph(&graph, &seq_writer, ®istry, self.use_compression)?; - // Simpler approach for this iteration: Open file to append header. - // It's a tiny write (26 bytes), overhead is negligible compared to main payload. - let mut file = std::fs::OpenOptions::new().append(true).open(path)?; let header = GlobalHeader::new(root_child_ref.offset, root_child_ref.length); - file.write_all(&header.to_bytes())?; + seq_writer.write_all(&header.to_bytes())?; + seq_writer.flush()?; Ok(()) } diff --git a/src/executor.rs b/src/executor.rs index 1296256..202e018 100644 --- a/src/executor.rs +++ b/src/executor.rs @@ -16,7 +16,7 @@ use std::io::Write; /// Context shared among all worker threads. struct ExecutionContext<'a, 'graph, W: Write + Send> { - graph: &'graph TaskGraph<'a>, // The graph holds data living for 'a + graph: &'graph TaskGraph<'a>, writer: &'graph SeqWriter, registry: &'graph CompressorRegistry, use_compression: bool, @@ -25,6 +25,7 @@ struct ExecutionContext<'a, 'graph, W: Write + Send> { root_result: Mutex>, } +#[cfg(feature = "parallel")] impl<'a, 'graph, W: Write + Send> ExecutionContext<'a, 'graph, W> { fn signal_error(&self, err: ParcodeError) { let mut guard = self.error_capture.lock().unwrap_or_else(|p| p.into_inner()); @@ -59,12 +60,18 @@ pub fn execute_graph<'a, W: Write + Send>( use_compression: bool, ) -> Result { #[cfg(all(feature = "parallel", not(target_arch = "wasm32")))] - return execute_graph_parallel(graph, writer, registry, use_compression); + { + execute_graph_parallel(graph, writer, registry, use_compression) + } - #[cfg(not(feature = "parallel"))] - return execute_graph_serial_fallback(graph, writer, registry, use_compression); + #[cfg(any(not(feature = "parallel"), target_arch = "wasm32"))] + { + execute_graph_serial_fallback(graph, writer, registry, use_compression) + } } // ------------ + +/// PLACEHOLDER #[cfg(all(feature = "parallel", not(target_arch = "wasm32")))] fn execute_graph_parallel<'a, W: Write + Send>( graph: &TaskGraph<'a>, @@ -83,8 +90,8 @@ fn execute_graph_parallel<'a, W: Write + Send>( root_result: Mutex::new(None), }; - // 2. Identify initial leaves (Nodes with 0 dependencies). - // These are the spark plugs that start the engine. + // 2. Identify initial leaves. + use std::sync::atomic::Ordering; let leaves: Vec<&Node<'a>> = graph .nodes() .iter() @@ -98,7 +105,6 @@ fn execute_graph_parallel<'a, W: Write + Send>( } // 3. Launch the Rayon Scope. - // The scope ensures all spawned threads complete before this block exits. rayon::scope(|s| { let ctx_ref = &ctx; for leaf in leaves { @@ -124,161 +130,54 @@ fn execute_graph_parallel<'a, W: Write + Send>( (*root_guard).ok_or_else(|| ParcodeError::Internal("Graph execution incomplete".into())) } -#[cfg(not(feature = "parallel"))] -fn execute_graph_serial_fallback<'a, W: Write + Send>( +/// PLACEHOLDER +pub fn execute_graph_sync<'a, W: Write + Send>( graph: &TaskGraph<'a>, writer: &SeqWriter, registry: &CompressorRegistry, use_compression: bool, ) -> Result { + // Buffer re-usable for compression. let mut shared_buffer = Vec::with_capacity(128 * 1024); let mut root_result: Option = None; - // TODO: maybe i will refactor this + // Iter nodes in reverse (ID descending) for node in graph.nodes().iter().rev() { // 1. Collect children results - let completed_children_raw = { - let mut guard = node.completed_children.lock().expect("Mutex poisoned"); - std::mem::take(&mut *guard) - }; - - let children_refs: Vec = completed_children_raw - .into_iter() - .map(|opt| { - opt.ok_or_else(|| crate::ParcodeError::Internal("Missing child result".into())) - }) - .collect::>()?; - - let is_chunkable = !children_refs.is_empty(); - - // 2. Execute Job - let raw_payload = node.job.execute(&children_refs)?; - - // 3. Compress - shared_buffer.clear(); - let config = node.job.config(); - let compression_id = if use_compression && config.compression_id == 0 { - 1 - } else { - config.compression_id - }; - let compressor = registry.get(compression_id)?; - - let footer_size = if is_chunkable { - (children_refs.len() * ChildRef::SIZE) + 4 - } else { - 0 - }; - shared_buffer.reserve(raw_payload.len() + footer_size + 1); - compressor.compress_append(&raw_payload, &mut shared_buffer)?; - - // 4. Footer & Meta - if is_chunkable { - for child in &children_refs { - shared_buffer.extend_from_slice(&child.to_bytes()); - } - let count = u32::try_from(children_refs.len()).unwrap_or(u32::MAX); - shared_buffer.extend_from_slice(&count.to_le_bytes()); - } - let meta = crate::format::MetaByte::new(is_chunkable, compressor.id()); - shared_buffer.push(meta.as_u8()); - - // 5. Write to SeqWriter - let offset = writer.write_all(&shared_buffer)?; - - let my_ref = ChildRef { - offset, - length: shared_buffer.len() as u64, - }; - - // 6. Propagate to parent - if let Some(parent_id) = node.parent { - let parent_node = graph.get_node(parent_id); - let slot = node.parent_slot.expect("Parent slot missing"); - parent_node.register_child_result(slot, my_ref)?; - } else { - root_result = Some(my_ref); - } - } - - writer.flush()?; - - root_result.ok_or_else(|| crate::ParcodeError::Internal("Graph execution incomplete".into())) -} - -/// Executes the graph synchronously with aggressive optimizations. -/// -/// # Optimizations -/// 1. **Implicit Topology:** Iterates nodes in reverse order (ID descending). -/// Since `ParcodeVisitor` builds the tree recursively (Parent creates self, then visits children), -/// children always have higher IDs than parents. Reverse iteration guarantees dependencies are met. -/// *Benefit:* Eliminates O(N) allocations for dependency tracking and queues. -/// -/// 2. **Lockless I/O:** Takes ownership of the underlying writer to bypass Mutex overhead. -/// -/// 3. **Buffer Reuse:** Uses a single, growing buffer for all compression ops. -/// -/// # Type Parameters -/// -/// * `W`: The writer type. Must implement `std::io::Write` and `Send`. -pub fn execute_graph_sync<'a, W: Write + Send>( - graph: &TaskGraph<'a>, - writer: SeqWriter, - registry: &CompressorRegistry, - use_compression: bool, -) -> Result { - // 1. Bypass the Mutex lock. We own the writer now. - let mut raw_writer = writer.into_inner()?; - // TODO: we revise this logic in another time. - // At the moment, i assume start offset is 0 for now for simplicity in generic context, or we could add a `base_offset` param. - let mut current_offset = 0u64; - - // 2. Reusable buffer. Start with 128KB to minimize initial reallocs. - let mut shared_buffer = Vec::with_capacity(128 * 1024); - let mut root_result: Option = None; - - // 3. Execution Loop: Reverse Iteration (Zero-Alloc Topology) - // The nodes are stored in a Vec. Index N is the Child, Index 0 is Root. - // Iterating N -> 0 ensures children are processed before parents. - let nodes = graph.nodes(); - for node in nodes.iter().rev() { - // --- STEP A: Gather Children Results --- - // Even though we are sync, the Node struct uses Mutex for safety. - // In this loop, contention is impossible. let completed_children_raw = { let mut guard = node .completed_children .lock() .map_err(|_| ParcodeError::Internal("Node mutex poisoned".into()))?; - // TAKE the value to free memory in the node immediately std::mem::take(&mut *guard) }; - // Efficiently unwrap without extra allocations if possible - // We can iterate the vector directly. - let is_chunkable = !completed_children_raw.is_empty(); - - // --- STEP B: Serialize (User logic) --- - // Map Option -> ChildRef - // We reconstruct the vector only because execute() signature requires slice. - // In a deeper refactor, we would change execute() to take an iterator to avoid this alloc. let children_refs: Vec = completed_children_raw .into_iter() .map(|opt| opt.ok_or_else(|| ParcodeError::Internal("Missing child result".into()))) .collect::>()?; + let is_chunkable = !children_refs.is_empty(); + + // 2. Execute Job (Serialization) let raw_payload = node.job.execute(&children_refs)?; - // --- STEP C: Compress & Format --- + // 3. Compress & Format shared_buffer.clear(); - let config = node.job.config(); + + // Default to LZ4 if compression enabled and config says 0 (None/Default) + // config.compression_id == 0 means use default. let compression_id = if use_compression && config.compression_id == 0 { - 1 // Default to LZ4 + 1 // LZ4 } else { config.compression_id }; - let compressor = registry.get(compression_id)?; + + // Fallback to None (ID 0) if requested algorithm unavailable + let compressor = registry + .get(compression_id) + .unwrap_or_else(|_| registry.get(0).unwrap()); let footer_size = if is_chunkable { (children_refs.len() * ChildRef::SIZE) + 4 @@ -286,13 +185,12 @@ pub fn execute_graph_sync<'a, W: Write + Send>( 0 }; - // Heuristic reserve shared_buffer.reserve(raw_payload.len() + footer_size + 1); - // Compress + // Compress directly to buffer compressor.compress_append(&raw_payload, &mut shared_buffer)?; - // Footer + // 4. Footer & Meta if is_chunkable { for child in &children_refs { shared_buffer.extend_from_slice(&child.to_bytes()); @@ -301,38 +199,29 @@ pub fn execute_graph_sync<'a, W: Write + Send>( shared_buffer.extend_from_slice(&count.to_le_bytes()); } - // MetaByte let meta = MetaByte::new(is_chunkable, compressor.id()); shared_buffer.push(meta.as_u8()); - // --- STEP D: Write to Disk (Lockless) --- - raw_writer.write_all(&shared_buffer)?; + // 5. Write to SeqWriter + let offset = writer.write_all(&shared_buffer)?; - let chunk_len = shared_buffer.len() as u64; let my_ref = ChildRef { - offset: current_offset, - length: chunk_len, + offset, + length: shared_buffer.len() as u64, }; - current_offset += chunk_len; - // --- STEP E: Propagate to Parent --- + // 6. Propagate result to parent if let Some(parent_id) = node.parent { - // Note: We are iterating strictly backwards. Parent ID is GUARANTEED - // to be smaller than current node.id, so parent is "future" work. let parent_node = graph.get_node(parent_id); - let slot = node.parent_slot.expect("Parent set but slot missing"); - - // This is the only Mutex usage remaining (registering result). - // It's just a memory write, extremely fast. + let slot = node.parent_slot.expect("Parent slot missing"); parent_node.register_child_result(slot, my_ref)?; } else { - // No parent? Must be root. root_result = Some(my_ref); } } - // Flush is critical - raw_writer.flush()?; + // Flush writer buffers + writer.flush()?; root_result .ok_or_else(|| ParcodeError::Internal("Graph execution completed but no root found".into())) @@ -340,6 +229,7 @@ pub fn execute_graph_sync<'a, W: Write + Send>( /// The worker function executed by Rayon threads. /// It handles Serialization -> Compression -> Writing -> Notification. +#[cfg(all(feature = "parallel", not(target_arch = "wasm32")))] fn process_node<'scope, 'a, W: Write + Send>( scope: &rayon::Scope<'scope>, ctx: &'scope ExecutionContext<'a, 'scope, W>, @@ -350,7 +240,7 @@ fn process_node<'scope, 'a, W: Write + Send>( return; } - // --- STEP 1: PREPARE DATA (CPU BOUND) --- + // 1. Prepare data let completed_children_raw = { let lock = node @@ -366,9 +256,6 @@ fn process_node<'scope, 'a, W: Write + Send>( } }; - // Retrieve the results from children (if any). - // We lock the mutex, take the vector out (swap with empty) to consume it efficiently. - // Since we used slots, the order is already correct. We just need to unwrap the Options. let children_refs: Vec = match completed_children_raw .into_iter() .map(|opt| opt.ok_or_else(|| ParcodeError::Internal("Missing child result".into()))) @@ -381,15 +268,8 @@ fn process_node<'scope, 'a, W: Write + Send>( } }; - // A node is "Chunkable" (Bit 0 set) if it had children dependencies. - // Note: The optimizer pass (Phase 2) might have removed children and inlined them, - // so we trust `children_refs`. let is_chunkable = !children_refs.is_empty(); - // Execute the job (Serialization). - // This is where the user's `Vec` turns into bytes. - // We pass the children refs so they can be embedded in the footer/table if needed - // by the specific implementation, or we append the standard footer below. let raw_payload = match node.job.execute(&children_refs) { Ok(bytes) => bytes, Err(e) => { @@ -398,14 +278,11 @@ fn process_node<'scope, 'a, W: Write + Send>( } }; - // --- STEP 2 & 3: COMPRESSION & FORMATTING (CPU BOUND) --- + // 2&3. Compression and formatting - // 1. Get Job Config let config = node.job.config(); - - // 2. Find Algorithm let compression_id = if ctx.use_compression && config.compression_id == 0 { - 1 // Default to LZ4 + 1 } else { config.compression_id }; @@ -418,26 +295,20 @@ fn process_node<'scope, 'a, W: Write + Send>( } }; - // 3. Prepare Final Buffer - // We need to calculate footer size to reserve space. let footer_size = if is_chunkable { (children_refs.len() * ChildRef::SIZE) + 4 } else { 0 }; - // Heuristic: Allocate enough for raw payload + footer + meta. - // If compression shrinks it, great. If it expands (rare), it will realloc. let estimated_capacity = raw_payload.len() + footer_size + 1; let mut final_buffer = Vec::with_capacity(estimated_capacity); - // 4. Compress directly into final buffer if let Err(e) = compressor.compress_append(&raw_payload, &mut final_buffer) { ctx.signal_error(e); return; } - // 5. Append Footer if is_chunkable { for child in &children_refs { final_buffer.extend_from_slice(&child.to_bytes()); @@ -446,12 +317,10 @@ fn process_node<'scope, 'a, W: Write + Send>( final_buffer.extend_from_slice(&count.to_le_bytes()); } - // 6. Append MetaByte let meta = MetaByte::new(is_chunkable, compressor.id()); final_buffer.push(meta.as_u8()); - // --- STEP 4: WRITING (I/O BOUND) --- - // This is the only serialization point (Mutex). + // 4. Writting let write_result = ctx.writer.write_all(&final_buffer); let offset = match write_result { @@ -462,36 +331,28 @@ fn process_node<'scope, 'a, W: Write + Send>( } }; - // Create the Reference for this newly written chunk. let my_ref = ChildRef { offset, length: final_buffer.len() as u64, }; - // --- STEP 5: PROPAGATION --- + // 5. Propagate if let Some(parent_id) = node.parent { let parent_node = ctx.graph.get_node(parent_id); - // A. Register our result in the parent let slot = node.parent_slot.expect("Parent set but slot missing"); if let Err(e) = parent_node.register_child_result(slot, my_ref) { ctx.signal_error(e); return; } - // B. Decrement parent's dependency counter. - // `fetch_sub` returns the PREVIOUS value. - // If previous was 1, it means it is NOW 0 -> Ready to fire. let prev_deps = parent_node.atomic_deps.fetch_sub(1, Ordering::SeqCst); if prev_deps == 1 { - // We are the last child! We have the honor of waking the parent. - // Spawn into the existing scope. scope.spawn(move |s| process_node(s, ctx, parent_node)); } } else { - // We have no parent. We are the ROOT. ctx.capture_root_result(my_ref); } } diff --git a/tests/miri_test.rs b/tests/miri_test.rs index 2a8e786..0fa886f 100644 --- a/tests/miri_test.rs +++ b/tests/miri_test.rs @@ -33,7 +33,7 @@ mod miri_tests { let mut buffer = Vec::new(); Parcode::builder() - .write_to_writer(&mut buffer, &world) + .write(&mut buffer, &world) .expect("Write failed"); let path = "miri_test.par"; From 5e8811544f0179b9eb251cad26118dee2a1685ce Mon Sep 17 00:00:00 2001 From: Retype15 Date: Tue, 30 Dec 2025 03:04:59 -0500 Subject: [PATCH 3/4] feat: Implement features "mmap", "parallel" and "common"(Default) - **async:** Implement rayon and make loading and saving multithreaded. - **mmap:** Includes mmap and uses it in open as in <=v0.5.3: `unsafe { Mmap::map(&file)? };`. - **common:** Enables the two previous features as one. Added alternative methods to use a buffer directly (write, write_sync). - Changed: writer_to_write -> write - Update a majority of documentation to use this new API. --- .axes/axes.toml | 3 +- Cargo.toml | 4 +-- README.md | 40 +++++++++++----------- src/api.rs | 61 ++++++++++++---------------------- src/compression.rs | 1 - src/error.rs | 3 +- src/executor.rs | 46 ++++++++++++++++++-------- src/format.rs | 23 ++++++++----- src/graph/core.rs | 12 +++++-- src/graph/job.rs | 25 ++++++++++++++ src/inspector.rs | 72 +++++++++++++++++++--------------------- src/io.rs | 15 +++++++-- src/lib.rs | 19 ++++++----- src/map.rs | 5 ++- src/reader.rs | 79 ++++++++++++++++++++++++++++---------------- src/rt.rs | 45 ++++++++++++++++++++----- src/visitor.rs | 2 +- src/visitor_impls.rs | 6 ++-- 18 files changed, 283 insertions(+), 178 deletions(-) diff --git a/.axes/axes.toml b/.axes/axes.toml index e9d27ae..7d453a5 100644 --- a/.axes/axes.toml +++ b/.axes/axes.toml @@ -5,8 +5,9 @@ description = "Blazing fast parallel caching for Rust. Serializes complex data t [scripts.check] run = [ "# === Running all standard quality tests... ===", + "cargo build ", "> cargo fmt -- --check", - "> cargo clippy -- -D warnings", + "> cargo clippy -- -D warnings", "> cargo test ", "# <#green>=== Completed! now you can make a PR. ===<#reset>", ] diff --git a/Cargo.toml b/Cargo.toml index fee090d..514e0d3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -34,8 +34,8 @@ memmap2 = { version = "0.9.9", optional = true } lz4_flex = { version = "0.12.0", optional = true } [features] -default = ["common"] -common = ["parallel", "mmap"] +default = ["standard"] +standard = ["parallel", "mmap"] parallel = ["dep:rayon"] mmap = ["dep:memmap2"] lz4_flex = ["dep:lz4_flex"] diff --git a/README.md b/README.md index d1a124a..8935525 100644 --- a/README.md +++ b/README.md @@ -46,16 +46,23 @@ Add this to your `Cargo.toml`: ```toml [dependencies] -parcode = "0.5" +parcode = "0.6" ``` To enable LZ4 compression: ```toml [dependencies] -parcode = { version = "0.4", features = ["lz4_flex"] } +parcode = { version = "0.6", features = ["lz4_flex"] } ``` +### Features + +* `mmap`: Enable memory mapping +* `parallel`: Enable parallelization +* `standard`: Default feature, activate features 'mmap' and 'parallel' (used as Default) +* `lz4_flex`: Enable LZ4 compression to ID 1. + --- ## Usage Guide @@ -119,10 +126,10 @@ Parcode::save("savegame.par", &world)?; Use the builder mode. ```rust -// Saves with LZ4 compression enabled to all metadata. +// Saves with LZ4 compression on ALL and write parcode serialized data into a new file. Parcode::builder() .compression(true) - .write("savegame_compressed.par", &world)?; + .save("savegame_compressed.par", &world)?; ``` ### 3. Read Data (Lazy) @@ -178,13 +185,12 @@ Scan a list of heavy objects without loading their heavy payloads. ```rust // Assume we have Vec -for player_proxy in world_mirror.all_players.iter()? { - let p = player_proxy?; // Resolve result +for Ok(player_proxy) in world_mirror.all_players.iter_lazy()? { - // We can check level WITHOUT loading the player's inventory from disk! + // We can check level WITHOUT loading the player's inventory of all players from disk, only you needed! if p.level > 50 { println!("High level player found!"); - // p.inventory.load()?; + p.inventory.load()?; } } ``` @@ -201,11 +207,9 @@ Parcode isn't limited to files. You can serialize directly to any `std::io::Writ let mut buffer = Vec::new(); // Serialize directly to RAM -Parcode::builder() - .compression(true) - .write(&mut buffer, &my_data)?; +Parcode::write(&mut buffer, &my_data)?; -// 'buffer' now contains the full Parcode file structure +// 'buffer' now contains the full Parcode serialized structure ``` ### Synchronous Mode @@ -213,9 +217,7 @@ Parcode::builder() For environments where threading is not available (WASM, embedded) or to reduce memory overhead. ```rust -Parcode::builder() - .compression(true) - .write_sync("sync_save.par", &data)?; +Parcode::write_sync("sync_save.par", &data)?; ``` ### Inspector @@ -246,12 +248,12 @@ Root Offset: 550368 Control exactly how your data structure maps to disk using `#[parcode(...)]`. -| Attribute | Effect | Best For | -| :------------------------------ | :--------------------------------------------- | :--------------------------------- | -| **(none)** | Field is serialized into the parent's payload. | Small primitives (`u32`, `bool`), short Strings, flags. | +| Attribute | Effect | Best For | +| :------------------------------ | :--------------------------------------------- | :------------------------------------------------------ | +| **(none)** | Field is serialized into the parent's payload. | Small primitives (`u32`, `bool`), short Strings, flags. | | `#[parcode(chunkable)]` | Field is stored in its own independent Chunk. | Structs, Vectors, or fields you want to load lazily (`.load()`). | | `#[parcode(map)]` | Field (`HashMap`) is sharded by hash. | Large Dictionaries/Indices where you need random access (`.get()`). | -| `#[parcode(compression="lz4")]` | Overrides compression for this chunk. | Highly compressible data (text, save states). | +| `#[parcode(compression="lz4")]` | Overrides compression for this chunk. | Highly compressible data (text, save states). | --- diff --git a/src/api.rs b/src/api.rs index c94d42c..2aa990f 100644 --- a/src/api.rs +++ b/src/api.rs @@ -50,7 +50,7 @@ //! // Enable compression //! Parcode::builder() //! .compression(true) -//! .write("data_custom.par", &my_data)?; +//! .save("data_custom.par", &my_data)?; //! # std::fs::remove_file("data_custom.par")?; //! # Ok::<(), parcode::ParcodeError>(()) //! ``` @@ -96,7 +96,9 @@ impl Parcode { ParcodeOptions::default().save(path, root_object) } - /// PLACEHOLDER + /// Writes an object to a generic writer with default settings. + /// + /// This is a convenience method that uses the default [`ParcodeOptions`]. pub fn write(writer: W, root_object: &T) -> Result<()> where T: ParcodeVisitor + Sync, @@ -105,7 +107,9 @@ impl Parcode { ParcodeOptions::default().write(writer, root_object) } - /// PLACEHOLDER + /// Serializes an object into a `Vec` in memory. + /// + /// This is a convenience method that creates a buffer and calls [`Self::write`]. pub fn serialize(root_object: &T) -> Result> where T: ParcodeVisitor + Sync, @@ -125,38 +129,26 @@ impl Parcode { ParcodeFile::open(path)?.load() } - /// PLACEHOLDER - pub fn load_bytes(data: Vec) -> Result { + /// Loads an object fully into memory from a byte slice. + pub fn load_bytes(data: Vec) -> Result + where + T: ParcodeNative, + { ParcodeFile::from_bytes(data)?.load() } /// Opens a Parcode resource using the configured storage backend. - /// - /// # Arguments - /// * `input`: Can be: - /// - `Vec`: Direct memory (WASM compatible). - /// - `Path / String`: File system path (Requires `mmap` or fallback read). - /// - /// # Example - /// ```rust,ignore - /// // Desktop (Mmap) - /// Parcode::open("data.par")?; - /// - /// // WASM (Memory) - /// let bytes = fetch(...).await; - /// Parcode::open(bytes)?; - /// ``` #[cfg(not(target_arch = "wasm32"))] pub fn open>(path: P) -> Result { ParcodeFile::open(path) } - /// PLACEHOLDER + /// Opens a Parcode resource from a byte slice. pub fn open_bytes(data: Vec) -> Result { ParcodeFile::from_bytes(data) } - /// PLACEHOLDER + /// Writes an object to a generic writer synchronously. pub fn write_sync(writer: W, root_object: &T) -> Result<()> where T: ParcodeVisitor, @@ -165,7 +157,7 @@ impl Parcode { ParcodeOptions::default().write_sync(writer, root_object) } - /// PLACEHOLDER + /// Saves an object to a file synchronously. #[cfg(not(target_arch = "wasm32"))] pub fn save_sync(path: P, root_object: &T) -> Result<()> where @@ -189,7 +181,7 @@ impl Parcode { ParcodeInspector::inspect(path) } - /// PLACEHOLDER + /// Generates a structural inspection report from a byte slice. pub fn inspect_bytes(data: Vec) -> Result { let file = ParcodeFile::from_bytes(data)?; ParcodeInspector::inspect_file(&file) @@ -229,7 +221,7 @@ impl Parcode { /// let data = vec![1, 2, 3]; /// Parcode::builder() /// .compression(true) -/// .write("data_comp.par", &data)?; +/// .save("data_comp.par", &data)?; /// # std::fs::remove_file("data_comp.par")?; /// # Ok::<(), parcode::ParcodeError>(()) /// ``` @@ -284,7 +276,7 @@ impl ParcodeOptions { /// // Enable compression /// Parcode::builder() /// .compression(true) - /// .write("data.par", &my_data)?; + /// .save("data.par", &my_data)?; /// # std::fs::remove_file("data.par")?; /// # Ok::<(), parcode::ParcodeError>(()) /// ``` @@ -347,7 +339,7 @@ impl ParcodeOptions { /// // Write with compression /// Parcode::builder() /// .compression(true) - /// .write("data_write.par", &data)?; + /// .save("data_write.par", &data)?; /// # std::fs::remove_file("data_write.par")?; /// # Ok::<(), parcode::ParcodeError>(()) /// ``` @@ -363,17 +355,6 @@ impl ParcodeOptions { /// Serializes an object graph to disk with the configured settings. /// /// This is a convenience wrapper around `write` that handles file creation. - /// - /// ## Type Parameters - /// - /// - `T`: The type to serialize. Must implement [`ParcodeVisitor`] - /// and `Sync` (for parallel execution). - /// - `P`: The path type (anything that implements `AsRef`). - /// - /// ## Parameters - /// - /// - `path`: The file path to write to. If the file exists, it will be truncated. - /// - `root_object`: A reference to the object to serialize. #[cfg(not(target_arch = "wasm32"))] pub fn save(&self, path: P, root_object: &T) -> Result<()> where @@ -476,7 +457,9 @@ impl ParcodeOptions { self.write_sync(file, root_object) } - /// PLACEHOLDER + /// Serializes the object graph to a generic writer synchronously. + /// + /// This method is equivalent to [`Self::write`] but avoids parallel execution. pub fn write_sync<'a, T, W>(&self, writer: W, root_object: &'a T) -> Result<()> where T: ParcodeVisitor, diff --git a/src/compression.rs b/src/compression.rs index fab4df3..d96d465 100644 --- a/src/compression.rs +++ b/src/compression.rs @@ -79,7 +79,6 @@ //! //! ```rust //! use parcode::compression::{CompressorRegistry, NoCompression}; -//! //! let mut registry = CompressorRegistry::new(); //! registry.register(Box::new(NoCompression)); //! ``` diff --git a/src/error.rs b/src/error.rs index e666833..00222ef 100644 --- a/src/error.rs +++ b/src/error.rs @@ -63,8 +63,9 @@ //! Ok(()) //! } //! # let state = GameState { level: 1 }; -//! # save_game_state(&state).unwrap(); +//! # save_game_state(&state)?; //! # std::fs::remove_file("game_err.par").ok(); +//! # Ok::<(), parcode::ParcodeError>(()) //! ``` //! //! ### Accessing Error Sources diff --git a/src/executor.rs b/src/executor.rs index 202e018..07641b4 100644 --- a/src/executor.rs +++ b/src/executor.rs @@ -14,8 +14,11 @@ use crate::graph::{Node, TaskGraph}; use crate::io::SeqWriter; use std::io::Write; -/// Context shared among all worker threads. -struct ExecutionContext<'a, 'graph, W: Write + Send> { +/// Context shared among all worker threads during parallel execution. +struct ExecutionContext<'a, 'graph, W> +where + W: Write + Send, +{ graph: &'graph TaskGraph<'a>, writer: &'graph SeqWriter, registry: &'graph CompressorRegistry, @@ -26,7 +29,10 @@ struct ExecutionContext<'a, 'graph, W: Write + Send> { } #[cfg(feature = "parallel")] -impl<'a, 'graph, W: Write + Send> ExecutionContext<'a, 'graph, W> { +impl<'a, 'graph, W> ExecutionContext<'a, 'graph, W> +where + W: Write + Send, +{ fn signal_error(&self, err: ParcodeError) { let mut guard = self.error_capture.lock().unwrap_or_else(|p| p.into_inner()); if guard.is_none() { @@ -53,12 +59,15 @@ impl<'a, 'graph, W: Write + Send> ExecutionContext<'a, 'graph, W> { /// # Type Parameters /// /// * `W`: The writer type. Must implement `std::io::Write` and `Send`. -pub fn execute_graph<'a, W: Write + Send>( +pub fn execute_graph<'a, W>( graph: &TaskGraph<'a>, writer: &SeqWriter, registry: &CompressorRegistry, use_compression: bool, -) -> Result { +) -> Result +where + W: Write + Send, +{ #[cfg(all(feature = "parallel", not(target_arch = "wasm32")))] { execute_graph_parallel(graph, writer, registry, use_compression) @@ -71,14 +80,17 @@ pub fn execute_graph<'a, W: Write + Send>( } // ------------ -/// PLACEHOLDER +/// Executes the graph in parallel using Rayon. #[cfg(all(feature = "parallel", not(target_arch = "wasm32")))] -fn execute_graph_parallel<'a, W: Write + Send>( +fn execute_graph_parallel<'a, W>( graph: &TaskGraph<'a>, writer: &SeqWriter, registry: &CompressorRegistry, use_compression: bool, -) -> Result { +) -> Result +where + W: Write + Send, +{ // 1. Setup the shared context. let ctx = ExecutionContext { graph, @@ -130,13 +142,16 @@ fn execute_graph_parallel<'a, W: Write + Send>( (*root_guard).ok_or_else(|| ParcodeError::Internal("Graph execution incomplete".into())) } -/// PLACEHOLDER -pub fn execute_graph_sync<'a, W: Write + Send>( +/// Executes the graph synchronously (single-threaded). +pub fn execute_graph_sync<'a, W>( graph: &TaskGraph<'a>, writer: &SeqWriter, registry: &CompressorRegistry, use_compression: bool, -) -> Result { +) -> Result +where + W: Write + Send, +{ // Buffer re-usable for compression. let mut shared_buffer = Vec::with_capacity(128 * 1024); let mut root_result: Option = None; @@ -177,7 +192,8 @@ pub fn execute_graph_sync<'a, W: Write + Send>( // Fallback to None (ID 0) if requested algorithm unavailable let compressor = registry .get(compression_id) - .unwrap_or_else(|_| registry.get(0).unwrap()); + .or_else(|_| registry.get(0)) + .map_err(|_| ParcodeError::Internal("Default compressor (ID 0) missing".into()))?; let footer_size = if is_chunkable { (children_refs.len() * ChildRef::SIZE) + 4 @@ -230,11 +246,13 @@ pub fn execute_graph_sync<'a, W: Write + Send>( /// The worker function executed by Rayon threads. /// It handles Serialization -> Compression -> Writing -> Notification. #[cfg(all(feature = "parallel", not(target_arch = "wasm32")))] -fn process_node<'scope, 'a, W: Write + Send>( +fn process_node<'scope, 'a, W>( scope: &rayon::Scope<'scope>, ctx: &'scope ExecutionContext<'a, 'scope, W>, node: &'scope Node<'a>, -) { +) where + W: Write + Send, +{ // 0. Fast abort check if ctx.should_abort() { return; diff --git a/src/format.rs b/src/format.rs index b0e3d50..f9c6bc9 100644 --- a/src/format.rs +++ b/src/format.rs @@ -177,18 +177,25 @@ impl ChildRef { if bytes.len() < Self::SIZE { return Err(ParcodeError::Format("Buffer too small for ChildRef".into())); } + + let offset_bytes = bytes.get(0..8).ok_or_else(|| { + ParcodeError::Format("Failed to read offset from ChildRef buffer".into()) + })?; + let length_bytes = bytes.get(8..16).ok_or_else(|| { + ParcodeError::Format("Failed to read length from ChildRef buffer".into()) + })?; + let offset = u64::from_le_bytes( - bytes - .get(0..8) - .and_then(|s| s.try_into().ok()) - .unwrap_or([0; 8]), + offset_bytes + .try_into() + .map_err(|_| ParcodeError::Format("Invalid offset bytes".into()))?, ); let length = u64::from_le_bytes( - bytes - .get(8..16) - .and_then(|s| s.try_into().ok()) - .unwrap_or([0; 8]), + length_bytes + .try_into() + .map_err(|_| ParcodeError::Format("Invalid length bytes".into()))?, ); + Ok(Self { offset, length }) } } diff --git a/src/graph/core.rs b/src/graph/core.rs index 37d23d0..e2c57fb 100644 --- a/src/graph/core.rs +++ b/src/graph/core.rs @@ -122,10 +122,14 @@ impl<'a> TaskGraph<'a> { } /// Retrieves a reference to a node by its ID. + /// + /// # Panics + /// + /// Panics if the `id` does not exist in the graph. pub fn get_node(&self, id: ChunkId) -> &Node<'a> { self.nodes .get(id.as_u32() as usize) - .expect("Node ID out of bounds") + .expect("TaskGraph invariant violated: Node ID out of bounds") } /// Returns true if the graph has no nodes. @@ -149,13 +153,15 @@ impl<'a> TaskGraph<'a> { /// dependencies (children) before the final job (consuming the data) can be fully constructed. /// /// # Panics - /// Panics if the `id` does not exist. + /// + /// Panics if the `id` does not exist in the graph. This is considered an internal + /// invariant violation. pub fn replace_job(&mut self, id: ChunkId, new_job: Box + 'a>) { let node_idx = id.as_u32() as usize; let node = self .nodes .get_mut(node_idx) - .expect("Node ID out of bounds during job replacement"); + .expect("TaskGraph invariant violated: Node ID not found during job replacement"); node.job = new_job; } } diff --git a/src/graph/job.rs b/src/graph/job.rs index 2155f60..e456911 100644 --- a/src/graph/job.rs +++ b/src/graph/job.rs @@ -16,17 +16,42 @@ pub struct JobConfig { /// Represents a unit of work: a piece of data that knows how to serialize itself. /// +/// This trait is the primary abstraction for the concurrent execution engine. Each node +/// in the [`TaskGraph`](crate::graph::TaskGraph) holds a `SerializationJob`. +/// /// # Lifetimes +/// /// * `'a`: The lifetime of the data being serialized. This allows the Job to hold /// references (e.g., `&'a [u8]`) instead of owning the data, enabling Zero-Copy writes. +/// +/// # Thread Safety +/// +/// Implementations must be `Send + Sync` to support parallel execution across threads. pub trait SerializationJob<'a>: Send + Sync { /// Executes the serialization logic, producing a raw byte buffer. + /// + /// This method is called by the executor when all dependencies of the node have + /// completed. + /// + /// ## Parameters + /// + /// * `children_refs`: A slice containing the offsets and lengths of all child chunks + /// in the final file. These can be used to build a footer or index for the chunk. + /// + /// ## Returns + /// + /// Returns a `Vec` containing the serialized payload. fn execute(&self, children_refs: &[ChildRef]) -> Result>; /// Returns an estimated size in bytes for scheduling heuristics. + /// + /// The executor uses this value to decide which jobs to run first or how to + /// distribute work across cores. An accurate estimate improves performance. fn estimated_size(&self) -> usize; /// Returns the specific configuration for this job. + /// + /// This includes settings like compression algorithm or sharding strategy. fn config(&self) -> JobConfig { JobConfig::default() } diff --git a/src/inspector.rs b/src/inspector.rs index 8c1ef55..1cc8daa 100644 --- a/src/inspector.rs +++ b/src/inspector.rs @@ -68,7 +68,10 @@ impl ParcodeInspector { } /// Convenience wrapper to inspect from path (matches old API if needed). - pub fn inspect>(path: P) -> Result { + pub fn inspect

(path: P) -> Result + where + P: AsRef, + { let file = ParcodeFile::open(path)?; Self::inspect_file(&file) } @@ -123,48 +126,41 @@ impl ParcodeInspector { repeat: u32, } - let slice = &payload.get(8..).expect("Missing Vec header"); - // We attempt to decode. If it fails, it's not a Vec header. - if let Ok((runs, _)) = bincode::serde::decode_from_slice::, _>( - slice, - bincode::config::standard(), - ) { - let total_items = if payload.len() >= 8 { - u64::from_le_bytes( - payload - .get(0..8) - .expect("Missing Vec header") - .try_into() - .unwrap_or([0; 8]), - ) - } else { - 0 - }; - - let distribution = format!( - "Vec<{}> items across {} logical shards", - total_items, - runs.iter().map(|r| r.repeat).sum::() - ); - return ("Vec Container".to_string(), Some(distribution)); + if let Some(slice) = payload.get(8..) { + // We attempt to decode. If it fails, it's not a Vec header. + if let Ok((runs, _)) = bincode::serde::decode_from_slice::, _>( + slice, + bincode::config::standard(), + ) { + let total_items = if let Some(header_bytes) = payload.get(0..8) { + u64::from_le_bytes(header_bytes.try_into().unwrap_or([0; 8])) + } else { + 0 + }; + + let distribution = format!( + "Vec<{}> items across {} logical shards", + total_items, + runs.iter().map(|r| r.repeat).sum::() + ); + return ("Vec Container".to_string(), Some(distribution)); + } } } // CHECK 2: Map Container (4 bytes exactly) - if payload.len() == 4 { - let num_shards = u32::from_le_bytes( - payload - .get(0..4) - .expect("Missing Map header") - .try_into() - .unwrap_or([0; 4]), + if payload.len() == 4 + && payload + .get(0..4) + .and_then(|b| b.try_into().ok()) + .map(u32::from_le_bytes) + == Some(node.child_count()) + { + let num_shards = node.child_count(); + return ( + "Map Container".to_string(), + Some(format!("Hashtable with {} buckets", num_shards)), ); - if num_shards == node.child_count() { - return ( - "Map Container".to_string(), - Some(format!("Hashtable with {} buckets", num_shards)), - ); - } } ("Generic Container".to_string(), None) diff --git a/src/io.rs b/src/io.rs index c67d58f..b08b8db 100644 --- a/src/io.rs +++ b/src/io.rs @@ -34,17 +34,26 @@ const WRITE_BUFFER_SIZE: usize = 16 * 1024 * 1024; /// - Network Streams (`TcpStream`) /// - Standard Output (`Stdout`) #[derive(Debug)] -pub struct SeqWriter { +pub struct SeqWriter +where + W: Write, +{ inner: Mutex>, } #[derive(Debug)] -struct WriterState { +struct WriterState +where + W: Write, +{ writer: BufWriter, current_offset: u64, } -impl SeqWriter { +impl SeqWriter +where + W: Write + Send, +{ /// Wraps any Writer (File, Vec, `TcpStream`) in a buffered sequential writer. pub fn new(writer: W) -> Self { Self { diff --git a/src/lib.rs b/src/lib.rs index 1df8844..16b8a05 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -123,11 +123,12 @@ //! stats: vec![10, 20, 30], //! }, //! }; -//! Parcode::save("game_lib.par", &state).unwrap(); +//! Parcode::save("game_lib.par", &state)?; //! //! // Load (eager) -//! let loaded: GameState = Parcode::load("game_lib.par").unwrap(); +//! let loaded: GameState = Parcode::load("game_lib.par")?; //! # std::fs::remove_file("game_lib.par").ok(); +//! # Ok::<(), parcode::ParcodeError>(()) //! ``` //! //! ### Lazy Loading @@ -159,10 +160,10 @@ //! stats: vec![10, 20, 30], //! }, //! }; -//! Parcode::save("game_lazy.par", &state).unwrap(); +//! Parcode::save("game_lazy.par", &state)?; //! -//! let file = Parcode::open("game_lazy.par").unwrap(); -//! let state_lazy = file.root::().unwrap(); +//! let file = Parcode::open("game_lazy.par")?; +//! let state_lazy = file.root::()?; //! //! // Access local fields instantly (already in memory) //! println!("Level: {}", state_lazy.level); @@ -170,6 +171,7 @@ //! // Load remote fields on-demand //! let player_name = state_lazy.player_data.name; //! # std::fs::remove_file("game_lazy.par").ok(); +//! # Ok::<(), parcode::ParcodeError>(()) //! ``` //! //! ### `HashMap` Sharding @@ -192,12 +194,13 @@ //! let mut users = HashMap::new(); //! users.insert(12345, User { name: "Alice".to_string() }); //! let db = Database { users }; -//! Parcode::save("db_map.par", &db).unwrap(); +//! Parcode::save("db_map.par", &db)?; //! -//! let file = Parcode::open("db_map.par").unwrap(); -//! let db_lazy = file.root::().unwrap(); +//! let file = Parcode::open("db_map.par")?; +//! let db_lazy = file.root::()?; //! let user = db_lazy.users.get(&12345u64).expect("User not found"); //! # std::fs::remove_file("db_map.par").ok(); +//! # Ok::<(), parcode::ParcodeError>(()) //! ``` //! //! ## Performance Considerations diff --git a/src/map.rs b/src/map.rs index 35ad3ce..bfec878 100644 --- a/src/map.rs +++ b/src/map.rs @@ -40,7 +40,10 @@ use twox_hash::XxHash64; /// /// # Returns /// A 64-bit hash value -pub(crate) fn hash_key(key: &K) -> u64 { +pub(crate) fn hash_key(key: &K) -> u64 +where + K: Hash, +{ let mut hasher = XxHash64::with_seed(0); key.hash(&mut hasher); hasher.finish() diff --git a/src/reader.rs b/src/reader.rs index 9b8775e..fc5e506 100644 --- a/src/reader.rs +++ b/src/reader.rs @@ -88,9 +88,10 @@ //! //! // Load entire object into memory //! let data = vec![1, 2, 3]; -//! Parcode::save("numbers_reader.par", &data).unwrap(); -//! let data: Vec = Parcode::load("numbers_reader.par").unwrap(); +//! Parcode::save("numbers_reader.par", &data)?; +//! let data: Vec = Parcode::load("numbers_reader.par")?; //! # std::fs::remove_file("numbers_reader.par").ok(); +//! # Ok::<(), parcode::ParcodeError>(()) //! ``` //! //! ### Lazy Loading (On-Demand) @@ -113,17 +114,18 @@ //! //! // Setup //! let state = GameState { level: 1, assets: Assets { data: vec![0; 10] } }; -//! Parcode::save("game_reader.par", &state).unwrap(); +//! Parcode::save("game_reader.par", &state)?; //! -//! let file = Parcode::open("game_reader.par").unwrap(); -//! let game_lazy = file.root::().unwrap(); +//! let file = Parcode::open("game_reader.par")?; +//! let game_lazy = file.root::()?; //! //! // Access local fields (instant, already in memory) //! println!("Level: {}", game_lazy.level); //! //! // Load remote fields on demand -//! let assets_data = game_lazy.assets.data.load().unwrap(); +//! let assets_data = game_lazy.assets.data.load()?; //! # std::fs::remove_file("game_reader.par").ok(); +//! # Ok::<(), parcode::ParcodeError>(()) //! ``` //! //! ### Random Access @@ -137,15 +139,16 @@ //! //! // Setup //! let data: Vec = (0..100).map(|i| MyStruct { val: i }).collect(); -//! Parcode::save("data_random.par", &data).unwrap(); +//! Parcode::save("data_random.par", &data)?; //! -//! let file = Parcode::open("data_random.par").unwrap(); -//! let root = file.root::>().unwrap(); +//! let file = Parcode::open("data_random.par")?; +//! let root = file.root::>()?; //! //! // Get item at index 50 without loading the entire vector //! // Note: Using 50 instead of 1,000,000 for a realistic small test -//! let item = root.get(50).unwrap(); +//! let item = root.get(50)?; //! # std::fs::remove_file("data_random.par").ok(); +//! # Ok::<(), parcode::ParcodeError>(()) //! ``` //! //! ### Streaming Iteration @@ -161,10 +164,10 @@ //! //! // Setup //! let data: Vec = (0..10).map(|i| MyStruct { val: i }).collect(); -//! Parcode::save("data_iter.par", &data).unwrap(); +//! Parcode::save("data_iter.par", &data)?; //! -//! let file = Parcode::open("data_iter.par").unwrap(); -//! let items: Vec = file.load().unwrap(); +//! let file = Parcode::open("data_iter.par")?; +//! let items: Vec = file.load()?; //! //! // Note: The current API doesn't have a direct `iter` on root for Vecs yet, //! // it usually goes through read_lazy or decode. @@ -173,6 +176,7 @@ //! process(item); //! } //! # std::fs::remove_file("data_iter.par").ok(); +//! # Ok::<(), parcode::ParcodeError>(()) //! ``` //! //! ## Performance Characteristics @@ -246,8 +250,8 @@ impl Deref for DataSource { fn deref(&self) -> &Self::Target { match self { #[cfg(all(feature = "mmap", not(target_arch = "wasm32")))] - DataSource::Mmap(mmap) => mmap.as_ref(), - DataSource::Memory(vec) => vec.as_slice(), + Self::Mmap(mmap) => mmap.as_ref(), + Self::Memory(vec) => vec.as_slice(), } } } @@ -278,15 +282,16 @@ impl Deref for DataSource { /// /// // Automatically selects parallel reconstruction for Vec /// let data = vec![1, 2, 3]; -/// Parcode::save("numbers_native.par", &data).unwrap(); -/// let data: Vec = Parcode::load("numbers_native.par").unwrap(); +/// let mut buffer = Vec::new(); +/// Parcode::write(&mut buffer, &data)?; +/// let data: Vec = Parcode::load_bytes(buffer)?; /// /// // Automatically selects sequential deserialization for primitives /// let val = 42; -/// Parcode::save("value_native.par", &val).unwrap(); -/// let value: i32 = Parcode::load("value_native.par").unwrap(); -/// # std::fs::remove_file("numbers_native.par").ok(); -/// # std::fs::remove_file("value_native.par").ok(); +/// let mut buffer = Vec::new(); +/// Parcode::write(&mut buffer, &val)?; +/// let value: i32 = Parcode::load_bytes(buffer)?; +/// # Ok::<(), parcode::ParcodeError>(()) /// ``` pub trait ParcodeNative: Sized { /// Reconstructs the object from the given graph node. @@ -649,10 +654,13 @@ impl ParcodeFile { return Err(ParcodeError::Format("File smaller than header".into())); } - let header_start = (file_size - GLOBAL_HEADER_SIZE as u64) as usize; - let header_bytes = &source[header_start..]; + let header_start = usize::try_from(file_size - GLOBAL_HEADER_SIZE as u64) + .map_err(|_| ParcodeError::Format("File size too large for usize".into()))?; + let header_bytes = source + .get(header_start..) + .ok_or_else(|| ParcodeError::Format("Failed to access global header".into()))?; - if header_bytes[0..4] != MAGIC_BYTES { + if header_bytes.get(0..4) != Some(&MAGIC_BYTES) { return Err(ParcodeError::Format("Invalid Magic Bytes".into())); } @@ -784,7 +792,11 @@ impl ParcodeFile { let chunk_end = usize::try_from(offset + length) .map_err(|_| ParcodeError::Format("Chunk end exceeds address space".into()))?; - let meta = MetaByte::from_byte(self.source[chunk_end - 1]); + let meta_byte = self + .source + .get(chunk_end - 1) + .ok_or_else(|| ParcodeError::Format("Failed to read chunk meta byte".into()))?; + let meta = MetaByte::from_byte(*meta_byte); let mut child_count = 0; let mut payload_end = chunk_end - 1; @@ -795,7 +807,10 @@ impl ParcodeFile { } let count_start = chunk_end - 5; - let count_bytes = &self.source[count_start..count_start + 4]; + let count_bytes = self + .source + .get(count_start..count_start + 4) + .ok_or_else(|| ParcodeError::Format("Failed to read child count".into()))?; child_count = Self::read_u32(count_bytes)?; let footer_size = child_count as usize * ChildRef::SIZE; @@ -859,7 +874,11 @@ impl<'a> ChunkNode<'a> { .map_err(|_| ParcodeError::Format("End offset exceeds address space".into()))?; // Access via self.reader.source (Deref to &[u8]) - let raw = &self.reader.source[start..end]; + let raw = self + .reader + .source + .get(start..end) + .ok_or_else(|| ParcodeError::Format("Payload out of bounds".into()))?; let method_id = self.meta.compression_method(); self.reader.registry.get(method_id)?.decompress(raw) @@ -1159,7 +1178,11 @@ impl<'a> ChunkNode<'a> { .map_err(|_| ParcodeError::Format("Offset exceeds usize range".into()))?; let entry_start = footer_start + (index * ChildRef::SIZE); - let bytes = &self.reader.source[entry_start..entry_start + ChildRef::SIZE]; + let bytes = self + .reader + .source + .get(entry_start..entry_start + ChildRef::SIZE) + .ok_or_else(|| ParcodeError::Format("Child reference out of bounds".into()))?; let r = ChildRef::from_bytes(bytes)?; self.reader.get_chunk(r.offset, r.length) diff --git a/src/rt.rs b/src/rt.rs index 8603eb2..bee9070 100644 --- a/src/rt.rs +++ b/src/rt.rs @@ -15,13 +15,19 @@ use std::vec::IntoIter; /// Wrapper that injects configuration into an existing Job. #[derive(Debug)] -pub struct ConfiguredJob<'a, J: ?Sized> { +pub struct ConfiguredJob<'a, J> +where + J: ?Sized, +{ config: JobConfig, inner: Box, _marker: PhantomData<&'a ()>, } -impl<'a, J: SerializationJob<'a> + ?Sized> ConfiguredJob<'a, J> { +impl<'a, J> ConfiguredJob<'a, J> +where + J: SerializationJob<'a> + ?Sized, +{ pub fn new(inner: Box, config: JobConfig) -> Self { Self { inner, @@ -31,7 +37,10 @@ impl<'a, J: SerializationJob<'a> + ?Sized> ConfiguredJob<'a, J> { } } -impl<'a, J: SerializationJob<'a> + ?Sized> SerializationJob<'a> for ConfiguredJob<'a, J> { +impl<'a, J> SerializationJob<'a> for ConfiguredJob<'a, J> +where + J: SerializationJob<'a> + ?Sized, +{ fn execute(&self, children_refs: &[ChildRef]) -> Result> { self.inner.execute(children_refs) } @@ -75,7 +84,10 @@ pub struct ParcodePromise<'a, T> { _m: PhantomData, } -impl<'a, T: DeserializeOwned> ParcodePromise<'a, T> { +impl<'a, T> ParcodePromise<'a, T> +where + T: DeserializeOwned, +{ /// Internal constructor. pub fn new(node: ChunkNode<'a>) -> Self { Self { @@ -100,7 +112,10 @@ pub struct ParcodeCollectionPromise<'a, T> { _m: PhantomData, } -impl<'a, T: ParcodeItem + Send + Sync + 'a> ParcodeCollectionPromise<'a, T> { +impl<'a, T> ParcodeCollectionPromise<'a, T> +where + T: ParcodeItem + Send + Sync + 'a, +{ /// Internal constructor. pub fn new(node: ChunkNode<'a>) -> Self { Self { @@ -589,7 +604,10 @@ where /// This iterator efficiently traverses the shards of a vector, maintaining /// a cursor position to avoid re-parsing previous items. #[derive(Debug)] -pub struct ParcodeLazyIterator<'a, T: ParcodeLazyRef<'a>> { +pub struct ParcodeLazyIterator<'a, T> +where + T: ParcodeLazyRef<'a>, +{ /// The shards (children of the container node). shards: std::vec::IntoIter>, @@ -608,7 +626,10 @@ pub struct ParcodeLazyIterator<'a, T: ParcodeLazyRef<'a>> { _marker: PhantomData, } -impl<'a, T: ParcodeLazyRef<'a>> ParcodeLazyIterator<'a, T> { +impl<'a, T> ParcodeLazyIterator<'a, T> +where + T: ParcodeLazyRef<'a>, +{ pub fn new(node: ChunkNode<'a>) -> Result { let total_items = usize::try_from(node.len()).unwrap_or(usize::MAX); let shards = node.children()?.into_iter(); @@ -658,7 +679,10 @@ impl<'a, T: ParcodeLazyRef<'a>> ParcodeLazyIterator<'a, T> { } } -impl<'a, T: ParcodeLazyRef<'a>> Iterator for ParcodeLazyIterator<'a, T> { +impl<'a, T> Iterator for ParcodeLazyIterator<'a, T> +where + T: ParcodeLazyRef<'a>, +{ type Item = Result; fn next(&mut self) -> Option { @@ -695,7 +719,10 @@ impl<'a, T: ParcodeLazyRef<'a>> Iterator for ParcodeLazyIterator<'a, T> { } } -impl<'a, T: ParcodeLazyRef<'a>> ExactSizeIterator for ParcodeLazyIterator<'a, T> { +impl<'a, T> ExactSizeIterator for ParcodeLazyIterator<'a, T> +where + T: ParcodeLazyRef<'a>, +{ fn len(&self) -> usize { self.total_items - self.items_yielded } diff --git a/src/visitor.rs b/src/visitor.rs index 0c8a163..0c17422 100644 --- a/src/visitor.rs +++ b/src/visitor.rs @@ -100,7 +100,7 @@ //! struct MyStructJob<'a> { data: &'a MyStruct } //! impl<'a> SerializationJob<'a> for MyStructJob<'a> { //! fn execute(&self, _: &[ChildRef]) -> Result> { -//! Ok(parcode::internal::bincode::serde::encode_to_vec(&self.data.local_field, parcode::internal::bincode::config::standard()).unwrap()) +//! Ok(parcode::internal::bincode::serde::encode_to_vec(&self.data.local_field, parcode::internal::bincode::config::standard()).map_err(|e| parcode::ParcodeError::Serialization(e.to_string()))?) //! } //! fn estimated_size(&self) -> usize { 4 } //! } diff --git a/src/visitor_impls.rs b/src/visitor_impls.rs index 47bd215..02eb58a 100644 --- a/src/visitor_impls.rs +++ b/src/visitor_impls.rs @@ -309,7 +309,6 @@ impl<'a> SerializationJob<'a> for MapContainerJob { impl ParcodeVisitor for HashMap where - K: Serialize + DeserializeOwned + Hash + Eq + Send + Sync + Clone + 'static, K: Serialize + DeserializeOwned + Hash + Eq + Send + Sync + Clone + 'static, V: Serialize + DeserializeOwned + Send + Sync + Clone + ParcodeVisitor + 'static, { @@ -439,7 +438,10 @@ where } } -impl ParcodeVisitor for &T { +impl ParcodeVisitor for &T +where + T: ParcodeVisitor, +{ fn visit<'a>( &'a self, graph: &mut TaskGraph<'a>, From fe33b09155e9dc9107a53cac15ffc16517a720d4 Mon Sep 17 00:00:00 2001 From: Retype15 Date: Tue, 30 Dec 2025 17:54:38 -0500 Subject: [PATCH 4/4] feat: Implement features "mmap", "parallel" and "common"(Default) - **async:** Implement rayon and make loading and saving multithreaded. - **mmap:** Includes mmap and uses it in open as in <=v0.5.3: `unsafe { Mmap::map(&file)? };`. - **common:** Enables the two previous features as one. Added alternative methods to use a buffer directly (write, write_sync). - Changed: writer_to_write -> write - Update a majority of documentation to use this new API. Update axes task runner configuration, refine gitignore, and introduce comprehensive API integration tests. --- .axes/axes.toml | 32 +++++----- .gitignore | 42 +++++-------- tests/api_test.rs | 155 ++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 189 insertions(+), 40 deletions(-) create mode 100644 tests/api_test.rs diff --git a/.axes/axes.toml b/.axes/axes.toml index 7d453a5..5f07d5f 100644 --- a/.axes/axes.toml +++ b/.axes/axes.toml @@ -1,14 +1,27 @@ name = "parcode" -version = "0.5.3" +version = "0.6.0" description = "Blazing fast parallel caching for Rust. Serializes complex data to a chunked file format to get CPU maximum power on data loads, but as a simple binary loader." +[vars] +_cargo_all = " " +_cargo_features = "" +_cargo_targets = "" +_cargo_all_feat_targets = "" +GREETING = "Hello from an axes variable!" + +[scripts] +fmt = "cargo fmt -- --check" +clippy = "cargo clippy -- -D warnings" +test = "cargo test " +build = "cargo build " + [scripts.check] run = [ "# === Running all standard quality tests... ===", - "cargo build ", - "> cargo fmt -- --check", - "> cargo clippy -- -D warnings", - "> cargo test ", + "", + "> ", + "> ", + "> ", "# <#green>=== Completed! now you can make a PR. ===<#reset>", ] desc = "Run a standard quality test required from make a PR." @@ -22,12 +35,3 @@ desc = "Commands to run when exiting a session (e.g., `docker-compose down`)" run = "# echo 'Exiting session...'" [options.open_with] - -[vars] -_cargo_all = " " -_cargo_features = "" -_cargo_targets = "" -_cargo_all_feat_targets = "" -GREETING = "Hello from an axes variable!" - -[env] diff --git a/.gitignore b/.gitignore index 140afd3..cfbae7d 100644 --- a/.gitignore +++ b/.gitignore @@ -1,27 +1,17 @@ -# Generated by Cargo -# will have compiled files and executables -debug -target +# Not included +/* +parcode-derive/target +examples/p_test.rs -# These are backup files generated by rustfmt -**/*.rs.bk - -# MSVC Windows builds of rustc generate these, which store debugging information -*.pdb - -# Generated by cargo mutants -# Contains mutation testing data -**/mutants.out*/ - -# RustRover -# JetBrains specific template is maintained in a separate JetBrains.gitignore that can -# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore -# and can be added to the global gitignore or merged into this file. For a more nuclear -# option (not recommended) you can uncomment the following to ignore the entire idea folder. -#.idea/ -Cargo.lock -parcode.txt -road.md -.gitignore -CONTRIBUTING.md -internal_todo.md +# Included +!.axes +!benches +!examples +!parcode-derive +!src +!tests +!Cargo.toml +!LICENSE +!README.md +!TODO.md +!whitepaper.md \ No newline at end of file diff --git a/tests/api_test.rs b/tests/api_test.rs new file mode 100644 index 0000000..3990181 --- /dev/null +++ b/tests/api_test.rs @@ -0,0 +1,155 @@ +#![allow(missing_docs)] + +use parcode::{Parcode, ParcodeObject}; +use serde::{Deserialize, Serialize}; + +#[derive(Serialize, Deserialize, ParcodeObject, PartialEq, Debug, Clone)] +struct SimpleData { + id: u32, + message: String, +} + +#[derive(Serialize, Deserialize, ParcodeObject, PartialEq, Debug, Clone)] +struct ComplexData { + title: String, + #[parcode(chunkable)] + numbers: Vec, + #[parcode(chunkable)] + inner: SimpleData, +} + +// Generator of data +fn create_complex_data() -> ComplexData { + ComplexData { + title: "Integration Test".to_string(), + numbers: (0..50_000).collect(), + inner: SimpleData { + id: 42, + message: "Hello World".to_string(), + }, + } +} + +// --- TESTS --- + +/// Standard File IO +/// Validate `Parcode::save`, `Parcode::load`, `execute_graph` (Default) +#[test] +#[cfg(not(target_arch = "wasm32"))] +fn test_standard_file_io() -> parcode::Result<()> { + let dir = tempfile::tempdir()?; + let file_path = dir.path().join("std_io.par"); + let data = create_complex_data(); + + Parcode::save(&file_path, &data)?; + + let loaded: ComplexData = Parcode::load(&file_path)?; + + assert_eq!(data, loaded); + Ok(()) +} + +/// Pure Memory IO +/// Validate `Parcode::write`, `Parcode::load_bytes`, `DataSource::Memory` +#[test] +fn test_memory_io() -> parcode::Result<()> { + let data = create_complex_data(); + let mut buffer = Vec::new(); + + Parcode::write(&mut buffer, &data)?; + + assert!(!buffer.is_empty()); + + let loaded: ComplexData = Parcode::load_bytes(buffer)?; + + assert_eq!(data, loaded); + Ok(()) +} + +/// Explicit Synchronous Write (Forced) +/// Validate `Parcode::write_sync`, `execute_graph_sync` +#[test] +fn test_explicit_sync_write() -> parcode::Result<()> { + let data = create_complex_data(); + let mut buffer = Vec::new(); + + Parcode::write_sync(&mut buffer, &data)?; + + let loaded: ComplexData = Parcode::load_bytes(buffer)?; + + assert_eq!(data, loaded); + Ok(()) +} + +/// Explicit Synchronous Write to File +/// Validate `Parcode::save_sync` +#[test] +#[cfg(not(target_arch = "wasm32"))] +fn test_explicit_sync_save_file() -> parcode::Result<()> { + let dir = tempfile::tempdir()?; + let file_path = dir.path().join("sync_save.par"); + let data = SimpleData { + id: 1, + message: "Sync".into(), + }; + + Parcode::save_sync(&file_path, &data)?; + let loaded: SimpleData = Parcode::load(&file_path)?; + + assert_eq!(data, loaded); + Ok(()) +} + +/// Lazy Loading from Memory +/// Validate `Parcode::open_bytes`, `DataSource::Memory` with Lazy +#[test] +fn test_lazy_load_memory() -> parcode::Result<()> { + let data = create_complex_data(); + let bytes = Parcode::serialize(&data)?; + + let file = Parcode::open_bytes(bytes)?; + + let root_mirror = file.root::()?; + + assert_eq!(root_mirror.title, "Integration Test"); + + let item_100 = root_mirror.numbers.get(100)?; + assert_eq!(item_100, 100); + + Ok(()) +} + +/// Compression +/// Validate `ParcodeOptions`, `CompressorRegistry` +#[test] +fn test_compression_config() -> parcode::Result<()> { + let data = create_complex_data(); + let mut buffer_compressed = Vec::new(); + let mut buffer_raw = Vec::new(); + + Parcode::builder() + .compression(true) + .write(&mut buffer_compressed, &data)?; + + Parcode::write(&mut buffer_raw, &data)?; + + let loaded: ComplexData = Parcode::load_bytes(buffer_compressed)?; + assert_eq!(data, loaded); + + Ok(()) +} + +/// Inspector +/// Validate `ParcodeInspector`, read metadata without deserializing +#[test] +fn test_inspector() -> parcode::Result<()> { + let data = create_complex_data(); + let bytes = Parcode::serialize(&data)?; + + let report = Parcode::inspect_bytes(bytes)?; + + assert_eq!(report.global_version, 4); + assert!(report.root_offset > 0); + + Ok(()) +}