Switch to crossbeam-channel + rayon::spawn (#327)

* Switch to crossbeam-channel + rayon::spawn

* Remove thread_spawn for evaluation altogether

This allows to avoid a deadlock when there is only one Rayon thread, and doesn't sacrifice performance, since the caller of .get_result() had to always block on the iterator to be finished anyway, and all the messages are already sent from separate threads.

* Fix `verbose_mode` test

This one is easier to "fix", since we're in control of the rayon pool - just adding one extra thread to the default number to make sure we always can one root "spawn" call without stealing from the actual pool.

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",
 ]

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

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<Deadline>,
     nth: AtomicUsize,
     best_candidate_size: Arc<AtomicMin>,
-    /// images are sent to the thread for evaluation
+    /// images are sent to the caller thread for evaluation
     #[cfg(feature = "parallel")]
-    eval_send: SyncSender<Candidate>,
+    eval_channel: (Sender<Candidate>, Receiver<Candidate>),
-    // the thread helps evaluate images asynchronously
-    #[cfg(feature = "parallel")]
-    eval_thread: thread::JoinHandle<Option<Candidate>>,
     // in non-parallel mode, images are evaluated synchronously
     #[cfg(not(feature = "parallel"))]
     eval_best_candidate: RefCell<Option<Candidate>>,
@@ -61,15 +56,13 @@ pub(crate) struct Evaluator {
 impl Evaluator {
     pub fn new(deadline: Arc<Deadline>) -> 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,
+            eval_channel,
-            #[cfg(feature = "parallel")]
-            eval_thread: thread::spawn(move || rx.into_iter().min_by_key(Candidate::cmp_key)),
             #[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<Candidate> {
-        drop(self.eval_send); // disconnect the sender, breaking the loop in the thread
+        let (eval_send, eval_recv) = self.eval_channel;
-        self.eval_thread.join().expect("eval thread")
+        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);
 

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<Option<SyncSender<String>>> = RefCell::new(None);
+        static VERBOSE_LOGS: RefCell<Option<Sender<String>>> = 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));