diff --git a/Cargo.lock b/Cargo.lock index d4ef4c95..ba0affde 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -162,6 +162,16 @@ dependencies = [ "cfg-if 1.0.0", ] +[[package]] +name = "crossbeam-channel" +version = "0.4.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b153fe7cbef478c567df0f972e02e6d736db11affe43dfc9c56a9374d1adfb87" +dependencies = [ + "crossbeam-utils 0.7.2", + "maybe-uninit", +] + [[package]] name = "crossbeam-channel" version = "0.5.0" @@ -169,7 +179,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dca26ee1f8d361640700bde38b2c37d8c22b3ce2d360e1fc1c74ea4b0aa7d775" dependencies = [ "cfg-if 1.0.0", - "crossbeam-utils", + "crossbeam-utils 0.8.0", ] [[package]] @@ -180,7 +190,7 @@ checksum = "94af6efb46fef72616855b036a624cf27ba656ffc9be1b9a3c931cfc7749a9a9" dependencies = [ "cfg-if 1.0.0", "crossbeam-epoch", - "crossbeam-utils", + "crossbeam-utils 0.8.0", ] [[package]] @@ -191,12 +201,23 @@ checksum = "ec0f606a85340376eef0d6d8fec399e6d4a544d648386c6645eb6d0653b27d9f" dependencies = [ "cfg-if 1.0.0", "const_fn", - "crossbeam-utils", + "crossbeam-utils 0.8.0", "lazy_static", "memoffset", "scopeguard", ] +[[package]] +name = "crossbeam-utils" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3c7c73a2d1e9fc0886a08b93e98eb643461230d5f1925e4036204d5f2e261a8" +dependencies = [ + "autocfg", + "cfg-if 0.1.10", + "lazy_static", +] + [[package]] name = "crossbeam-utils" version = "0.8.0" @@ -320,6 +341,12 @@ dependencies = [ "cfg-if 0.1.10", ] +[[package]] +name = "maybe-uninit" +version = "2.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "60302e4db3a61da70c0cb7991976248362f30319e88850c487b9b95bbf059e00" + [[package]] name = "memoffset" version = "0.5.6" @@ -408,6 +435,7 @@ dependencies = [ "clap", "cloudflare-zlib", "crc", + "crossbeam-channel 0.4.4", "image", "indexmap", "itertools", @@ -461,9 +489,9 @@ version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9ab346ac5921dc62ffa9f89b7a773907511cdfa5490c572ae9be1be33e8afa4a" dependencies = [ - "crossbeam-channel", + "crossbeam-channel 0.5.0", "crossbeam-deque", - "crossbeam-utils", + "crossbeam-utils 0.8.0", "lazy_static", "num_cpus", ] diff --git a/Cargo.toml b/Cargo.toml index 634b4e9e..3a3b7cbc 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -40,6 +40,7 @@ indexmap = "1.6.0" libdeflater = { version = "0.5.0", optional = true } log = "0.4.11" stderrlog = { version = "0.5.0", optional = true } +crossbeam-channel = "0.4.4" [dependencies.rayon] optional = true diff --git a/src/evaluate.rs b/src/evaluate.rs index 254a0928..904e492c 100644 --- a/src/evaluate.rs +++ b/src/evaluate.rs @@ -12,16 +12,14 @@ use crate::png::STD_WINDOW; #[cfg(not(feature = "parallel"))] use crate::rayon; use crate::Deadline; +#[cfg(feature = "parallel")] +use crossbeam_channel::{bounded, unbounded, Receiver, Sender}; use rayon::prelude::*; #[cfg(not(feature = "parallel"))] use std::cell::RefCell; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering::SeqCst; -#[cfg(feature = "parallel")] -use std::sync::mpsc::*; use std::sync::Arc; -#[cfg(feature = "parallel")] -use std::thread; struct Candidate { image: PngData, @@ -49,19 +47,34 @@ pub(crate) struct Evaluator { best_candidate_size: Arc, /// images are sent to the thread for evaluation #[cfg(feature = "parallel")] - eval_send: SyncSender, + eval_send: Sender, // the thread helps evaluate images asynchronously #[cfg(feature = "parallel")] - eval_thread: thread::JoinHandle>, + eval_thread: Receiver>, // in non-parallel mode, images are evaluated synchronously #[cfg(not(feature = "parallel"))] eval_best_candidate: RefCell>, } +// Like `std::thread::spawn`, but uses Rayon to create a thread. +// +// This allows this function to work on targets like WebAssembly, +// where `std::thread::spawn` doesn't work but Rayon can thanks to its +// support for custom spawn handlers. +// +// Unlike `std::thread::spawn`, `rayon::spawn` doesn't support closure results +// (it doesn't return a `JoinHandle`), so we simulate it with a crossbeam +// channel instead (where `Receiver` acts as a `JoinHandle`). +fn thread_spawn(f: impl FnOnce() -> T + Send + 'static) -> Receiver { + let (tx, rx) = bounded(0); + rayon::spawn(move || tx.send(f()).unwrap()); + rx +} + impl Evaluator { pub fn new(deadline: Arc) -> Self { #[cfg(feature = "parallel")] - let (tx, rx) = sync_channel(4); + let (tx, rx) = unbounded(); Self { deadline, best_candidate_size: Arc::new(AtomicMin::new(None)), @@ -69,7 +82,7 @@ impl Evaluator { #[cfg(feature = "parallel")] eval_send: tx, #[cfg(feature = "parallel")] - eval_thread: thread::spawn(move || rx.into_iter().min_by_key(Candidate::cmp_key)), + eval_thread: thread_spawn(move || rx.into_iter().min_by_key(Candidate::cmp_key)), #[cfg(not(feature = "parallel"))] eval_best_candidate: RefCell::new(None), } @@ -80,7 +93,7 @@ impl Evaluator { #[cfg(feature = "parallel")] fn get_best_candidate(self) -> Option { drop(self.eval_send); // disconnect the sender, breaking the loop in the thread - self.eval_thread.join().expect("eval thread") + self.eval_thread.recv().expect("eval thread") } #[cfg(not(feature = "parallel"))] diff --git a/tests/flags.rs b/tests/flags.rs index 40023be0..fba5eb38 100644 --- a/tests/flags.rs +++ b/tests/flags.rs @@ -55,9 +55,9 @@ fn test_it_converts( #[test] fn verbose_mode() { + use crossbeam_channel::{unbounded, Sender}; use log::{set_logger, set_max_level, Level, LevelFilter, Log, Metadata, Record}; use std::cell::RefCell; - use std::sync::mpsc::{sync_channel, SyncSender}; // Rust runs tests in parallel by default. // We want to make sure that we verify only logs from our test. @@ -66,7 +66,7 @@ fn verbose_mode() { // initialise it with Some(sender) only on threads spawned within // our test. thread_local! { - static VERBOSE_LOGS: RefCell>> = RefCell::new(None); + static VERBOSE_LOGS: RefCell>> = RefCell::new(None); } struct LogTester; @@ -97,7 +97,7 @@ fn verbose_mode() { let input = PathBuf::from("tests/files/verbose_mode.png"); let (output, opts) = get_opts(&input); - let (sender, receiver) = sync_channel(4); + let (sender, receiver) = unbounded(); let thread_init = move || { // Initialise logs storage for all threads within our test.