Enter leave test failing

Cargo.toml

@@ -13,7 +13,8 @@ resolver = "2"
 edition = "2021"
 
 [workspace.dependencies]
-columnar = "0.3"
+# columnar = "0.3"
+columnar = { path = "../columnar" }
 
 [profile.release]
 opt-level = 3

communication/src/allocator/mod.rs

@@ -114,13 +114,13 @@ struct Broadcaster<T> {
 
 impl<T: Clone> Push<T> for Broadcaster<T> {
     fn push(&mut self, element: &mut Option<T>) {
-        // Push defensive copies to pushers after the first.
-        for pusher in self.pushers.iter_mut().skip(1) {
-            self.spare.clone_from(element);
-            pusher.push(&mut self.spare);
-        }
+        // // Push defensive copies to pushers after the first.
+        // for pusher in self.pushers.iter_mut().skip(1) {
+        //     self.spare.clone_from(element);
+        //     pusher.push(&mut self.spare);
+        // }
         // Push the element itself at the first pusher.
-        for pusher in self.pushers.iter_mut().take(1) {
+        for pusher in self.pushers.iter_mut() {//}.take(1) {
             pusher.push(element);
         }
     }

communication/src/allocator/thread.rs

@@ -94,12 +94,12 @@ impl<T> Pull<T> for Puller<T> {
     #[inline]
     fn pull(&mut self) -> &mut Option<T> {
         let mut borrow = self.source.borrow_mut();
-        // if let Some(element) = self.current.take() {
-        //     // TODO : Arbitrary constant.
-        //     if borrow.1.len() < 16 {
-        //         borrow.1.push_back(element);
-        //     }
-        // }
+        if let Some(element) = self.current.take() {
+            // TODO : Arbitrary constant.
+            if borrow.1.len() < 16 {
+                borrow.1.push_back(element);
+            }
+        }
         self.current = borrow.0.pop_front();
         &mut self.current
     }

container/src/lib.rs

@@ -165,6 +165,10 @@ impl<T, C: SizableContainer + PushInto<T>> PushInto<T> for CapacityContainerBuil
         // Maybe flush
         if self.current.at_capacity() {
             self.pending.push_back(std::mem::take(&mut self.current));
+            if let Some(spare) = self.empty.take() {
+                self.current = spare;
+                self.current.clear();
+            }
         }
     }
 }
@@ -186,9 +190,12 @@ impl<C: Container + Clone + 'static> ContainerBuilder for CapacityContainerBuild
     fn finish(&mut self) -> Option<&mut C> {
         if !self.current.is_empty() {
             self.pending.push_back(std::mem::take(&mut self.current));
+            if let Some(spare) = &mut self.empty {
+                std::mem::swap(&mut self.current, spare);
+                self.current.clear();
+            }
         }
-        self.empty = self.pending.pop_front();
-        self.empty.as_mut()
+        self.extract()
     }
 }
 

timely/Cargo.toml

@@ -19,6 +19,7 @@ default = ["getopts"]
 getopts = ["getopts-dep", "timely_communication/getopts"]
 
 [dependencies]
+bytemuck = "1.18.0"
 columnar = { workspace = true }
 getopts-dep = { package = "getopts", version = "0.2.21", optional = true }
 bincode = { version = "1.0" }
@@ -32,5 +33,4 @@ crossbeam-channel = "0.5"
 smallvec = { version = "1.13.2", features = ["serde", "const_generics"] }
 
 [dev-dependencies]
-bytemuck = "1.18.0"
 rand = { version = "0.8", features = ["small_rng"] }

timely/examples/columnar.rs

@@ -1,7 +1,7 @@
 //! Wordcount based on flatcontainer.
 
 use {
-    std::collections::HashMap,
+    std::collections::BTreeMap,
     timely::{Container, container::CapacityContainerBuilder},
     timely::dataflow::channels::pact::{ExchangeCore, Pipeline},
     timely::dataflow::InputHandleCore,
@@ -21,8 +21,6 @@ fn main() {
 
     type Container = Column<WordCount>;
 
-    use columnar::Len;
-
     let config = timely::Config {
         communication: timely::CommunicationConfig::ProcessBinary(3),
         worker: timely::WorkerConfig::default(),
@@ -37,9 +35,7 @@ fn main() {
         worker.dataflow::<usize, _, _>(|scope| {
             input
                 .to_stream(scope)
-                .unary(
-                    Pipeline,
-                    "Split",
+                .unary(Pipeline, "Split",
                     |_cap, _info| {
                         move |input, output| {
                             while let Some((time, data)) = input.next() {
@@ -58,38 +54,33 @@ fn main() {
                     ExchangeCore::<ColumnBuilder<WordCount>,_>::new_core(|x: &WordCountReference<&str,&i64>| x.text.len() as u64),
                     "WordCount",
                     |_capability, _info| {
-                        let mut queues = HashMap::new();
-                        let mut counts = HashMap::new();
+                        let mut queues = Vec::new();
+                        let mut counts = BTreeMap::new();
 
                         move |input, output| {
                             while let Some((time, data)) = input.next() {
-                                queues
-                                    .entry(time.retain())
-                                    .or_insert(Vec::new())
-                                    .push(std::mem::take(data));
+                                queues.push((time.retain(), std::mem::take(data)));
                             }
 
                             for (key, val) in queues.iter_mut() {
                                 if !input.frontier().less_equal(key.time()) {
                                     let mut session = output.session(key);
-                                    for batch in val.drain(..) {
-                                        for wordcount in batch.iter() {
-                                            let total =
-                                            if let Some(count) = counts.get_mut(wordcount.text) {
-                                                *count += wordcount.diff;
-                                                *count
-                                            }
-                                            else {
-                                                counts.insert(wordcount.text.to_string(), *wordcount.diff);
-                                                *wordcount.diff
-                                            };
-                                            session.give(WordCountReference { text: wordcount.text, diff: total });
+                                    for wordcount in val.iter() {
+                                        let total =
+                                        if let Some(count) = counts.get_mut(wordcount.text) {
+                                            *count += wordcount.diff;
+                                            *count
                                         }
+                                        else {
+                                            counts.insert(wordcount.text.to_string(), *wordcount.diff);
+                                            *wordcount.diff
+                                        };
+                                        session.give(WordCountReference { text: wordcount.text, diff: total });
                                     }
                                 }
                             }
 
-                            queues.retain(|_key, val| !val.is_empty());
+                            queues.retain(|(key, _val)| input.frontier().less_equal(key.time()) );
                         }
                     },
                 )
@@ -99,13 +90,35 @@ fn main() {
         });
 
         // introduce data and watch!
-        for round in 0..10 {
-            input.send(WordCountReference { text: "flat container", diff: 1 });
-            input.advance_to(round + 1);
-            while probe.less_than(input.time()) {
-                worker.step();
+
+        if let Some(filename) = std::env::args().nth(1) {
+            let file = std::fs::File::open(filename).unwrap();
+            use std::io::BufRead;
+            let mut lines = std::io::BufReader::new(file);
+            let mut text = String::default();
+            let mut index = 0;
+            while lines.read_line(&mut text).unwrap() > 0 {
+                if index % worker.peers() == worker.index() {
+                    input.send(WordCountReference { text: &text, diff: 1 });
+                }
+                text.clear();
+                input.advance_to(index + 1);
+                while probe.less_than(input.time()) {
+                    worker.step();
+                }
+                index += 1;
             }
         }
+        else {
+            for round in 0.. {
+                input.send(WordCountReference { text: "flat container", diff: 1 });
+                input.advance_to(round + 1);
+                while probe.less_than(input.time()) {
+                    worker.step();
+                }
+            }
+        }
+
     })
     .unwrap();
 }
@@ -177,19 +190,19 @@ mod container {
         type Iter<'a> = IterOwn<<C::Container as columnar::Container<C>>::Borrowed<'a>>;
         fn iter<'a>(&'a self) -> Self::Iter<'a> {
             match self {
-                Column::Typed(t) => t.borrow().into_iter(),
-                Column::Bytes(b) => <<C::Container as columnar::Container<C>>::Borrowed<'a> as FromBytes>::from_bytes(&mut Indexed::decode(bytemuck::cast_slice(b))).into_iter(),
-                Column::Align(a) => <<C::Container as columnar::Container<C>>::Borrowed<'a> as FromBytes>::from_bytes(&mut Indexed::decode(a)).into_iter(),
+                Column::Typed(t) => t.borrow().into_index_iter(),
+                Column::Bytes(b) => <<C::Container as columnar::Container<C>>::Borrowed<'a> as FromBytes>::from_bytes(&mut Indexed::decode(bytemuck::cast_slice(b))).into_index_iter(),
+                Column::Align(a) => <<C::Container as columnar::Container<C>>::Borrowed<'a> as FromBytes>::from_bytes(&mut Indexed::decode(a)).into_index_iter(),
             }
         }
 
         type Item<'a> = C::Ref<'a>;
         type DrainIter<'a> = IterOwn<<C::Container as columnar::Container<C>>::Borrowed<'a>>;
         fn drain<'a>(&'a mut self) -> Self::DrainIter<'a> {
             match self {
-                Column::Typed(t) => t.borrow().into_iter(),
-                Column::Bytes(b) => <<C::Container as columnar::Container<C>>::Borrowed<'a> as FromBytes>::from_bytes(&mut Indexed::decode(bytemuck::cast_slice(b))).into_iter(),
-                Column::Align(a) => <<C::Container as columnar::Container<C>>::Borrowed<'a> as FromBytes>::from_bytes(&mut Indexed::decode(a)).into_iter(),
+                Column::Typed(t) => t.borrow().into_index_iter(),
+                Column::Bytes(b) => <<C::Container as columnar::Container<C>>::Borrowed<'a> as FromBytes>::from_bytes(&mut Indexed::decode(bytemuck::cast_slice(b))).into_index_iter(),
+                Column::Align(a) => <<C::Container as columnar::Container<C>>::Borrowed<'a> as FromBytes>::from_bytes(&mut Indexed::decode(a)).into_index_iter(),
             }
         }
     }
@@ -327,9 +340,12 @@ mod builder {
         fn finish(&mut self) -> Option<&mut Self::Container> {
             if !self.current.is_empty() {
                 self.pending.push_back(Column::Typed(std::mem::take(&mut self.current)));
+                if let Some(Column::Typed(spare)) = self.empty.take() {
+                    self.current = spare;
+                    self.current.clear();
+                }
             }
-            self.empty = self.pending.pop_front();
-            self.empty.as_mut()
+            self.extract()
         }
     }
 

timely/src/lib.rs

@@ -145,6 +145,7 @@ mod encoding {
     // We will pad out anything we write to make the result `u64` aligned.
     impl<T: Data> Bytesable for Bincode<T> {
         fn from_bytes(bytes: Bytes) -> Self {
+            println!("Bincode::from_bytes: {:?}", std::any::type_name::<T>());
             let typed = ::bincode::deserialize(&bytes[..]).expect("bincode::deserialize() failed");
             let typed_size = ::bincode::serialized_size(&typed).expect("bincode::serialized_size() failed") as usize;
             assert_eq!(bytes.len(), (typed_size + 7) & !7);