ripgrep: move minimum version to Rust stable

This also updates some code to make use of our more liberal versioning
requirement, including the use of crossbeam-channel instead of the MsQueue
from the older an unmaintained crossbeam 0.3. This does regrettably add
a sizable number of dependencies, however, compile times seem mostly
unaffected.

Closes #1019

ignore/src/lib.rs

@@ -46,7 +46,7 @@ See the documentation for `WalkBuilder` for many other options.
 
 #![deny(missing_docs)]
 
-extern crate crossbeam;
+extern crate crossbeam_channel as channel;
 extern crate globset;
 #[macro_use]
 extern crate lazy_static;

ignore/src/walk.rs

@@ -10,7 +10,7 @@ use std::thread;
 use std::time::Duration;
 use std::vec;
 
-use crossbeam::sync::MsQueue;
+use channel;
 use same_file::Handle;
 use walkdir::{self, WalkDir};
 
@@ -956,7 +956,14 @@ impl WalkParallel {
     ) where F: FnMut() -> Box<FnMut(Result<DirEntry, Error>) -> WalkState + Send + 'static> {
         let mut f = mkf();
         let threads = self.threads();
-        let queue = Arc::new(MsQueue::new());
+        // TODO: Figure out how to use a bounded channel here. With an
+        // unbounded channel, the workers can run away and will up memory
+        // with all of the file paths. But a bounded channel doesn't work since
+        // our producers are also are consumers, so they end up getting stuck.
+        //
+        // We probably need to rethink parallel traversal completely to fix
+        // this.
+        let (tx, rx) = channel::unbounded();
         let mut any_work = false;
         // Send the initial set of root paths to the pool of workers.
         // Note that we only send directories. For files, we send to them the
@@ -976,7 +983,7 @@ impl WalkParallel {
                         }
                     }
                 };
-            queue.push(Message::Work(Work {
+            tx.send(Message::Work(Work {
                 dent: dent,
                 ignore: self.ig_root.clone(),
             }));
@@ -994,7 +1001,8 @@ impl WalkParallel {
         for _ in 0..threads {
             let worker = Worker {
                 f: mkf(),
-                queue: queue.clone(),
+                tx: tx.clone(),
+                rx: rx.clone(),
                 quit_now: quit_now.clone(),
                 is_waiting: false,
                 is_quitting: false,
@@ -1007,6 +1015,8 @@ impl WalkParallel {
             };
             handles.push(thread::spawn(|| worker.run()));
         }
+        drop(tx);
+        drop(rx);
         for handle in handles {
             handle.join().unwrap();
         }
@@ -1099,8 +1109,10 @@ impl Work {
 struct Worker {
     /// The caller's callback.
     f: Box<FnMut(Result<DirEntry, Error>) -> WalkState + Send + 'static>,
-    /// A queue of work items. This is multi-producer and multi-consumer.
-    queue: Arc<MsQueue<Message>>,
+    /// The push side of our mpmc queue.
+    tx: channel::Sender<Message>,
+    /// The receive side of our mpmc queue.
+    rx: channel::Receiver<Message>,
     /// Whether all workers should quit at the next opportunity. Note that
     /// this is distinct from quitting because of exhausting the contents of
     /// a directory. Instead, this is used when the caller's callback indicates
@@ -1235,7 +1247,7 @@ impl Worker {
         };
 
         if !should_skip_path && !should_skip_filesize {
-            self.queue.push(Message::Work(Work {
+            self.tx.send(Message::Work(Work {
                 dent: dent,
                 ignore: ig.clone(),
             }));
@@ -1252,7 +1264,7 @@ impl Worker {
             if self.is_quit_now() {
                 return None;
             }
-            match self.queue.try_pop() {
+            match self.rx.try_recv() {
                 Some(Message::Work(work)) => {
                     self.waiting(false);
                     self.quitting(false);
@@ -1294,7 +1306,7 @@ impl Worker {
                     self.quitting(false);
                     if self.num_waiting() == self.threads {
                         for _ in 0..self.threads {
-                            self.queue.push(Message::Quit);
+                            self.tx.send(Message::Quit);
                         }
                     } else {
                         // You're right to consider this suspicious, but it's