Merge pull request #241 from oligamiq/master

add rayon feature

Author: johannesvollmer

.github/workflows/rust.yml

@@ -16,10 +16,14 @@ jobs:
     - uses: actions/checkout@v2
     - name: Cache Cargo Dependencies
       uses: Swatinem/[email protected]
-    - name: Build
-      run: cargo build --verbose
-    - name: Run tests
-      run: cargo test --verbose
+    - name: Build without default features
+      run: cargo build --no-default-features --verbose
+    - name: Run tests without default features
+      run: cargo test --no-default-features --verbose
+    - name: Build with rayon feature
+      run: cargo build --features rayon --verbose
+    - name: Run tests with rayon feature
+      run: cargo test --features rayon --verbose
 
 
   macos:
@@ -31,10 +35,14 @@ jobs:
       - uses: actions/checkout@v2
       - name: Cache Cargo Dependencies
         uses: Swatinem/[email protected]
-      - name: Build
-        run: cargo build --verbose
-      - name: Run tests
-        run: cargo test --verbose
+      - name: Build without default features
+        run: cargo build --no-default-features --verbose
+      - name: Run tests without default features
+        run: cargo test --no-default-features --verbose
+      - name: Build with rayon feature
+        run: cargo build --features rayon --verbose
+      - name: Run tests with rayon feature
+        run: cargo test --features rayon --verbose
 
   verify-msrv:
     strategy:
@@ -84,12 +92,18 @@ jobs:
       - name: Start Docker
         run: sudo systemctl start docker
 
-      - name: Cross-Compile project to powerpc-unknown-linux-gnu
-        run: cross build --target powerpc-unknown-linux-gnu --verbose
+      - name: Cross-Compile project to powerpc-unknown-linux-gnu without rayon feature
+        run: cross build --target powerpc-unknown-linux-gnu --no-default-features --verbose
 
       # https://github.com/cross-rs/cross#supported-targets
-      - name: Cross-Run Tests in powerpc-unknown-linux-gnu using Qemu
-        run: cross test --target powerpc-unknown-linux-gnu --verbose
+      - name: Cross-Run Tests in powerpc-unknown-linux-gnu using Qemu without rayon feature
+        run: cross test --target powerpc-unknown-linux-gnu --no-default-features --verbose
+
+      - name: Cross-Compile project to powerpc-unknown-linux-gnu with rayon feature
+        run: cross build --target powerpc-unknown-linux-gnu --features rayon --verbose
+
+      - name: Cross-Run Tests in powerpc-unknown-linux-gnu with rayon feature using Qemu
+        run: cross test --target powerpc-unknown-linux-gnu --features rayon --verbose
 
   wasm32:
     runs-on: ubuntu-latest

Cargo.toml

@@ -32,7 +32,7 @@ half = "2.1.0"                 # 16 bit float pixel data type
 bit_field = "^0.10.1"          # exr file version bit flags
 miniz_oxide = "^0.8.0"         # zip compression for pxr24
 smallvec = "^1.7.0"            # make cache-friendly allocations        TODO profile if smallvec is really an improvement!
-rayon-core = "^1.11.0"         # threading for parallel compression     TODO make this an optional feature?
+rayon-core = { version = "^1.11.0", optional = true }         # threading for parallel compression
 zune-inflate = { version = "^0.2.3", default-features = false, features = ["zlib"] }  # zip decompression, faster than miniz_oxide
 
 [dev-dependencies]
@@ -43,23 +43,39 @@ walkdir = "2.3.2"         # automatically test things for all files in a directo
 rand = "0.8.5"            # used for fuzz testing
 rayon = "1.5.3"           # run tests for many files in parallel
 
+[features]
+default = ["rayon"]
+
+# rayon is used for parallel compression
+rayon = ["dep:rayon-core"]
+
+[[example]]
+name = "7_write_raw_blocks"
+required-features = ["rayon"]
+
+[[example]]
+name = "8_read_raw_blocks"
+required-features = ["rayon"]
 
 [[bench]]
 name = "read"
 harness = false
