Apply various Clippy recommendations (#603)

Author: frankmcsherry

bytes/src/lib.rs

@@ -96,7 +96,7 @@ pub mod arc {
                 sequestered: self.sequestered.clone(),
             };
 
-            unsafe { self.ptr = self.ptr.offset(index as isize); }
+            unsafe { self.ptr = self.ptr.add(index); }
             self.len -= index;
 
             result
@@ -161,7 +161,7 @@ pub mod arc {
         /// shared4.try_merge(shared2).ok().expect("Failed to merge 4 and 231");
         /// ```
         pub fn try_merge(&mut self, other: Bytes) -> Result<(), Bytes> {
-            if Arc::ptr_eq(&self.sequestered, &other.sequestered) && ::std::ptr::eq(unsafe { self.ptr.offset(self.len as isize) }, other.ptr) {
+            if Arc::ptr_eq(&self.sequestered, &other.sequestered) && ::std::ptr::eq(unsafe { self.ptr.add(self.len) }, other.ptr) {
                 self.len += other.len;
                 Ok(())
             }

communication/src/allocator/process.rs

@@ -35,7 +35,7 @@ impl AllocateBuilder for ProcessBuilder {
 
         // Initialize buzzers; send first, then recv.
         for worker in self.buzzers_send.iter() {
-            let buzzer = Buzzer::new();
+            let buzzer = Buzzer::default();
             worker.send(buzzer).expect("Failed to send buzzer");
         }
         let mut buzzers = Vec::with_capacity(self.buzzers_recv.len());
@@ -88,8 +88,8 @@ impl Process {
 
         counters_recv
             .into_iter()
-            .zip(buzzers_send.into_iter())
-            .zip(buzzers_recv.into_iter())
+            .zip(buzzers_send)
+            .zip(buzzers_recv)
             .enumerate()
             .map(|(index, ((recv, bsend), brecv))| {
                 ProcessBuilder {

communication/src/allocator/thread.rs

@@ -15,11 +15,12 @@ pub struct ThreadBuilder;
 
 impl AllocateBuilder for ThreadBuilder {
     type Allocator = Thread;
-    fn build(self) -> Self::Allocator { Thread::new() }
+    fn build(self) -> Self::Allocator { Thread::default() }
 }
 
 
 /// An allocator for intra-thread communication.
+#[derive(Default)]
 pub struct Thread {
     /// Shared counts of messages in channels.
     events: Rc<RefCell<Vec<usize>>>,
@@ -53,13 +54,6 @@ pub type ThreadPusher<T> = CountPusher<T, Pusher<T>>;
 pub type ThreadPuller<T> = CountPuller<T, Puller<T>>;
 
 impl Thread {
-    /// Allocates a new thread-local channel allocator.
-    pub fn new() -> Self {
-        Thread {
-            events: Rc::new(RefCell::new(Default::default())),
-        }
-    }
-
     /// Creates a new thread-local channel from an identifier and shared counts.
     pub fn new_from<T: 'static>(identifier: usize, events: Rc<RefCell<Vec<usize>>>)
         -> (ThreadPusher<T>, ThreadPuller<T>)

communication/src/allocator/zero_copy/allocator.rs

@@ -82,7 +82,7 @@ impl<A: AllocateBuilder> TcpBuilder<A> {
         // Fulfill puller obligations.
         let mut recvs = Vec::with_capacity(self.peers);
         for promise in self.promises.into_iter() {
-            let buzzer = crate::buzzer::Buzzer::new();
+            let buzzer = crate::buzzer::Buzzer::default();
             let queue = MergeQueue::new(buzzer);
             promise.send(queue.clone()).expect("Failed to send MergeQueue");
             recvs.push(queue.clone());

communication/src/allocator/zero_copy/allocator_process.rs

@@ -60,7 +60,7 @@ impl ProcessBuilder {
         // Fulfill puller obligations.
         let mut recvs = Vec::with_capacity(self.peers);
         for puller in self.pullers.into_iter() {
-            let buzzer = crate::buzzer::Buzzer::new();
+            let buzzer = crate::buzzer::Buzzer::default();
             let queue = MergeQueue::new(buzzer);
             puller.send(queue.clone()).expect("Failed to send MergeQueue");
             recvs.push(queue.clone());

communication/src/allocator/zero_copy/initialize.rs

@@ -63,10 +63,8 @@ pub fn initialize_networking_from_sockets<S: Stream + 'static>(
 -> ::std::io::Result<(Vec<TcpBuilder<ProcessBuilder>>, CommsGuard)>
 {
     // Sockets are expected to be blocking,
-    for socket in sockets.iter_mut() {
-        if let Some(socket) = socket {
-            socket.set_nonblocking(false).expect("failed to set socket to blocking");
-        }
+    for socket in sockets.iter_mut().flatten() {
+        socket.set_nonblocking(false).expect("failed to set socket to blocking");
     }
 
     let processes = sockets.len();

communication/src/allocator/zero_copy/tcp.rs

@@ -1,4 +1,4 @@
-//!
+//! Methods related to reading from and writing to TCP connections
 
 use std::io::{self, Write};
 use crossbeam_channel::{Sender, Receiver};
@@ -67,9 +67,9 @@ where
         assert!(!buffer.empty().is_empty());
 
         // Attempt to read some more bytes into self.buffer.
-        let read = match reader.read(&mut buffer.empty()) {
+        let read = match reader.read(buffer.empty()) {
             Err(x) => tcp_panic("reading data", x),
-            Ok(n) if n == 0 => {
+            Ok(0) => {
                 tcp_panic(
                     "reading data",
                     std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "socket closed"),
@@ -102,7 +102,7 @@ where
                     panic!("Clean shutdown followed by data.");
                 }
                 buffer.ensure_capacity(1);
-                if reader.read(&mut buffer.empty()).unwrap_or_else(|e| tcp_panic("reading EOF", e)) > 0 {
+                if reader.read(buffer.empty()).unwrap_or_else(|e| tcp_panic("reading EOF", e)) > 0 {
                     panic!("Clean shutdown followed by data.");
                 }
             }
@@ -141,7 +141,7 @@ pub fn send_loop<S: Stream>(
     logger.as_mut().map(|l| l.log(StateEvent { send: true, process, remote, start: true, }));
 
     let mut sources: Vec<MergeQueue> = sources.into_iter().map(|x| {
-        let buzzer = crate::buzzer::Buzzer::new();
+        let buzzer = crate::buzzer::Buzzer::default();
         let queue = MergeQueue::new(buzzer);
         x.send(queue.clone()).expect("failed to send MergeQueue");
         queue

communication/src/buzzer.rs

@@ -8,13 +8,11 @@ pub struct Buzzer {
     thread: Thread,
 }
 
+impl Default for Buzzer {
+    fn default() -> Self { Self { thread: std::thread::current() } }
+}
+
 impl Buzzer {
-    /// Creates a new buzzer for the current thread.
-    pub fn new() -> Self {
-        Self {
-            thread: std::thread::current()
-        }
-    }
     /// Unparks the target thread.
     pub fn buzz(&self) {
         self.thread.unpark()

communication/src/initialize.rs

@@ -155,15 +155,15 @@ impl Config {
                 Ok((vec![GenericBuilder::Thread(ThreadBuilder)], Box::new(())))
             },
             Config::Process(threads) => {
-                Ok((Process::new_vector(threads).into_iter().map(|x| GenericBuilder::Process(x)).collect(), Box::new(())))
+                Ok((Process::new_vector(threads).into_iter().map(GenericBuilder::Process).collect(), Box::new(())))
             },
             Config::ProcessBinary(threads) => {
-                Ok((ProcessBuilder::new_vector(threads).into_iter().map(|x| GenericBuilder::ProcessBinary(x)).collect(), Box::new(())))
+                Ok((ProcessBuilder::new_vector(threads).into_iter().map(GenericBuilder::ProcessBinary).collect(), Box::new(())))
             },
             Config::Cluster { threads, process, addresses, report, log_fn } => {
                 match initialize_networking(addresses, process, threads, report, log_fn) {
                     Ok((stuff, guard)) => {
-                        Ok((stuff.into_iter().map(|x| GenericBuilder::ZeroCopy(x)).collect(), Box::new(guard)))
+                        Ok((stuff.into_iter().map(GenericBuilder::ZeroCopy).collect(), Box::new(guard)))
                     },
                     Err(err) => Err(format!("failed to initialize networking: {}", err))
                 }

communication/src/lib.rs

@@ -181,11 +181,11 @@ fn promise_futures<T>(sends: usize, recvs: usize) -> (Vec<Vec<Sender<T>>>, Vec<V
     let mut senders: Vec<_> = (0 .. sends).map(|_| Vec::with_capacity(recvs)).collect();
     let mut recvers: Vec<_> = (0 .. recvs).map(|_| Vec::with_capacity(sends)).collect();
 
-    for sender in 0 .. sends {
-        for recver in 0 .. recvs {
+    for sender in senders.iter_mut() {
+        for recver in recvers.iter_mut() {
             let (send, recv) = crossbeam_channel::unbounded();
-            senders[sender].push(send);
-            recvers[recver].push(recv);
+            sender.push(send);
+            recver.push(recv);
         }
     }