Port schunk_simple.c example, impl Send for Schunk & typesize in constructors

.github/workflows/CI.yml

@@ -125,3 +125,12 @@ jobs:
         run: |
           cargo clean  # ensure we're starting fresh, no funny business
           cargo test --no-default-features --features static -vv --release
+
+      - name: Run examples
+        shell: bash
+        run: |
+          for example in $(find examples -name '*.rs'); do
+            example_name=$(basename $example .rs)
+            echo "------- Running example: ${example_name} -------"
+            cargo run --example $example_name
+          done

examples/schunk_simple.rs

@@ -0,0 +1,83 @@
+const KB: usize = 1024;
+const MB: usize = 1024 * KB;
+
+const CHUNKSIZE: usize = 1_000 * 1_000;
+const NCHUNKS: usize = 100;
+const NTHREADS: usize = 4;
+
+fn main() {
+    blosc2::init();
+
+    let mut data = vec![0_i32; CHUNKSIZE];
+    let mut data_dest = vec![0_i32; CHUNKSIZE];
+
+    println!(
+        "Blosc2 version info: {} ({})",
+        blosc2::get_version_string().unwrap(),
+        String::from_utf8(blosc2::BLOSC2_VERSION_DATE.to_vec()).unwrap()
+    );
+
+    // Create a super-chunk container
+    let cparams = blosc2::CParams::default()
+        .set_typesize::<i32>()
+        .set_clevel(blosc2::CLevel::Nine)
+        .set_nthreads(NTHREADS);
+    let dparams = blosc2::DParams::default().set_nthreads(NTHREADS);
+    let storage = blosc2::schunk::Storage::default()
+        .set_cparams(cparams)
+        .set_dparams(dparams);
+    let mut schunk = blosc2::schunk::SChunk::new(storage);
+
+    let ttotal = std::time::Instant::now();
+    for nchunk in 0..NCHUNKS {
+        for i in 0..CHUNKSIZE {
+            data[i] = (i * nchunk) as i32;
+        }
+        let nchunks = schunk.append_buffer(&data).unwrap();
+        if nchunks != nchunk + 1 {
+            panic!(
+                "Unexpected nchunks! Got {}, but expected {}",
+                nchunks,
+                nchunk + 1
+            );
+        }
+    }
+
+    println!(
+        "Compression ratio: {:.1} MB -> {:.1} MB ({:.1}x)",
+        schunk.nbytes() as f32 / MB as f32,
+        schunk.cbytes() as f32 / MB as f32,
+        schunk.compression_ratio()
+    );
+    println!(
+        "Compression time: {:.3}s, {:.1}MB/s",
+        ttotal.elapsed().as_secs_f32(),
+        schunk.nbytes() as f32 / (ttotal.elapsed().as_secs_f32() * MB as f32)
+    );
+
+    // Retrieve and decompress the chunks (0-based count)
+    let ttotal = std::time::Instant::now();
+    for nchunk in (0..(NCHUNKS - 1)).rev() {
+        // The error check in c-blosc2 example is removed b/c will have been raised thru .unwrap call
+        let dsize = schunk.decompress_chunk(nchunk, &mut data_dest).unwrap();
+        assert_eq!(dsize, CHUNKSIZE);
+    }
+    println!(
+        "Decompression time: {:.3}s, {:.1}MB/s",
+        ttotal.elapsed().as_secs_f32(),
+        schunk.nbytes() as f32 / (ttotal.elapsed().as_secs_f32() * MB as f32)
+    );
+
+    // Check integrity of the second chunk (made of non-zeros)
+    schunk.decompress_chunk(1, &mut data_dest).unwrap();
+    for i in 0..CHUNKSIZE {
+        if data_dest[i] != i as i32 {
+            panic!(
+                "Decompressed data differs from original {}, {}",
+                i as i32, data_dest[i]
+            );
+        }
+    }
+
+    blosc2::destroy();
+}

src/lib.rs

@@ -6,6 +6,8 @@ use std::ffi::{c_void, CStr, CString};
 use std::sync::Arc;
 use std::{io, mem};
 
+pub use blosc2_sys::BLOSC2_VERSION_DATE;
+
 /// Result type used in this library
 pub type Result<T> = std::result::Result<T, Error>;
 