+required-features = ["rayon"]
 
 [[bench]]
 name = "profiling"
 harness = false
+required-features = ["rayon"]
 
 [[bench]]
 name = "write"
 harness = false
+required-features = ["rayon"]
 
 [[bench]]
 name = "pixel_format_conversion"
 harness = false
-
+required-features = ["rayon"]
 
 # recommended release settings for max runtime performance
 [profile.release]

src/block/reader.rs

@@ -5,6 +5,7 @@ use std::convert::TryFrom;
 use std::fmt::Debug;
 use std::io::{Read, Seek};
 use std::sync::mpsc;
+#[cfg(feature = "rayon")]
 use rayon_core::{ThreadPool, ThreadPoolBuildError};
 
 use smallvec::alloc::sync::Arc;
@@ -212,6 +213,7 @@ pub trait ChunksReader: Sized + Iterator<Item=Result<Chunk>> + ExactSizeIterator
         OnProgressChunksReader { chunks_reader: self, callback: on_progress, decoded_chunks: 0 }
     }
 
+    #[cfg(feature = "rayon")]
     /// Decompress all blocks in the file, using multiple cpu cores, and call the supplied closure for each block.
     /// The order of the blocks is not deterministic.
     /// You can also use `parallel_decompressor` to obtain an iterator instead.
@@ -235,6 +237,7 @@ pub trait ChunksReader: Sized + Iterator<Item=Result<Chunk>> + ExactSizeIterator
         Ok(())
     }
 
+    #[cfg(feature = "rayon")]
     /// Return an iterator that decompresses the chunks with multiple threads.
     /// The order of the blocks is not deterministic.
     /// Use `ParallelBlockDecompressor::new` if you want to use your own thread pool.
@@ -380,6 +383,7 @@ impl<R: ChunksReader> SequentialBlockDecompressor<R> {
     }
 }
 
+#[cfg(feature = "rayon")]
 /// Decompress the chunks in a file in parallel.
 /// The first call to `next` will fill the thread pool with jobs,
 /// starting to decompress the next few blocks.
@@ -399,6 +403,7 @@ pub struct ParallelBlockDecompressor<R: ChunksReader> {
     pool: ThreadPool,
 }
 
