clippy edits

Author: frankmcsherry

benches/barrier.rs

@@ -14,12 +14,12 @@ fn main() {
             .to_stream(scope)
             .unary_stream(Pipeline, "Distinct", move |input, output| {
                 input.for_each(|time, data| {
-                    let mut counts = counts_by_time.entry(time.time().clone())
-                                                   .or_insert(HashMap::new());
+                    let counts = counts_by_time.entry(time.time().clone())
+                                               .or_insert(HashMap::new());
                     let mut session = output.session(&time);
                     for &datum in data.iter() {
-                        let mut count = counts.entry(datum)
-                                              .or_insert(0);
+                        let count = counts.entry(datum)
+                                          .or_insert(0);
                         if *count == 0 {
                            session.give(datum);
                         }

examples/distinct.rs

@@ -50,20 +50,20 @@ fn main() {
 
                         // hold on to edge changes until it is time.
                         input1.for_each(|time, data| {
-                            edge_stash.entry(time).or_insert(vec![]).extend(data.drain(..));
+                            edge_stash.entry(time).or_insert(Vec::new()).extend(data.drain(..));
                         });
 
                         // hold on to rank changes until it is time.
                         input2.for_each(|time, data| {
-                            rank_stash.entry(time).or_insert(vec![]).extend(data.drain(..));
+                            rank_stash.entry(time).or_insert(Vec::new()).extend(data.drain(..));
                         });
 
                         let frontiers = &[input1.frontier(), input2.frontier()];
 
                         for (time, edge_changes) in edge_stash.iter_mut() {
                             if frontiers.iter().all(|f| !f.less_equal(time)) {
 
-                                let mut session = output.session(&time);
+                                let mut session = output.session(time);
 
                                 compact(edge_changes);
 
@@ -93,12 +93,12 @@ fn main() {
                             }
                         }
 
-                        edge_stash.retain(|_key, val| val.len() > 0);
+                        edge_stash.retain(|_key, val| !val.is_empty());
 
                         for (time, rank_changes) in rank_stash.iter_mut() {
                             if frontiers.iter().all(|f| !f.less_equal(time)) {
 
-                                let mut session = output.session(&time);
+                                let mut session = output.session(time);
 
                                 compact(rank_changes);
 
@@ -137,7 +137,7 @@ fn main() {
                             }
                         }
 
-                        rank_stash.retain(|_key, val| val.len() > 0);
+                        rank_stash.retain(|_key, val| !val.is_empty());
 
                     }
                 }
@@ -179,7 +179,7 @@ fn main() {
 }
 
 fn compact<T: Ord>(list: &mut Vec<(T, i64)>) {
-    if list.len() > 0 {
+    if !list.is_empty() {
         list.sort_by(|x,y| x.0.cmp(&y.0));
         for i in 0 .. list.len() - 1 {
             if list[i].0 == list[i+1].0 {
@@ -193,7 +193,7 @@ fn compact<T: Ord>(list: &mut Vec<(T, i64)>) {
 
 // this method allocates some rank between elements of `edges`.
 fn allocate(rank: i64, edges: &[(usize, i64)], send: &mut Vec<(usize, i64)>) {
-    if edges.len() > 0 {
+    if !edges.is_empty() {
         assert!(rank >= 0);
         assert!(edges.iter().all(|x| x.1 > 0));
 

examples/pagerank.rs

@@ -49,7 +49,7 @@ fn main() {
                             }
                         }
 
-                        queues.retain(|_key, val| val.len() > 0);
+                        queues.retain(|_key, val| !val.is_empty());
                     }})
                  .inspect(|x| println!("seen: {:?}", x))
                  .probe_with(&mut probe);

examples/wordcount.rs

@@ -126,7 +126,7 @@ impl<D> Content<D> {
     }
 
     /// Pushes `buffer` into `pusher`, ensuring that `buffer` remains valid once returned.
-    #[inline]
+    #[inline(always)]
     pub fn push_at<T, P: Push<(T, Content<D>)>>(buffer: &mut Vec<D>, time: T, pusher: &mut P) {
 
         let data = Content::from_typed(buffer);
@@ -139,13 +139,15 @@ impl<D> Content<D> {
             *buffer = typed;
         }
         else {
+            // println!("re-allocating (nothing returned)");
             *buffer = Vec::with_capacity(Content::<D>::default_length());
         }
 
         // TODO : Assert failing, but not sure if is bug when deser can make arbitrary lengths
         // TODO : in clone. Revisit!
         // assert!(buffer.capacity() == Content::<D>::default_length());
         if buffer.capacity() != Content::<D>::default_length() {
+            // println!("re-allocating (wrong size)");
             *buffer = Vec::with_capacity(Content::<D>::default_length());
         }
     }

src/dataflow/channels/message.rs

@@ -19,7 +19,7 @@ use dataflow::channels::{Message, Content};
 
 use abomonation::Abomonation;
 
-/// A ParallelizationContract allocates paired `Push` and `Pull` implementors.
+/// A `ParallelizationContract` allocates paired `Push` and `Pull` implementors.
 pub trait ParallelizationContract<T: 'static, D: 'static> {
     /// Type implementing `Push` produced by this pact.
     type Pusher: Push<(T, Content<D>)>+'static;
@@ -36,7 +36,7 @@ impl<T: 'static, D: 'static> ParallelizationContract<T, D> for Pipeline {
     type Pusher = Pusher<T, D, ThreadPusher<Message<T, D>>>;
     type Puller = Puller<T, D, ThreadPuller<Message<T, D>>>;
     fn connect<A: Allocate>(self, allocator: &mut A, identifier: usize) -> (Self::Pusher, Self::Puller) {
-        // ignore &mut A and use thread allocator
+        // ignore `&mut A` and use thread allocator
         let (pusher, puller) = Thread::new::<Message<T, D>>();
 
         (Pusher::new(pusher, allocator.index(), allocator.index(), identifier),
@@ -56,9 +56,7 @@ impl<D, F: Fn(&D)->u64> Exchange<D, F> {
     }
 }
 
-// Exchange uses a Box<Pushable> because it cannot know what type of pushable will return from the allocator.
-// The PactObserver will do some buffering for Exchange, cutting down on the virtual calls, but we still
-// would like to get the vectors it sends back, so that they can be re-used if possible.
+// Exchange uses a `Box<Pushable>` because it cannot know what type of pushable will return from the allocator.
 impl<T: Eq+Data+Abomonation, D: Data+Abomonation, F: Fn(&D)->u64+'static> ParallelizationContract<T, D> for Exchange<D, F> {
     // TODO: The closure in the type prevents us from naming it. 
     //       Could specialize `ExchangePusher` to a time-free version.
@@ -170,7 +168,7 @@ impl<T, D, P: Pull<Message<T, D>>> Pull<(T, Content<D>)> for Puller<T, D, P> {
 
         ::std::mem::swap(&mut previous, self.puller.pull());
 
-        if let Some(ref message) = previous.as_ref() {
+        if let Some(message) = previous.as_ref() {
 
             ::logging::log(&::logging::MESSAGES, ::logging::MessagesEvent {
                 is_send: false,

src/dataflow/channels/pact.rs

@@ -15,7 +15,7 @@ pub struct Counter<T: Ord, D, P: Push<(T, Content<D>)>> {
 }
 
 impl<T: Ord, D, P: Push<(T, Content<D>)>> Push<(T, Content<D>)> for Counter<T, D, P> where T : Eq+Clone+'static {
-    #[inline] 
+    #[inline(always)]
     fn push(&mut self, message: &mut Option<(T, Content<D>)>) {
         if let Some((ref time, ref data)) = *message {
             self.produced.borrow_mut().update(time.clone(), data.len() as i64);
@@ -38,6 +38,7 @@ impl<T, D, P: Push<(T, Content<D>)>> Counter<T, D, P> where T : Ord+Clone+'stati
         }
     }
     /// A references to shared changes in counts, for cloning or draining.
+    #[inline(always)]
     pub fn produced(&self) -> &Rc<RefCell<ChangeBatch<T>>> {
         &self.produced
     }

src/dataflow/channels/pushers/counter.rs

@@ -59,7 +59,7 @@ impl<T: Eq+Clone+'static, D: Data+Abomonation, P: Push<(T, Content<D>)>, H: Fn(&
                 if (self.pushers.len() & (self.pushers.len() - 1)) == 0 {
                     let mask = (self.pushers.len() - 1) as u64;
                     for datum in data.drain(..) {
-                        let index = (((self.hash_func)(&time, &datum)) & mask) as usize;
+                        let index = (((self.hash_func)(time, &datum)) & mask) as usize;
 
                         self.buffers[index].push(datum);
                         if self.buffers[index].len() == self.buffers[index].capacity() {
@@ -78,7 +78,7 @@ impl<T: Eq+Clone+'static, D: Data+Abomonation, P: Push<(T, Content<D>)>, H: Fn(&
                 // as a last resort, use mod (%)
                 else {
                     for datum in data.drain(..) {
-                        let index = (((self.hash_func)(&time, &datum)) % self.pushers.len() as u64) as usize;
+                        let index = (((self.hash_func)(time, &datum)) % self.pushers.len() as u64) as usize;
                         self.buffers[index].push(datum);
                         if self.buffers[index].len() == self.buffers[index].capacity() {
                             self.flush(index);

src/dataflow/channels/pushers/exchange.rs

@@ -95,7 +95,7 @@ impl<S: Scope, K: ExchangeData+Hash+Eq, V: ExchangeData> Aggregate<S, K, V> for
 
 			// pop completed aggregates, send along whatever
 			notificator.for_each(|time,_,_| {
-				if let Some(aggs) = aggregates.remove(&time.time()) {
+				if let Some(aggs) = aggregates.remove(time.time()) {
 					let mut session = output.session(&time);
 					for (key, agg) in aggs {
 						session.give(emit(key, agg));

src/dataflow/operators/aggregation/aggregate.rs

@@ -72,7 +72,7 @@ impl<S: Scope, K: ExchangeData+Hash+Eq, V: ExchangeData> StateMachine<S, K, V> f
             // stash each input and request a notification when ready
             input.for_each(|time, data| {
                 // stash if not time yet
-                if notificator.frontier(0).iter().any(|x| x.less_than(&time.time())) {
+                if notificator.frontier(0).iter().any(|x| x.less_than(time.time())) {
                     pending.entry(time.time().clone()).or_insert_with(Vec::new).extend(data.drain(..));
                     notificator.notify_at(time);
                 }

src/dataflow/operators/aggregation/state_machine.rs

@@ -40,7 +40,7 @@ impl<G: Scope, D: ExchangeData> Broadcast<D> for Stream<G, D> {
 
         let channel_id = scope.new_identifier();
 
-        assert!(pushers.len() == scope.peers());
+        assert_eq!(pushers.len(), scope.peers());
 
         let receiver = Puller::new(puller, scope.index(), channel_id);
 

src/dataflow/operators/broadcast.rs

@@ -58,7 +58,7 @@ impl<G: Scope, D: Data> Accumulate<G, D> for Stream<G, D> {
         let mut accums = HashMap::new();
         self.unary_notify(Pipeline, "Accumulate", vec![], move |input, output, notificator| {
             input.for_each(|time, data| {
-                logic(&mut accums.entry(time.time().clone()).or_insert(default.clone()), data);
+                logic(&mut accums.entry(time.time().clone()).or_insert_with(|| default.clone()), data);
                 notificator.notify_at(time);
             });
 

src/dataflow/operators/count.rs

@@ -36,7 +36,7 @@ impl<'a, T: Timestamp, D, P: Pull<(T, Content<D>)>> InputHandle<T, D, P> {
     /// Reads the next input buffer (at some timestamp `t`) and a corresponding capability for `t`.
     /// The timestamp `t` of the input buffer can be retrieved by invoking `.time()` on the capability.
     /// Returns `None` when there's no more data available.
-    #[inline]
+    #[inline(always)]
     pub fn next(&mut self) -> Option<(Capability<T>, &mut Content<D>)> {
         let internal = &mut self.internal;
         self.pull_counter.next().map(|(time, content)| {
@@ -77,7 +77,7 @@ impl<'a, T: Timestamp, D, P: Pull<(T, Content<D>)>+'a> FrontieredInputHandle<'a,
     /// Reads the next input buffer (at some timestamp `t`) and a corresponding capability for `t`.
     /// The timestamp `t` of the input buffer can be retrieved by invoking `.time()` on the capability.
     /// Returns `None` when there's no more data available.
-    #[inline]
+    #[inline(always)]
     pub fn next(&mut self) -> Option<(Capability<T>, &mut Content<D>)> {
         self.handle.next()
     }
@@ -106,7 +106,7 @@ impl<'a, T: Timestamp, D, P: Pull<(T, Content<D>)>+'a> FrontieredInputHandle<'a,
     }
 
     /// Inspect the frontier associated with this input.
-    #[inline]
+    #[inline(always)]
     pub fn frontier(&self) -> &'a MutableAntichain<T> {
         self.frontier
     }

src/dataflow/operators/generic/handles.rs

@@ -6,7 +6,7 @@ use dataflow::operators::Capability;
 
 /// Tracks requests for notification and delivers available notifications.
 ///
-/// Notificator is meant to manage the delivery of requested notifications in the presence of
+/// `Notificator` is meant to manage the delivery of requested notifications in the presence of
 /// inputs that may have outstanding messages to deliver. The notificator tracks the frontiers,
 /// as presented from the outside, for each input. Requested notifications can be served only
 /// once there are no frontier elements less-or-equal to them, and there are no other pending
@@ -240,7 +240,7 @@ fn notificator_delivers_notifications_in_topo_order() {
 
 /// Tracks requests for notification and delivers available notifications.
 ///
-/// FrontierNotificator is meant to manage the delivery of requested notifications in the
+/// `FrontierNotificator` is meant to manage the delivery of requested notifications in the
 /// presence of inputs that may have outstanding messages to deliver.
 /// The notificator inspects the frontiers, as presented from the outside, for each input.
 /// Requested notifications can be served only once there are no frontier elements less-or-equal
@@ -359,7 +359,7 @@ impl<T: Timestamp> FrontierNotificator<T> {
         // It should be safe to append any ordered subset of self.pending to self.available,
         // in that the sequence of capabilities in self.available will remain non-decreasing.
 
-        if self.pending.len() > 0 {
+        if !self.pending.is_empty() {
 
             self.pending.sort_by(|x,y| x.0.time().cmp(y.0.time()));
             for i in 0 .. self.pending.len() - 1 {

src/dataflow/operators/generic/notificator.rs

@@ -92,7 +92,7 @@ impl<T:Timestamp, D: Data, D2: Data, F: Fn(D)->(u64, D2)> Operate<T> for Operato
             let outputs = self.outputs.iter_mut();
 
             // TODO : This results in small sends for many parts, as sessions does the buffering
-            let mut sessions: Vec<_> = outputs.map(|x| x.session(&time)).collect();
+            let mut sessions: Vec<_> = outputs.map(|x| x.session(time)).collect();
 
             for (part, datum) in data.drain(..).map(&self.route) {
                 sessions[part as usize].give(datum);

src/dataflow/operators/partition.rs

@@ -70,11 +70,11 @@ impl<S: Scope, D: Data> Reclock<S, D> for Stream<S, D> {
             notificator.for_each(|cap,_,_| {
                 let mut session = output.session(&cap);
                 for &mut (ref t, ref mut data) in &mut stash {
-                    if t.less_equal(&cap.time()) {
+                    if t.less_equal(cap.time()) {
                         session.give_content(data);
                     }
                 }
-                stash.retain(|x| !x.0.less_equal(&cap.time()));
+                stash.retain(|x| !x.0.less_equal(cap.time()));
             });
         })
     }

src/dataflow/operators/reclock.rs

@@ -96,7 +96,7 @@ impl<G: Scope> UnorderedInput<G> for G {
             peers:    peers,
         });
 
-        return ((helper, cap), Stream::new(Source { index: index, port: 0 }, registrar, self.clone()));
+        ((helper, cap), Stream::new(Source { index: index, port: 0 }, registrar, self.clone()))
     }
 }
 
@@ -130,7 +130,7 @@ impl<T:Timestamp> Operate<T> for UnorderedOperator<T> {
     {
         self.produced.borrow_mut().drain_into(&mut produced[0]);
         self.internal.borrow_mut().drain_into(&mut internal[0]);
-        return false;
+        false
     }
 
     fn notify_me(&self) -> bool { false }

src/dataflow/operators/unordered_input.rs

@@ -212,7 +212,7 @@ impl<T:Ord> ChangeBatch<T> {
     /// assert!(!batch2.is_empty());
     ///```
     pub fn drain_into(&mut self, other: &mut ChangeBatch<T>) where T: Clone {
-        if other.updates.len() == 0 {
+        if other.updates.is_empty() {
             ::std::mem::swap(self, other);
         }
         else {