Support WebAssembly and asm.js

This adds support for both WebAssembly and asm.js by abstracting the
worker thread into an ImmediateWorker and a ThreadWorker which uses the
ImmediateWorker. The ThreadWorker is exposed as the worker to be used on
every platform other than WebAssembly and asm.js. There the
ImmediateWorker itself is exposed instead, because threading isn't
supported on those platforms just yet.

Resolves #75

Author: CryZe

src/decoder.rs

@@ -10,8 +10,7 @@ use std::io::Read;
 use std::mem;
 use std::ops::Range;
 use std::sync::Arc;
-use std::sync::mpsc::{self, Sender};
-use worker_thread::{RowData, spawn_worker_thread, WorkerMsg};
+use worker::{RowData, PlatformWorker, Worker};
 
 pub const MAX_COMPONENTS: usize = 4;
 
@@ -133,7 +132,7 @@ impl<R: Read> Decoder<R> {
 
         let mut previous_marker = Marker::SOI;
         let mut pending_marker = None;
-        let mut worker_chan = None;
+        let mut worker = None;
         let mut scans_processed = 0;
         let mut planes = vec![Vec::new(); self.frame.as_ref().map_or(0, |frame| frame.components.len())];
 
@@ -193,8 +192,8 @@ impl<R: Read> Decoder<R> {
                     if self.frame.is_none() {
                         return Err(Error::Format("scan encountered before frame".to_owned()));
                     }
-                    if worker_chan.is_none() {
-                        worker_chan = Some(spawn_worker_thread()?);
+                    if worker.is_none() {
+                        worker = Some(PlatformWorker::new()?);
                     }
 
                     let frame = self.frame.clone().unwrap();
@@ -216,7 +215,7 @@ impl<R: Read> Decoder<R> {
                     }
 
                     let is_final_scan = scan.component_indices.iter().all(|&i| self.coefficients_finished[i] == !0);
-                    let (marker, data) = self.decode_scan(&frame, &scan, worker_chan.as_ref().unwrap(), is_final_scan)?;
+                    let (marker, data) = self.decode_scan(&frame, &scan, worker.as_mut().unwrap(), is_final_scan)?;
 
                     if let Some(data) = data {
                         for (i, plane) in data.into_iter().enumerate().filter(|&(_, ref plane)| !plane.is_empty()) {
@@ -357,7 +356,7 @@ impl<R: Read> Decoder<R> {
     fn decode_scan(&mut self,
                    frame: &FrameInfo,
                    scan: &ScanInfo,
-                   worker_chan: &Sender<WorkerMsg>,
+                   worker: &mut PlatformWorker,
                    produce_data: bool)
                    -> Result<(Option<Marker>, Option<Vec<Vec<u8>>>)> {
         assert!(scan.component_indices.len() <= MAX_COMPONENTS);
@@ -394,7 +393,7 @@ impl<R: Read> Decoder<R> {
                     quantization_table: self.quantization_tables[component.quantization_table_index].clone().unwrap(),
                 };
 
-                worker_chan.send(WorkerMsg::Start(row_data))?;
+                worker.start(row_data)?;
             }
         }
 
@@ -516,7 +515,7 @@ impl<R: Read> Decoder<R> {
                         mem::replace(&mut mcu_row_coefficients[i], vec![0i16; coefficients_per_mcu_row])
                     };
 
-                    worker_chan.send(WorkerMsg::AppendRow((i, row_coefficients)))?;
+                    worker.append_row((i, row_coefficients))?;
                 }
             }
         }
@@ -528,10 +527,7 @@ impl<R: Read> Decoder<R> {
             let mut data = vec![Vec::new(); frame.components.len()];
 
             for (i, &component_index) in scan.component_indices.iter().enumerate() {
-                let (tx, rx) = mpsc::channel();
-                worker_chan.send(WorkerMsg::GetResult((i, tx)))?;
-
-                data[component_index] = rx.recv()?;
+                data[component_index] = worker.get_result(i)?;
             }
 
             Ok((marker, Some(data)))

src/lib.rs

@@ -42,4 +42,4 @@ mod idct;
 mod marker;
 mod parser;
 mod upsampler;
-mod worker_thread;
+mod worker;

src/worker/immediate.rs

@@ -0,0 +1,74 @@
+use decoder::MAX_COMPONENTS;
+use error::Result;
+use idct::dequantize_and_idct_block;
+use std::mem;
+use std::sync::Arc;
+use parser::Component;
+use super::{RowData, Worker};
+
+pub struct ImmediateWorker {
+    offsets: [usize; MAX_COMPONENTS],
+    results: Vec<Vec<u8>>,
+    components: Vec<Option<Component>>,
+    quantization_tables: Vec<Option<Arc<[u16; 64]>>>,
+}
+
+impl ImmediateWorker {
+    pub fn new_immediate() -> ImmediateWorker {
+        ImmediateWorker {
+            offsets: [0; MAX_COMPONENTS],
+            results: vec![Vec::new(); MAX_COMPONENTS],
+            components: vec![None; MAX_COMPONENTS],
+            quantization_tables: vec![None; MAX_COMPONENTS],
+        }
+    }
+    pub fn start_immediate(&mut self, data: RowData) {
+        assert!(self.results[data.index].is_empty());
+
+        self.offsets[data.index] = 0;
+        self.results[data.index].resize(data.component.block_size.width as usize * data.component.block_size.height as usize * 64, 0u8);
+        self.components[data.index] = Some(data.component);
+        self.quantization_tables[data.index] = Some(data.quantization_table);
+    }
+    pub fn append_row_immediate(&mut self, (index, data): (usize, Vec<i16>)) {
+        // Convert coefficients from a MCU row to samples.
+
+        let component = self.components[index].as_ref().unwrap();
+        let quantization_table = self.quantization_tables[index].as_ref().unwrap();
+        let block_count = component.block_size.width as usize * component.vertical_sampling_factor as usize;
+        let line_stride = component.block_size.width as usize * 8;
+
+        assert_eq!(data.len(), block_count * 64);
+
+        for i in 0..block_count {
+            let x = (i % component.block_size.width as usize) * 8;
+            let y = (i / component.block_size.width as usize) * 8;
+            dequantize_and_idct_block(&data[i * 64..(i + 1) * 64],
+                                    quantization_table,
+                                    line_stride,
+                                    &mut self.results[index][self.offsets[index] + y * line_stride + x..]);
+        }
+
+        self.offsets[index] += data.len();
+    }
+    pub fn get_result_immediate(&mut self, index: usize) -> Vec<u8> {
+        mem::replace(&mut self.results[index], Vec::new())
+    }
+}
+
+impl Worker for ImmediateWorker {
+    fn new() -> Result<Self> {
+        Ok(ImmediateWorker::new_immediate())
+    }
+    fn start(&mut self, data: RowData) -> Result<()> {
+        self.start_immediate(data);
+        Ok(())
+    }
+    fn append_row(&mut self, row: (usize, Vec<i16>)) -> Result<()> {
+        self.append_row_immediate(row);
+        Ok(())
+    }
+    fn get_result(&mut self, index: usize) -> Result<Vec<u8>> {
+        Ok(self.get_result_immediate(index))
+    }
+}

src/worker/mod.rs

@@ -0,0 +1,24 @@
+mod threaded;
+mod immediate;
+
+#[cfg(not(any(target_arch = "asmjs", target_arch = "wasm32")))]
+pub use self::threaded::ThreadedWorker as PlatformWorker;
+#[cfg(any(target_arch = "asmjs", target_arch = "wasm32"))]
+pub use self::immediate::ImmediateWorker as PlatformWorker;
+
+use error::Result;
+use parser::Component;
+use std::sync::Arc;
+
+pub struct RowData {
+    pub index: usize,
+    pub component: Component,
+    pub quantization_table: Arc<[u16; 64]>,
+}
+
+pub trait Worker: Sized {
+    fn new() -> Result<Self>;
+    fn start(&mut self, row_data: RowData) -> Result<()>;
+    fn append_row(&mut self, row: (usize, Vec<i16>)) -> Result<()>;
+    fn get_result(&mut self, index: usize) -> Result<Vec<u8>>;
+}

src/worker/threaded.rs

@@ -0,0 +1,53 @@
+use error::Result;
+use std::sync::mpsc::{self, Sender};
+use std::thread;
+use super::{RowData, Worker};
+use super::immediate::ImmediateWorker;
+
+enum WorkerMsg {
+    Start(RowData),
+    AppendRow((usize, Vec<i16>)),
+    GetResult((usize, Sender<Vec<u8>>)),
+}
+
+pub struct ThreadedWorker {
+    sender: Sender<WorkerMsg>,
+}
+
+impl Worker for ThreadedWorker {
+    fn new() -> Result<Self> {
+        let thread_builder = thread::Builder::new().name("worker thread".to_owned());
+        let (tx, rx) = mpsc::channel();
+
+        thread_builder.spawn(move || {
+            let mut worker = ImmediateWorker::new_immediate();
+
+            while let Ok(message) = rx.recv() {
+                match message {
+                    WorkerMsg::Start(data) => {
+                        worker.start_immediate(data);
+                    },
+                    WorkerMsg::AppendRow(row) => {
+                        worker.append_row_immediate(row);
+                    },
+                    WorkerMsg::GetResult((index, chan)) => {
+                        let _ = chan.send(worker.get_result_immediate(index));
+                    },
+                }
+            }
+        })?;
+
+        Ok(ThreadedWorker { sender: tx })
+    }
+    fn start(&mut self, row_data: RowData) -> Result<()> {
+        Ok(self.sender.send(WorkerMsg::Start(row_data))?)
+    }
+    fn append_row(&mut self, row: (usize, Vec<i16>)) -> Result<()> {
+        Ok(self.sender.send(WorkerMsg::AppendRow(row))?)
+    }
+    fn get_result(&mut self, index: usize) -> Result<Vec<u8>> {
+        let (tx, rx) = mpsc::channel();
+        self.sender.send(WorkerMsg::GetResult((index, tx)))?;
+        Ok(rx.recv()?)
+    }
+}