[exploration] Use crossbeam MPMC channel instead of std::sync::mpsc

Cargo.toml

@@ -44,6 +44,7 @@ ctrlc = "3.1"
 humantime = "1.1.1"
 lscolors = "0.6"
 globset = "0.4"
+crossbeam-channel = "0.3"
 
 [dependencies.clap]
 version = "2.31.2"

src/exec/job.rs

@@ -9,56 +9,41 @@
 use super::CommandTemplate;
 use crate::exit_codes::ExitCode;
 use crate::walk::WorkerResult;
-use std::path::PathBuf;
-use std::sync::mpsc::Receiver;
+use crossbeam_channel::Receiver;
 use std::sync::{Arc, Mutex};
 
+fn read_values(
+    rx: Receiver<WorkerResult>,
+    show_filesystem_errors: bool,
+) -> impl Iterator<Item = std::path::PathBuf> {
+    rx.into_iter().filter_map(move |value| match value {
+        WorkerResult::Entry(val) => Some(val),
+        WorkerResult::Error(err) => {
+            if show_filesystem_errors {
+                print_error!("{}", err);
+            }
+            None
+        }
+    })
+}
+
 /// An event loop that listens for inputs from the `rx` receiver. Each received input will
 /// generate a command with the supplied command template. The generated command will then
 /// be executed, and this process will continue until the receiver's sender has closed.
 pub fn job(
-    rx: Arc<Mutex<Receiver<WorkerResult>>>,
+    rx: Receiver<WorkerResult>,
     cmd: Arc<CommandTemplate>,
     out_perm: Arc<Mutex<()>>,
     show_filesystem_errors: bool,
 ) {
-    loop {
-        // Create a lock on the shared receiver for this thread.
-        let lock = rx.lock().unwrap();
-
-        // Obtain the next result from the receiver, else if the channel
-        // has closed, exit from the loop
-        let value: PathBuf = match lock.recv() {
-            Ok(WorkerResult::Entry(val)) => val,
-            Ok(WorkerResult::Error(err)) => {
-                if show_filesystem_errors {
-                    print_error!("{}", err);
-                }
-                continue;
-            }
-            Err(_) => break,
-        };
-
-        // Drop the lock so that other threads can read from the the receiver.
-        drop(lock);
-        // Generate a command and execute it.
-        cmd.generate_and_execute(&value, Arc::clone(&out_perm));
-    }
+    read_values(rx, show_filesystem_errors)
+        .for_each(|value| cmd.generate_and_execute(&value, Arc::clone(&out_perm)));
 }
 
 pub fn batch(
     rx: Receiver<WorkerResult>,
     cmd: &CommandTemplate,
     show_filesystem_errors: bool,
 ) -> ExitCode {
-    let paths = rx.iter().filter_map(|value| match value {
-        WorkerResult::Entry(val) => Some(val),
-        WorkerResult::Error(err) => {
-            if show_filesystem_errors {
-                print_error!("{}", err);
-            }
-            None
-        }
-    });
-    cmd.generate_and_execute_batch(paths)
+    cmd.generate_and_execute_batch(read_values(rx, show_filesystem_errors))
 }

src/walk.rs

@@ -19,11 +19,11 @@ use std::io;
 use std::path::PathBuf;
 use std::process;
 use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::mpsc::{channel, Receiver, Sender};
 use std::sync::{Arc, Mutex};
 use std::thread;
 use std::time;
 
+use crossbeam_channel::{bounded, Receiver, Sender};
 use ignore::overrides::OverrideBuilder;
 use ignore::{self, WalkBuilder};
 use regex::bytes::Regex;
@@ -54,7 +54,7 @@ pub fn scan(path_vec: &[PathBuf], pattern: Arc<Regex>, config: Arc<FdOptions>) -
     let first_path_buf = path_iter
         .next()
         .expect("Error: Path vector can not be empty");
-    let (tx, rx) = channel();
+    let (tx, rx) = bounded(MAX_BUFFER_LENGTH);
 
     let mut override_builder = OverrideBuilder::new(first_path_buf.as_path());
 
@@ -155,14 +155,12 @@ fn spawn_receiver(
             if cmd.in_batch_mode() {
                 exec::batch(rx, cmd, show_filesystem_errors)
             } else {
-                let shared_rx = Arc::new(Mutex::new(rx));
-
                 let out_perm = Arc::new(Mutex::new(()));
 
                 // Each spawned job will store it's thread handle in here.
                 let mut handles = Vec::with_capacity(threads);
                 for _ in 0..threads {
-                    let rx = Arc::clone(&shared_rx);
+                    let rx = rx.clone();
                     let cmd = Arc::clone(cmd);
                     let out_perm = Arc::clone(&out_perm);