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..0e07d308 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::{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, @@ -47,12 +45,9 @@ pub(crate) struct Evaluator { deadline: Arc, nth: AtomicUsize, best_candidate_size: Arc, - /// images are sent to the thread for evaluation + /// images are sent to the caller thread for evaluation #[cfg(feature = "parallel")] - eval_send: SyncSender, - // the thread helps evaluate images asynchronously - #[cfg(feature = "parallel")] - eval_thread: thread::JoinHandle>, + eval_channel: (Sender, Receiver), // in non-parallel mode, images are evaluated synchronously #[cfg(not(feature = "parallel"))] eval_best_candidate: RefCell>, @@ -61,15 +56,13 @@ pub(crate) struct Evaluator { impl Evaluator { pub fn new(deadline: Arc) -> Self { #[cfg(feature = "parallel")] - let (tx, rx) = sync_channel(4); + let eval_channel = unbounded(); Self { deadline, best_candidate_size: Arc::new(AtomicMin::new(None)), nth: AtomicUsize::new(0), #[cfg(feature = "parallel")] - eval_send: tx, - #[cfg(feature = "parallel")] - eval_thread: thread::spawn(move || rx.into_iter().min_by_key(Candidate::cmp_key)), + eval_channel, #[cfg(not(feature = "parallel"))] eval_best_candidate: RefCell::new(None), } @@ -79,8 +72,9 @@ impl Evaluator { /// Or `None` if all reductions were worse than baseline. #[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") + let (eval_send, eval_recv) = self.eval_channel; + drop(eval_send); // disconnect the sender, breaking the loop in the thread + eval_recv.into_iter().min_by_key(Candidate::cmp_key) } #[cfg(not(feature = "parallel"))] @@ -110,7 +104,7 @@ impl Evaluator { // sends it off asynchronously for compression, // but results will be collected via the message queue #[cfg(feature = "parallel")] - let eval_send = self.eval_send.clone(); + let eval_send = self.eval_channel.0.clone(); rayon::spawn(move || { let filters_iter = STD_FILTERS.par_iter().with_max_len(1); diff --git a/tests/flags.rs b/tests/flags.rs index 40023be0..69d79f6a 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. @@ -118,6 +118,7 @@ fn verbose_mode() { #[cfg(feature = "rayon")] rayon::ThreadPoolBuilder::new() .start_handler(move |_| thread_init()) + .num_threads(rayon::current_num_threads() + 1) .build() .unwrap() .install(move || rayon::spawn(thread_exec));