Align ContainerBytes implementations with padding

Author: frankmcsherry

timely/src/dataflow/channels/mod.rs

@@ -74,7 +74,6 @@ where
         let time_size = ::bincode::serialized_size(&time).expect("bincode::serialized_size() failed") as usize;
         // We expect to find the `data` payload at `8 + 8 + round_up(time_size)`;
         let bytes_read = 8 + 8 + ((time_size + 7) & !7);
-        // let bytes_read = bytes.len() - slice.len();
         bytes.extract_to(bytes_read);
         let data: C = ContainerBytes::from_bytes(bytes);
         Self { time, data, from, seq }
@@ -113,6 +112,8 @@ pub trait ContainerBytes {
 
 mod implementations {
 
+    use std::io::Write;
+
     use serde::{Serialize, Deserialize};
     use crate::dataflow::channels::ContainerBytes;
 
@@ -122,11 +123,16 @@ mod implementations {
         }
 
         fn length_in_bytes(&self) -> usize {
-            ::bincode::serialized_size(&self).expect("bincode::serialized_size() failed") as usize
+            let length = ::bincode::serialized_size(&self).expect("bincode::serialized_size() failed") as usize;
+            (length + 7) & !7
         }
 
-        fn into_bytes<W: ::std::io::Write>(&self, writer: &mut W) {
-            ::bincode::serialize_into(writer, &self).expect("bincode::serialize_into() failed");
+        fn into_bytes<W: Write>(&self, writer: &mut W) {
+            let mut counter = WriteCounter::new(writer);
+            ::bincode::serialize_into(&mut counter, &self).expect("bincode::serialize_into() failed");
+            let written = counter.count;
+            let written_slop = ((written + 7) & !7) - written;
+            counter.write(&[0u8; 8][..written_slop]).unwrap();
         }
     }
 
@@ -137,11 +143,60 @@ mod implementations {
         }
 
         fn length_in_bytes(&self) -> usize {
-            ::bincode::serialized_size(&self).expect("bincode::serialized_size() failed") as usize
+            let length = ::bincode::serialized_size(&self).expect("bincode::serialized_size() failed") as usize;
+            (length + 7) & !7
+        }
+
+        fn into_bytes<W: Write>(&self, writer: &mut W) {
+            let mut counter = WriteCounter::new(writer);
+            ::bincode::serialize_into(&mut counter, &self).expect("bincode::serialize_into() failed");
+            let written = counter.count;
+            let written_slop = ((written + 7) & !7) - written;
+            counter.write(&[0u8; 8][..written_slop]).unwrap();
+        }
+    }
+
+    use write_counter::WriteCounter;
+    /// A `Write` wrapper that counts the bytes written.
+    mod write_counter {
+
+        use ::std::io::{Write, IoSlice, Result};
+        use std::fmt::Arguments;
+
+        /// A write wrapper that tracks the bytes written.
+        pub struct WriteCounter<W> {
+            inner: W,
+            pub count: usize,
+        }
+
+        impl<W> WriteCounter<W> {
+            /// Creates a new counter wrapper from a writer.
+            pub fn new(inner: W) -> Self {
+                Self { inner, count: 0 }
+            }
         }
 
-        fn into_bytes<W: ::std::io::Write>(&self, writer: &mut W) {
-            ::bincode::serialize_into(writer, &self).expect("bincode::serialize_into() failed");
+        impl<W: Write> Write for WriteCounter<W> {
+            fn write(&mut self, buf: &[u8]) -> Result<usize> {
+                let written = self.inner.write(buf)?;
+                self.count += written;
+                Ok(written)
+            }
+            fn flush(&mut self) -> Result<()> {
+                self.inner.flush()
+            }
+            fn write_vectored(&mut self, bufs: &[IoSlice<'_>]) -> Result<usize> {
+                let written = self.inner.write_vectored(bufs)?;
+                self.count += written;
+                Ok(written)
+            }
+            fn write_all(&mut self, buf: &[u8]) -> Result<()> {
+                self.count += buf.len();
+                self.inner.write_all(buf)
+            }
+            fn write_fmt(&mut self, _fmt: Arguments<'_>) -> Result<()> {
+                unimplemented!()
+            }
         }
     }
 }

timely/src/dataflow/operators/core/partition.rs

@@ -12,7 +12,7 @@ pub trait Partition<G: Scope, C: Container> {
     /// Produces `parts` output streams, containing records produced and assigned by `route`.
     ///
     /// # Examples
-    /// ```ignore
+    /// ```
     /// use timely::dataflow::operators::ToStream;
     /// use timely::dataflow::operators::core::{Partition, Inspect};
     /// use timely_container::CapacityContainerBuilder;