Merge pull request #240 from gents83/master

Formatting and isolating rayon feature completely

Author: HeroicKatora

src/decoder.rs

@@ -1,13 +1,3 @@
-use alloc::borrow::ToOwned;
-use alloc::sync::Arc;
-use alloc::vec::Vec;
-use alloc::{format, vec};
-use core::cmp;
-use core::mem;
-use core::ops::Range;
-use std::convert::TryInto;
-use std::io::Read;
-use crate::read_u8;
 use crate::error::{Error, Result, UnsupportedFeature};
 use crate::huffman::{fill_default_mjpeg_tables, HuffmanDecoder, HuffmanTable};
 use crate::marker::Marker;
@@ -16,8 +6,18 @@ use crate::parser::{
     AdobeColorTransform, AppData, CodingProcess, Component, Dimensions, EntropyCoding, FrameInfo,
     IccChunk, ScanInfo,
 };
+use crate::read_u8;
 use crate::upsampler::Upsampler;
-use crate::worker::{PreferWorkerKind, RowData, Worker, with_worker};
+use crate::worker::{compute_image_parallel, with_worker, PreferWorkerKind, RowData, Worker};
+use alloc::borrow::ToOwned;
+use alloc::sync::Arc;
+use alloc::vec::Vec;
+use alloc::{format, vec};
+use core::cmp;
+use core::mem;
+use core::ops::Range;
+use std::convert::TryInto;
+use std::io::Read;
 
 pub const MAX_COMPONENTS: usize = 4;
 
@@ -202,7 +202,8 @@ impl<R: Read> Decoder<R> {
     pub fn read_info(&mut self) -> Result<()> {
         with_worker(PreferWorkerKind::Multithreaded, |worker| {
             self.decode_internal(true, worker)
-        }).map(|_| ())
+        })
+        .map(|_| ())
     }
 
     /// Configure the decoder to scale the image during decoding.
@@ -395,8 +396,7 @@ impl<R: Read> Decoder<R> {
                             }
                         }
 