+#[cfg(feature = "rayon")]
 impl<R: ChunksReader> ParallelBlockDecompressor<R> {
 
     /// Create a new decompressor. Does not immediately spawn any tasks.
@@ -512,7 +517,9 @@ impl<R: ChunksReader> Iterator for SequentialBlockDecompressor<R> {
     fn size_hint(&self) -> (usize, Option<usize>) { self.remaining_chunks_reader.size_hint() }
 }
 
+#[cfg(feature = "rayon")]
 impl<R: ChunksReader> ExactSizeIterator for ParallelBlockDecompressor<R> {}
+#[cfg(feature = "rayon")]
 impl<R: ChunksReader> Iterator for ParallelBlockDecompressor<R> {
     type Item = Result<UncompressedBlock>;
     fn next(&mut self) -> Option<Self::Item> { self.decompress_next_block() }

src/block/writer.rs

@@ -6,12 +6,14 @@ use std::io::Seek;
 use std::iter::Peekable;
 use std::ops::Not;
 use std::sync::mpsc;
+#[cfg(feature = "rayon")]
 use rayon_core::{ThreadPool, ThreadPoolBuildError};
 
 use smallvec::alloc::collections::BTreeMap;
 
 use crate::block::UncompressedBlock;
 use crate::block::chunk::Chunk;
+#[cfg(feature = "rayon")]
 use crate::compression::Compression;
 use crate::error::{Error, Result, UnitResult, usize_to_u64};
 use crate::io::{Data, Tracking, Write};
@@ -80,6 +82,7 @@ pub trait ChunksWriter: Sized {
         SequentialBlocksCompressor::new(meta, self)
     }
 
+    #[cfg(feature = "rayon")]
     /// Obtain a new writer that can compress blocks to chunks on multiple threads, which are then passed to this writer.
     /// Returns none if the sequential compressor should be used instead (thread pool creation failure or too large performance overhead).
     fn parallel_blocks_compressor<'w>(&'w mut self, meta: &'w MetaData) -> Option<ParallelBlocksCompressor<'w, Self>> {
@@ -101,6 +104,7 @@ pub trait ChunksWriter: Sized {
         Ok(())
     }
 
+    #[cfg(feature = "rayon")]
     /// Compresses all blocks to the file.
     /// The index of the block must be in increasing line order within the header.
     /// Obtain iterator with `MetaData::collect_ordered_blocks(...)` or similar methods.
@@ -332,6 +336,7 @@ impl<'w, W> SequentialBlocksCompressor<'w, W> where W: 'w + ChunksWriter {
 }
 
 /// Compress blocks to a chunk writer with multiple threads.
+#[cfg(feature = "rayon")]
 #[derive(Debug)]
 #[must_use]
 pub struct ParallelBlocksCompressor<'w, W> {
@@ -348,6 +353,7 @@ pub struct ParallelBlocksCompressor<'w, W> {
     next_incoming_chunk_index: usize, // used to remember original chunk order
 }
 
+#[cfg(feature = "rayon")]
 impl<'w, W> ParallelBlocksCompressor<'w, W> where W: 'w + ChunksWriter {
 
     /// New blocks writer. Returns none if sequential compression should be used.

src/image/read/image.rs

@@ -29,7 +29,11 @@ impl<F, L> ReadImage<F, L> where F: FnMut(f64)
     pub fn new(read_layers: L, on_progress: F) -> Self {
         Self {
             on_progress, read_layers,
-            pedantic: false, parallel: true,
+            pedantic: false,
+            #[cfg(not(feature = "rayon"))]
+            parallel: false,
+            #[cfg(feature = "rayon")]
+            parallel: true,
         }
     }
 
@@ -120,6 +124,10 @@ impl<F, L> ReadImage<F, L> where F: FnMut(f64)
 
         // TODO propagate send requirement further upwards
         if parallel {
+            #[cfg(not(feature = "rayon"))]
+            return Err(crate::error::Error::unsupported("parallel decompression requires the rayon feature"));
+
+            #[cfg(feature = "rayon")]
             block_reader.decompress_parallel(pedantic, |meta_data, block|{
                 image_collector.read_block(&meta_data.headers, block)
             })?;

src/image/write/mod.rs

@@ -73,7 +73,13 @@ impl<'img, WritableLayers> WritableImage<'img, WritableLayers> for &'img Image<W
         WriteImageWithOptions {
             image: self,
             check_compatibility: true,
+
+            #[cfg(not(feature = "rayon"))]
+            parallel: false,
+
+            #[cfg(feature = "rayon")]
             parallel: true,
+
             on_progress: ignore_progress
         }
     }
@@ -164,7 +170,13 @@ impl<'img, L, F> WriteImageWithOptions<'img, L, F>
                 );
 
                 let chunk_writer = chunk_writer.on_progress(self.on_progress);
-                if self.parallel { chunk_writer.compress_all_blocks_parallel(&meta, blocks)?; }
+                if self.parallel {
+                    #[cfg(not(feature = "rayon"))]
+                    return Err(crate::error::Error::unsupported("parallel compression requires the rayon feature"));
+
+                    #[cfg(feature = "rayon")]
+                    chunk_writer.compress_all_blocks_parallel(&meta, blocks)?;
+                }
                 else { chunk_writer.compress_all_blocks_sequential(&meta, blocks)?; }
                 /*let blocks_writer = chunk_writer.as_blocks_writer(&meta);