@@ -563,8 +565,9 @@ pub mod schunk {
         /// assert_eq!(chunk.info().unwrap().nbytes(), 10);
         /// ```
         pub fn uninit<T>(cparams: CParams, len: usize) -> Result<Self> {
+            let mut cparams = cparams;
             if mem::size_of::<T>() != cparams.0.typesize as usize {
-                return Err("typesize mismatch between CParams and T".into());
+                cparams.0.typesize = mem::size_of::<T>() as _;
             }
             let mut dst = Vec::with_capacity(
                 (len * cparams.0.typesize as usize) + ffi::BLOSC_EXTENDED_HEADER_LENGTH as usize,
@@ -597,8 +600,9 @@ pub mod schunk {
         /// assert_eq!(chunk.decompress::<f32>().unwrap(), vec![0.123_f32, 0.123, 0.123, 0.123]);
         /// ```
         pub fn repeatval<T>(cparams: CParams, value: T, len: usize) -> Result<Self> {
+            let mut cparams = cparams;
             if mem::size_of::<T>() != cparams.0.typesize as usize {
-                return Err("typesize mismatch between CParams and T".into());
+                cparams.0.typesize = mem::size_of::<T>() as _;
             }
             let mut dst = Vec::with_capacity(
                 (len * cparams.0.typesize as usize) + ffi::BLOSC_EXTENDED_HEADER_LENGTH as usize,
@@ -632,8 +636,9 @@ pub mod schunk {
         /// assert_eq!(chunk.info().unwrap().nbytes(), 40);  // 10 elements * 4 bytes each
         /// ```
         pub fn zeros<T>(cparams: CParams, len: usize) -> Result<Self> {
+            let mut cparams = cparams;
             if mem::size_of::<T>() != cparams.0.typesize as usize {
-                return Err("typesize mismatch between CParams and T".into());
+                cparams.0.typesize = mem::size_of::<T>() as _;
             }
             let mut dst = Vec::with_capacity(
                 (len * cparams.0.typesize as usize) + ffi::BLOSC_EXTENDED_HEADER_LENGTH as usize,
@@ -787,6 +792,8 @@ pub mod schunk {
     #[derive(Clone)]
     pub struct SChunk(pub(crate) Arc<RwLock<*mut ffi::blosc2_schunk>>);
 
+    unsafe impl Send for SChunk {}
+
     // Loosely inspired by blosc2-python implementation
     impl SChunk {
         pub fn new(storage: Storage) -> Self {
@@ -844,16 +851,16 @@ pub mod schunk {
             Ok(nchunks as _)
         }
 
-        /// Decompress a chunk, returning number of bytes written to output buffer
+        /// Decompress a chunk, returning number of elements of `T` written to output buffer
         #[inline]
         pub fn decompress_chunk<T>(&mut self, nchunk: usize, dst: &mut [T]) -> Result<usize> {
             let chunk = Chunk::from_schunk(self, nchunk)?;
             let info = chunk.info()?;
-            if dst.len() < info.nbytes as usize {
+            if dst.len() * mem::size_of::<T>() < info.nbytes as usize {
                 let msg = format!(
-                    "Not large enough, need {} but got {}",
+                    "Not large enough, need {} bytes but got buffer w/ {} bytes of storage",
                     info.nbytes,
-                    dst.len()
+                    dst.len() * mem::size_of::<T>()
                 );
                 return Err(msg.into());
             }
@@ -874,7 +881,7 @@ pub mod schunk {
                 let msg = format!("Non-initialized error decompressing chunk '{}'", nchunk);
                 return Err(msg.into());
             } else {
-                Ok(size as _)
+                Ok((size / mem::size_of::<T>() as i32) as _)
             }
         }
 
@@ -1229,8 +1236,7 @@ impl CParams {
 impl Default for CParams {
     #[inline]
     fn default() -> Self {
-        let mut cparams = unsafe { ffi::blosc2_get_blosc2_cparams_defaults() };
-        cparams.typesize = 1;
+        let cparams = unsafe { ffi::blosc2_get_blosc2_cparams_defaults() };
         Self(cparams)
     }
 }
@@ -1895,6 +1901,38 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_schunk_thread_shared() -> Result<()> {
+        let input = b"some data";
+        let storage = schunk::Storage::default()
+            .set_contiguous(true)
+            .set_cparams(CParams::from(&input[0]))
+            .set_dparams(DParams::default());
+        let mut schunk = schunk::SChunk::new(storage);
+
+        schunk.append_buffer(input)?;
+
+        let mut schunk2 = schunk.clone();
+        std::thread::spawn(move || {
+            assert_eq!(schunk2.n_chunks(), 1);
+            schunk2.append_buffer(b"more data").unwrap();
+        })
+        .join()
+        .unwrap();
+
+        assert_eq!(schunk.n_chunks(), 2);
+        assert_eq!(
+            b"some data",
+            schunk.decompress_chunk_vec(0).unwrap().as_slice()
+        );
+        assert_eq!(
+            b"more data",
+            schunk.decompress_chunk_vec(1).unwrap().as_slice()
+        );
+
+        Ok(())
+    }
+
     #[cfg(not(target_os = "windows"))]
     #[test]
     fn test_schunk_write() -> Result<()> {