-                        let (marker, data) =
-                            self.decode_scan(&frame, &scan, worker, &finished)?;
+                        let (marker, data) = self.decode_scan(&frame, &scan, worker, &finished)?;
 
                         if let Some(data) = data {
                             for (i, plane) in data
@@ -542,12 +542,16 @@ impl<R: Read> Decoder<R> {
         let frame = self.frame.as_ref().unwrap();
 
         if {
-            let required_mem = frame.components.len()
+            let required_mem = frame
+                .components
+                .len()
                 .checked_mul(frame.output_size.width.into())
                 .and_then(|m| m.checked_mul(frame.output_size.height.into()));
             required_mem.map_or(true, |m| self.decoding_buffer_size_limit < m)
         } {
-            return Err(Error::Format("size of decoded image exceeds maximum allowed size".to_owned()));
+            return Err(Error::Format(
+                "size of decoded image exceeds maximum allowed size".to_owned(),
+            ));
         }
 
         // If we're decoding a progressive jpeg and a component is unfinished, render what we've got
@@ -579,12 +583,12 @@ impl<R: Read> Decoder<R> {
                     * usize::from(component.vertical_sampling_factor)
                     * 64;
 
-                let mut tasks = (0..frame.mcu_size.height)
-                    .map(|mcu_y| {
-                        let offset = usize::from(mcu_y) * coefficients_per_mcu_row;
-                        let row_coefficients = self.coefficients[i][offset..offset + coefficients_per_mcu_row].to_vec();
-                        (i, row_coefficients)
-                    });
+                let mut tasks = (0..frame.mcu_size.height).map(|mcu_y| {
+                    let offset = usize::from(mcu_y) * coefficients_per_mcu_row;
+                    let row_coefficients =
+                        self.coefficients[i][offset..offset + coefficients_per_mcu_row].to_vec();
+                    (i, row_coefficients)
+                });
 
                 // FIXME: additional potential work stealing opportunities for rayon case if we
                 // also internally can parallelize over components.
@@ -825,7 +829,9 @@ impl<R: Read> Decoder<R> {
                                 &mut mcu_row_coefficients[i][block_offset..block_offset + 64]
                             } else {
                                 &mut dummy_block[..64]
-                            }.try_into().unwrap();
+                            }
+                            .try_into()
+                            .unwrap();
 
                             if scan.successive_approximation_high == 0 {
                                 decode_block(
@@ -1174,65 +1180,7 @@ fn compute_image(
     }
 }
 
-#[cfg(feature = "rayon")]
-fn compute_image_parallel(
-    components: &[Component],
-    data: Vec<Vec<u8>>,
-    output_size: Dimensions,
-    is_jfif: bool,
-    color_transform: Option<AdobeColorTransform>,
-) -> Result<Vec<u8>> {
-    use rayon::prelude::*;
-
-    let color_convert_func = choose_color_convert_func(components.len(), is_jfif, color_transform)?;
-    let upsampler = Upsampler::new(components, output_size.width, output_size.height)?;
-    let line_size = output_size.width as usize * components.len();
-    let mut image = vec![0u8; line_size * output_size.height as usize];
-
-    image
-        .par_chunks_mut(line_size)
-        .with_max_len(1)
-        .enumerate()
-        .for_each(|(row, line)| {
-            upsampler.upsample_and_interleave_row(
-                &data,
-                row,
-                output_size.width as usize,
-                line,
-                color_convert_func,
-            );
-        });
-
-    Ok(image)
-}
-
-#[cfg(not(feature = "rayon"))]
-fn compute_image_parallel(
-    components: &[Component],
-    data: Vec<Vec<u8>>,
-    output_size: Dimensions,
-    is_jfif: bool,
-    color_transform: Option<AdobeColorTransform>,
-) -> Result<Vec<u8>> {
-    let color_convert_func = choose_color_convert_func(components.len(), is_jfif, color_transform)?;
-    let upsampler = Upsampler::new(components, output_size.width, output_size.height)?;
-    let line_size = output_size.width as usize * components.len();
-    let mut image = vec![0u8; line_size * output_size.height as usize];
-
-    for (row, line) in image.chunks_mut(line_size).enumerate() {
-        upsampler.upsample_and_interleave_row(
-            &data,
-            row,
-            output_size.width as usize,
-            line,
-            color_convert_func,
-        );
-    }
-
-    Ok(image)
-}
-
-fn choose_color_convert_func(
+pub(crate) fn choose_color_convert_func(
     component_count: usize,
     _is_jfif: bool,
     color_transform: Option<AdobeColorTransform>,
@@ -1255,7 +1203,7 @@ fn choose_color_convert_func(
                 None => {
                     // Assume CMYK because no APP14 marker was found
                     Ok(color_convert_line_cmyk)
-                },
+                }
             }
         }
         _ => panic!(),

src/worker/mod.rs

@@ -1,12 +1,17 @@
 mod immediate;
 mod multithreaded;
-#[cfg(feature = "rayon")]
+#[cfg(all(
+    not(any(target_arch = "asmjs", target_arch = "wasm32")),
+    feature = "rayon"
+))]
 mod rayon;
 
+use crate::decoder::choose_color_convert_func;
+use crate::error::Result;
+use crate::parser::{AdobeColorTransform, Component, Dimensions};
+use crate::upsampler::Upsampler;
 use alloc::sync::Arc;
 use alloc::vec::Vec;
-use crate::error::Result;
-use crate::parser::Component;
 
 pub struct RowData {
     pub index: usize,
@@ -19,30 +24,66 @@ pub trait Worker {
     fn append_row(&mut self, row: (usize, Vec<i16>)) -> Result<()>;
     fn get_result(&mut self, index: usize) -> Result<Vec<u8>>;
     /// Default implementation for spawning multiple tasks.
-    fn append_rows(&mut self, row: &mut dyn Iterator<Item=(usize, Vec<i16>)>)
-        -> Result<()>
-    {
+    fn append_rows(&mut self, row: &mut dyn Iterator<Item = (usize, Vec<i16>)>) -> Result<()> {
         for item in row {
             self.append_row(item)?;
         }
         Ok(())
     }
 }
 
+#[allow(dead_code)]
 pub enum PreferWorkerKind {
     Immediate,
     Multithreaded,
 }
 
-
 /// Execute something with a worker system.
 pub fn with_worker<T>(prefer: PreferWorkerKind, f: impl FnOnce(&mut dyn Worker) -> T) -> T {
     match prefer {
-        #[cfg(all(not(any(target_arch = "asmjs", target_arch = "wasm32")), feature = "rayon"))]
+        #[cfg(all(
+            not(any(target_arch = "asmjs", target_arch = "wasm32")),
+            feature = "rayon"
+        ))]
         PreferWorkerKind::Multithreaded => self::rayon::with_rayon(f),
+        #[allow(unreachable_patterns)]
         #[cfg(not(any(target_arch = "asmjs", target_arch = "wasm32")))]
         PreferWorkerKind::Multithreaded => self::multithreaded::with_multithreading(f),
         _ => self::immediate::with_immediate(f),
     }
 }
 
