From 090e5e0301e5d1dd7979938f573ceff8c08e15ac Mon Sep 17 00:00:00 2001
From: Moritz Hoffmann <antiguru@gmail.com>
Date: Fri, 19 Nov 2021 20:12:24 -0500
Subject: [PATCH] exchange: Improve comments

Signed-off-by: Moritz Hoffmann <antiguru@gmail.com>
---
 timely/src/dataflow/channels/mod.rs              | 4 ++--
 timely/src/dataflow/channels/pushers/exchange.rs | 5 +++++
 2 files changed, 7 insertions(+), 2 deletions(-)

diff --git a/timely/src/dataflow/channels/mod.rs b/timely/src/dataflow/channels/mod.rs
index 352adfff3..7d174f083 100644
--- a/timely/src/dataflow/channels/mod.rs
+++ b/timely/src/dataflow/channels/mod.rs
@@ -54,7 +54,7 @@ impl<T, D> Message<T, D> {
 
         Self::push_at_no_allocation(buffer, time, pusher);
 
-        // TODO: Unclear we always want this here.
+        // Allocate a default buffer to avoid oddly sized or empty buffers
         if buffer.capacity() != Self::default_length() {
             *buffer = Vec::with_capacity(Self::default_length());
         }
@@ -67,7 +67,7 @@ impl<T, D> Message<T, D> {
     #[inline]
     pub fn push_at_no_allocation<P: Push<Bundle<T, D>>>(buffer: &mut Vec<D>, time: T, pusher: &mut P) {
 
-        let data = ::std::mem::replace(buffer, Vec::new());
+        let data = ::std::mem::take(buffer);
         let message = Message::new(time, data, 0, 0);
         let mut bundle = Some(Bundle::from_typed(message));
 
diff --git a/timely/src/dataflow/channels/pushers/exchange.rs b/timely/src/dataflow/channels/pushers/exchange.rs
index fad8340c6..21c781cff 100644
--- a/timely/src/dataflow/channels/pushers/exchange.rs
+++ b/timely/src/dataflow/channels/pushers/exchange.rs
@@ -63,6 +63,9 @@ impl<T: Eq+Data, D: Data, P: Push<Bundle<T, D>>, H: FnMut(&T, &D)->u64> Push<Bun
                 let mask = (self.pushers.len() - 1) as u64;
                 for datum in data.drain(..) {
                     let index = (((self.hash_func)(time, &datum)) & mask) as usize;
+
+                    // Ensure allocated buffers: If the buffer's capacity is less than its default
+                    // capacity, increase the capacity such that it matches the default.
                     if self.buffers[index].capacity() < Message::<T, D>::default_length() {
                         let to_reserve = Message::<T, D>::default_length() - self.buffers[index].capacity();
                         self.buffers[index].reserve(to_reserve);
@@ -85,6 +88,8 @@ impl<T: Eq+Data, D: Data, P: Push<Bundle<T, D>>, H: FnMut(&T, &D)->u64> Push<Bun
             else {
                 for datum in data.drain(..) {
                     let index = (((self.hash_func)(time, &datum)) % self.pushers.len() as u64) as usize;
+                    // Ensure allocated buffers: If the buffer's capacity is less than its default
+                    // capacity, increase the capacity such that it matches the default.
                     if self.buffers[index].capacity() < Message::<T, D>::default_length() {
                         let to_reserve = Message::<T, D>::default_length() - self.buffers[index].capacity();
                         self.buffers[index].reserve(to_reserve);