Replace crossbeam-channel by the std variant (#653)

Replaces crossbeam-channel with the channels in the standard library. The
implementations in the standard library are derived from crossbeam, but
seem to be more actively maintained.

Signed-off-by: Moritz Hoffmann <[email protected]>

Author: antiguru

communication/Cargo.toml

@@ -24,7 +24,6 @@ serde = { version = "1.0", features = ["derive"] }
 timely_bytes = { path = "../bytes", version = "0.13" }
 timely_container = { path = "../container", version = "0.15.0" }
 timely_logging = { path = "../logging", version = "0.13" }
-crossbeam-channel = "0.5"
 
 # Lgalloc only supports linux and macos, don't depend on any other OS.
 [target.'cfg(any(target_os = "linux", target_os = "macos"))'.dev-dependencies]

communication/src/allocator/counters.rs

@@ -2,6 +2,7 @@
 
 use std::rc::Rc;
 use std::cell::RefCell;
+use std::sync::mpsc::Sender;
 
 use crate::{Push, Pull};
 
@@ -51,8 +52,6 @@ impl<T, P: Push<T>> Push<T> for Pusher<T, P> {
     }
 }
 
-use crossbeam_channel::Sender;
-
 /// The push half of an intra-thread channel.
 pub struct ArcPusher<T, P: Push<T>> {
     index: usize,

communication/src/allocator/process.rs

@@ -6,7 +6,7 @@ use std::sync::{Arc, Mutex};
 use std::any::Any;
 use std::time::Duration;
 use std::collections::{HashMap};
-use crossbeam_channel::{Sender, Receiver};
+use std::sync::mpsc::{Sender, Receiver};
 
 use crate::allocator::thread::{ThreadBuilder};
 use crate::allocator::{Allocate, AllocateBuilder, PeerBuilder, Thread};
@@ -80,7 +80,7 @@ impl PeerBuilder for Process {
         let mut counters_send = Vec::with_capacity(peers);
         let mut counters_recv = Vec::with_capacity(peers);
         for _ in 0 .. peers {
-            let (send, recv) = crossbeam_channel::unbounded();
+            let (send, recv) = std::sync::mpsc::channel();
             counters_send.push(send);
             counters_recv.push(recv);
         }
@@ -130,7 +130,7 @@ impl Allocate for Process {
                 let mut pushers = Vec::with_capacity(self.peers);
                 let mut pullers = Vec::with_capacity(self.peers);
                 for buzzer in self.buzzers.iter() {
-                    let (s, r): (Sender<T>, Receiver<T>) = crossbeam_channel::unbounded();
+                    let (s, r): (Sender<T>, Receiver<T>) = std::sync::mpsc::channel();
                     // TODO: the buzzer in the pusher may be redundant, because we need to buzz post-counter.
                     pushers.push((Pusher { target: s }, buzzer.clone()));
                     pullers.push(Puller { source: r, current: None });

communication/src/allocator/zero_copy/allocator.rs

@@ -2,7 +2,7 @@
 use std::rc::Rc;
 use std::cell::RefCell;
 use std::collections::{VecDeque, HashMap, hash_map::Entry};
-use crossbeam_channel::{Sender, Receiver};
+use std::sync::mpsc::{Sender, Receiver};
 
 use timely_bytes::arc::Bytes;
 

communication/src/allocator/zero_copy/allocator_process.rs

@@ -3,7 +3,7 @@
 use std::rc::Rc;
 use std::cell::RefCell;
 use std::collections::{VecDeque, HashMap, hash_map::Entry};
-use crossbeam_channel::{Sender, Receiver};
+use std::sync::mpsc::{Sender, Receiver};
 
 use timely_bytes::arc::Bytes;
 

communication/src/allocator/zero_copy/tcp.rs

@@ -1,7 +1,7 @@
 //! Methods related to reading from and writing to TCP connections
 
 use std::io::{self, Write};
-use crossbeam_channel::{Sender, Receiver};
+use std::sync::mpsc::{Sender, Receiver};
 
 use crate::networking::MessageHeader;
 

communication/src/lib.rs

@@ -107,6 +107,8 @@ pub use allocator::Generic as Allocator;
 pub use allocator::{Allocate, Exchangeable};
 pub use initialize::{initialize, initialize_from, Config, WorkerGuards};
 
+use std::sync::mpsc::{Sender, Receiver};
+
 use timely_bytes::arc::Bytes;
 
 /// A type that can be serialized and deserialized through `Bytes`.
@@ -169,8 +171,6 @@ impl<T, P: ?Sized + Pull<T>> Pull<T> for Box<P> {
 }
 
 
-use crossbeam_channel::{Sender, Receiver};
-
 /// Allocate a matrix of send and receive changes to exchange items.
 ///
 /// This method constructs channels for `sends` threads to create and send
@@ -183,7 +183,7 @@ fn promise_futures<T>(sends: usize, recvs: usize) -> (Vec<Vec<Sender<T>>>, Vec<V
 
     for sender in senders.iter_mut() {
         for recver in recvers.iter_mut() {
-            let (send, recv) = crossbeam_channel::unbounded();
+            let (send, recv) = std::sync::mpsc::channel();
             sender.push(send);
             recver.push(recv);
         }