TimelyDataflow / TimelyDataflow/timely-dataflow

Clarify relationship of `push_into` and `Accountable`

Open
#706 0 comments 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

Dominant language
Rust
Stars
3.6k
Forks
293
Avg merge
14h 46m
Merged PRs (30d)
4

Description

At the moment, we have a hand-wavy relationship between pushing records at containers and the effect on progress tracking. Specifically, we need to ensure that a set of containers has the same summed record count as the containers after an exchange channel. Progress tracking tracks the record count at the output of an operator before a tee, not after.

Vectors uphold the property that their length is equal to the number of records pushed, so they implicitly uphold the guarantee that the records count after an exchange is the same count as before summed across all produced containers.

However, it is easy to implement a container that does not have this property. Consider implementing the container traits for HashMap:

diff --git i/container/src/lib.rs w/container/src/lib.rs
index 1b12c488..576a35b0 100644
--- i/container/src/lib.rs
+++ w/container/src/lib.rs
@@ -224,6 +224,49 @@ impl<T: Clone> PushInto<&&T> for Vec<T> {
     }
 }
 
+mod hash_map {
+    use std::collections::HashMap;
+    use std::hash::Hash;
+
+    use super::*;
+
+    impl<K, V> Accountable for HashMap<K, V> {
+        #[inline] fn record_count(&self) -> i64 { i64::try_from(HashMap::len(self)).unwrap() }
+        #[inline] fn is_empty(&self) -> bool { HashMap::is_empty(self) }
+    }
+
+    impl<K, V> DrainContainer for HashMap<K, V> {
+        type Item<'a> = (K, V) where Self: 'a;
+        type DrainIter<'a> = std::collections::hash_map::Drain<'a, K, V> where Self: 'a;
+        #[inline] fn drain(&mut self) -> Self::DrainIter<'_> {
+            HashMap::drain(self)
+        }
+    }
+
+    impl<K: Eq+Hash, V> SizableContainer for HashMap<K, V> {
+        fn at_capacity(&self) -> bool {
+            self.len() == self.capacity()
+        }
+        fn ensure_capacity(&mut self, stash: &mut Option<Self>) {
+            if self.capacity() == 0 {
+                *self = stash.take().unwrap_or_default();
+                self.clear();
+            }
+            let preferred = buffer::default_capacity::<(K, V)>();
+            if self.capacity() < preferred {
+                self.reserve(preferred - self.capacity());
+            }
+        }
+    }
+
+    impl<K: Eq+Hash, V> PushInto<(K, V)> for HashMap<K, V> {
+        #[inline]
+        fn push_into(&mut self, (key, value): (K, V)) {
+            self.insert(key, value);
+        }
+    }
+}
+
 mod rc {
     impl<T: crate::Accountable> crate::Accountable for std::rc::Rc<T> {
         #[inline] fn record_count(&self) -> i64 { self.as_ref().record_count() }

The implementation compiles, but HashMap::insert does not have the property that the length of the hash map equals the number of insert calls.

Contributor guide

Open the contributing guide

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Research direction

Start in container/src/lib.rs and read the Accountable, PushInto, DrainContainer, and SizableContainer implementations, then trace how progress tracking counts records before a tee and after an exchange channel. Done means the relationship between push_into and Accountable is explicitly defined and the summed record-count invariant is addressed for containers such as HashMap.

Written by the indexing model from the issue text.

Assessment

Tech stack
rust
Domain
distributed-systems
Issue type
Refactor
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Needs clarification
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.