+pub fn compute_image_parallel(
+    components: &[Component],
+    data: Vec<Vec<u8>>,
+    output_size: Dimensions,
+    is_jfif: bool,
+    color_transform: Option<AdobeColorTransform>,
+) -> Result<Vec<u8>> {
+    #[cfg(all(
+        not(any(target_arch = "asmjs", target_arch = "wasm32")),
+        feature = "rayon"
+    ))]
+    return rayon::compute_image_parallel(components, data, output_size, is_jfif, color_transform);
+
+    #[allow(unreachable_code)]
+    {
+        let color_convert_func =
+            choose_color_convert_func(components.len(), is_jfif, color_transform)?;
+        let upsampler = Upsampler::new(components, output_size.width, output_size.height)?;
+        let line_size = output_size.width as usize * components.len();
+        let mut image = vec![0u8; line_size * output_size.height as usize];
+
+        for (row, line) in image.chunks_mut(line_size).enumerate() {
+            upsampler.upsample_and_interleave_row(
+                &data,
+                row,
+                output_size.width as usize,
+                line,
+                color_convert_func,
+            );
+        }
+
+        Ok(image)
+    }
+}

src/worker/multithreaded.rs

@@ -4,12 +4,16 @@
 //! and allow scaling to more cores.
 //! However, that would be more complex, so we use this as a starting point.
 
-use std::{mem, sync::mpsc::{self, Receiver, Sender}};
+use super::immediate::ImmediateWorker;
+use super::{RowData, Worker};
 use crate::decoder::MAX_COMPONENTS;
 use crate::error::Result;
-use super::{RowData, Worker};
-use super::immediate::ImmediateWorker;
+use std::{
+    mem,
+    sync::mpsc::{self, Receiver, Sender},
+};
 
+#[allow(dead_code)]
 pub fn with_multithreading<T>(f: impl FnOnce(&mut dyn Worker) -> T) -> T {
     let mut worker = MpscWorker::default();
     f(&mut worker)
@@ -23,7 +27,7 @@ enum WorkerMsg {
 
 #[derive(Default)]
 pub struct MpscWorker {
-    senders: [Option<Sender<WorkerMsg>>; MAX_COMPONENTS]
+    senders: [Option<Sender<WorkerMsg>>; MAX_COMPONENTS],
 }
 
 impl MpscWorker {
@@ -43,14 +47,18 @@ impl MpscWorker {
         // and in all other message-passing methods because there's not that many rows
         // and this should be cheaper than spawning MAX_COMPONENTS many threads up front
         let sender = self.senders[component].as_mut().unwrap();
-        sender.send(WorkerMsg::Start(row_data)).expect("jpeg-decoder worker thread error");
+        sender
+            .send(WorkerMsg::Start(row_data))
+            .expect("jpeg-decoder worker thread error");
         Ok(())
     }
 
     fn append_row(&mut self, row: (usize, Vec<i16>)) -> Result<()> {
         let component = row.0;
         let sender = self.senders[component].as_mut().unwrap();
-        sender.send(WorkerMsg::AppendRow(row.1)).expect("jpeg-decoder worker thread error");
+        sender
+            .send(WorkerMsg::AppendRow(row.1))
+            .expect("jpeg-decoder worker thread error");
         Ok(())
     }
 
@@ -61,7 +69,9 @@ impl MpscWorker {
     ) -> Result<Vec<u8>> {
         let (tx, rx) = mpsc::channel();
         let sender = mem::take(&mut self.senders[index]).unwrap();
-        sender.send(WorkerMsg::GetResult(tx)).expect("jpeg-decoder worker thread error");
+        sender
+            .send(WorkerMsg::GetResult(tx))
+            .expect("jpeg-decoder worker thread error");
         Ok(collect(rx))
     }
 }
@@ -91,14 +101,14 @@ fn create_worker() -> (Sender<WorkerMsg>, impl FnOnce() + 'static) {
                     // to attempt to access nonexistent components
                     data.index = 0;
                     worker.start_immediate(data);
-                },
+                }
                 WorkerMsg::AppendRow(row) => {
                     worker.append_row_immediate((0, row));
-                },
+                }
                 WorkerMsg::GetResult(chan) => {
                     let _ = chan.send(worker.get_result_immediate(0));
                     break;
-                },
+                }
             }
         }
     };