Counter drops a flush once produced is drained, stranding records in an exchange #824

A two-worker dataflow input -> exchange -> probe never completes if worker 0 advances the input, sends a record, flushes, steps once, and then drops the input. Without the worker.step() it completes.

[dependencies]
timely = "=0.31.0"

use timely::container::CapacityContainerBuilder;
use timely::dataflow::operators::{Exchange, Input, Probe};
use timely::dataflow::InputHandle;

fn main() {
    // Pass any argument to step the worker once before the input closes.
    let step = std::env::args().nth(1).is_some();
    timely::execute_from_args(["-w", "2"].map(String::from).into_iter(), move |worker| {
        let mut input: InputHandle<u64, CapacityContainerBuilder<Vec<(u64, u64)>>> =
            InputHandle::new();
        worker.dataflow::<u64, _, _>(|scope| {
            scope.input_from(&mut input).exchange(|x: &(u64, u64)| x.0).probe();
        });
        if worker.index() == 0 {
            input.advance_to(1);
            input.send((0, 0));
            input.flush();
            if step {
                worker.step();
            }
        }
        // `input` drops here, closing the epoch; the worker then runs the
        // dataflow to completion.
    })
    .unwrap();
    println!("done");
}

Expected: both runs print done.

Actual: cargo run prints done. cargo run -- step never returns.

This reproduces with timely = "=0.31.0" from crates.io and with a path dependency on current master (3012f8f).

Cause

Counter::push forwards a flush (None) only while the shared produced batch is non-empty:

https://github.com/TimelyDataflow/timely-dataflow/blob/timely-v0.31.0/timely/src/dataflow/channels/pushers/counter.rs#L25-L28

The Counter does not own that batch. The operator drains it each time it is scheduled; for an input, that happens in Operator::schedule:

https://github.com/TimelyDataflow/timely-dataflow/blob/timely-v0.31.0/timely/src/dataflow/operators/core/input.rs#L198-L203

The Exchange pusher behind the Counter buffers records per peer and sends a partly filled buffer only when the time changes or a None arrives:

https://github.com/TimelyDataflow/timely-dataflow/blob/timely-v0.31.0/timely/src/dataflow/channels/pushers/exchange.rs#L138-L147

So in the program above:

1. flush pushes the record through the Counter. produced holds (1, +1) and the exchange buffers the record.
2. worker.step() schedules the input operator (it was activated by advance_to), which drains produced into its shared progress and so reports the +1.
3. Dropping the input runs close_epoch, which calls done() on the Counter. produced is empty, so the Counter swallows the None, and the exchange never sends its buffer.

The produced count has been announced, but the record it counts is never delivered, so the frontier downstream of the exchange cannot pass time 1. If the input is advanced instead of dropped, the record is held until the next send at a later time.

Suggested fix

Have the Counter remember that it pushed data since the last flush it forwarded, instead of relying only on the shared batch (diff against master):

diff --git a/timely/src/dataflow/channels/pushers/counter.rs b/timely/src/dataflow/channels/pushers/counter.rs
index 60064af2..1e3a6cf9 100644
--- a/timely/src/dataflow/channels/pushers/counter.rs
+++ b/timely/src/dataflow/channels/pushers/counter.rs
@@ -13,6 +13,8 @@ use crate::Accountable;
 pub struct Counter<T, P> {
     pushee: P,
     produced: Rc<RefCell<ChangeBatch<T>>>,
+    /// Set when a message is pushed; cleared when a flush is forwarded.
+    dirty: bool,
 }
 
 impl<T: Clone+Ord, C: Accountable, P> Push<Message<T, C>> for Counter<T, P> where P: Push<Message<T, C>> {
@@ -23,10 +25,12 @@ impl<T: Clone+Ord, C: Accountable, P> Push<Message<T, C>> for Counter<T, P> wher
             for time in message.stamp.iter() {
                 produced.update(time.clone(), message.data.record_count());
             }
+            self.dirty = true;
         }
 
         // only propagate `None` if dirty (indicates flush)
-        if message.is_some() || !self.produced.borrow_mut().is_empty() {
+        if message.is_some() || self.dirty || !self.produced.borrow_mut().is_empty() {
+            if message.is_none() { self.dirty = false; }
             self.pushee.push(message);
         }
     }
@@ -38,6 +42,7 @@ impl<T, P> Counter<T, P> where T : Ord+Clone+'static {
         Counter {
             pushee,
             produced: Rc::new(RefCell::new(ChangeBatch::new())),
+            dirty: false,
         }
     }
     /// A references to shared changes in counts, for cloning or draining.

This forwards every flush the current code forwards, plus the one after a drain. With this change on master, the program prints done in both runs, and cargo test -p timely passes.

We found this with a deterministic simulator while testing a C port of timely.