Crash in nested scope

[ryzhyk]

I am using the Variable implementation from the dataflog.rs example to construct recursive relations. However, my program crashes any time I use a nested scope. Here is the minimal failing example. The definition of struct Variable is copied from dataflog.rs verbatim. The actual dataflow consists of collection1 and collection2, with collection2 being an exact copy of collection1.

Note that there is no recursion here, so this does not actually require a nested scope. I only use it demonstrate the failure. It still crashes if I use an actual recursive definition.

extern crate timely;
extern crate differential_dataflow;

use timely::dataflow::*;
use timely::dataflow::operators::probe;
use timely::dataflow::operators::feedback::Handle;

use differential_dataflow::input::Input;
use differential_dataflow::{Collection,Data,Hashable};
use timely::progress::nested::product::Product;
use timely::dataflow::scopes::Child;
use timely::dataflow::operators::*;
use differential_dataflow::operators::*;
use differential_dataflow::lattice::Lattice;

pub struct Variable<'a, G: Scope, D: Default+Data+Hashable>
where G::Timestamp: Lattice+Ord {
    feedback: Option<Handle<G::Timestamp, u64,(D, Product<G::Timestamp, u64>, isize)>>,
    current: Collection<Child<'a, G, u64>, D>,
    cycle: Collection<Child<'a, G, u64>, D>,
}

impl<'a, G: Scope, D: Default+Data+Hashable> Variable<'a, G, D> where G::Timestamp: Lattice+Ord {
    /// Creates a new `Variable` from a supplied `source` stream.
    pub fn from(source: &Collection<Child<'a, G, u64>, D>) -> Variable<'a, G, D> {
        let (feedback, cycle) = source.inner.scope().loop_variable(u64::max_value(), 1);
        let cycle = Collection::new(cycle);
        let mut result = Variable { feedback: Some(feedback), current: cycle.clone(), cycle: cycle };
        result.add(source);
        result
    }
    /// Adds a new source of data to the `Variable`.
    pub fn add(&mut self, source: &Collection<Child<'a, G, u64>, D>) {
        self.current = self.current.concat(source);
    }
}

impl<'a, G: Scope, D: Default+Data+Hashable> ::std::ops::Deref for Variable<'a, G, D> where G::Timestamp: Lattice+Ord {
    type Target = Collection<Child<'a, G, u64>, D>;
    fn deref(&self) -> &Self::Target {
        &self.cycle
    }
}

impl<'a, G: Scope, D: Default+Data+Hashable> Drop for Variable<'a, G, D> where G::Timestamp: Lattice+Ord {
    fn drop(&mut self) {
        if let Some(feedback) = self.feedback.take() {
            self.current.distinct()
                        .inner
                        .connect_loop(feedback);
        }
    }
}

fn main() {
    timely::execute_from_args(std::env::args(), move |worker| {
        
        let mut probe = probe::Handle::new();
        let mut input1 = worker.dataflow(|scope| {

            let (input1, collection1) = scope.new_collection();
            let (_, collection2) = scope.new_collection();

            let collection2 = scope.scoped(|inner| {
                let var1 = Variable::from(&collection1.enter(inner));
                let mut var2 = Variable::from(&collection2.enter(inner));

                var2.add(&var1);
                var2.leave()
            });

            collection2.inspect(|x| println!("\t{:?}", x))
                       .probe_with(&mut probe);

            input1
        });

        input1.insert(1);

        input1.advance_to(1); input1.flush();
        worker.step_while(|| probe.less_than(input1.time()));
    }).unwrap();
}

Here is the stack trace:

thread 'worker thread 0' panicked at 'failed to find index', libcore/option.rs:916:5
note: Some details are omitted, run with `RUST_BACKTRACE=full` for a verbose backtrace.
stack backtrace:
   0: std::sys::unix::backtrace::tracing::imp::unwind_backtrace
             at libstd/sys/unix/backtrace/tracing/gcc_s.rs:49
   1: std::sys_common::backtrace::print
             at libstd/sys_common/backtrace.rs:71
             at libstd/sys_common/backtrace.rs:59
   2: std::panicking::default_hook::{{closure}}
             at libstd/panicking.rs:207
   3: std::panicking::default_hook
             at libstd/panicking.rs:223
   4: std::panicking::rust_panic_with_hook
             at libstd/panicking.rs:402
   5: std::panicking::begin_panic_fmt
             at libstd/panicking.rs:349
   6: rust_begin_unwind
             at libstd/panicking.rs:325
   7: core::panicking::panic_fmt
             at libcore/panicking.rs:72
   8: core::option::expect_failed
             at libcore/option.rs:916
   9: <core::option::Option<T>>::expect
             at /checkout/src/libcore/option.rs:302
  10: <differential_dataflow::operators::group::history_replay::HistoryReplayer<'a, V1, V2, T, R1, R2> as differential_dataflow::operators::group::PerKeyCompute<'a, V1, V2, T, R1, R2>>::compute
             at ./src/operators/group.rs:840
  11: <differential_dataflow::operators::arrange::Arranged<G, K, V, R, T1> as differential_dataflow::operators::group::GroupArranged<G, K, V, R>>::group_arranged::{{closure}}
             at ./src/operators/group.rs:425
  12: <timely::dataflow::stream::Stream<G, D1> as timely::dataflow::operators::generic::unary::Unary<G, D1>>::unary_notify::{{closure}}::{{closure}}
             at /home/lryzhyk/.cargo/git/checkouts/timely-dataflow-7b255df3956d217b/e741c1d/src/dataflow/operators/generic/unary.rs:94
  13: <timely::dataflow::stream::Stream<G, D1> as timely::dataflow::operators::generic::operator::Operator<G, D1>>::unary_frontier::{{closure}}::{{closure}}
             at /home/lryzhyk/.cargo/git/checkouts/timely-dataflow-7b255df3956d217b/e741c1d/src/dataflow/operators/generic/operator.rs:337
  14: <timely::dataflow::operators::generic::builder_rc::OperatorBuilder<G>>::build::{{closure}}
             at /home/lryzhyk/.cargo/git/checkouts/timely-dataflow-7b255df3956d217b/e741c1d/src/dataflow/operators/generic/builder_rc.rs:128
  15: <timely::dataflow::operators::generic::builder_raw::OperatorCore<T, PEP, PIP> as timely::progress::operate::Operate<T>>::pull_internal_progress
             at /home/lryzhyk/.cargo/git/checkouts/timely-dataflow-7b255df3956d217b/e741c1d/src/dataflow/operators/generic/builder_raw.rs:211
  16: <timely::progress::nested::subgraph::PerOperatorState<T>>::exchange_progress
             at /home/lryzhyk/.cargo/git/checkouts/timely-dataflow-7b255df3956d217b/e741c1d/src/progress/nested/subgraph.rs:802
  17: <timely::progress::nested::subgraph::Subgraph<TOuter, TInner> as timely::progress::operate::Operate<TOuter>>::pull_internal_progress
             at /home/lryzhyk/.cargo/git/checkouts/timely-dataflow-7b255df3956d217b/e741c1d/src/progress/nested/subgraph.rs:542
  18: <timely::progress::nested::subgraph::PerOperatorState<T>>::exchange_progress
             at /home/lryzhyk/.cargo/git/checkouts/timely-dataflow-7b255df3956d217b/e741c1d/src/progress/nested/subgraph.rs:802
  19: <timely::progress::nested::subgraph::Subgraph<TOuter, TInner> as timely::progress::operate::Operate<TOuter>>::pull_internal_progress
             at /home/lryzhyk/.cargo/git/checkouts/timely-dataflow-7b255df3956d217b/e741c1d/src/progress/nested/subgraph.rs:542
  20: timely::dataflow::scopes::root::Wrapper::step::{{closure}}
             at /home/lryzhyk/.cargo/git/checkouts/timely-dataflow-7b255df3956d217b/e741c1d/src/dataflow/scopes/root.rs:159
  21: <core::option::Option<T>>::map
             at /checkout/src/libcore/option.rs:404
  22: timely::dataflow::scopes::root::Wrapper::step
             at /home/lryzhyk/.cargo/git/checkouts/timely-dataflow-7b255df3956d217b/e741c1d/src/dataflow/scopes/root.rs:159
  23: <timely::dataflow::scopes::root::Root<A>>::step
             at /home/lryzhyk/.cargo/git/checkouts/timely-dataflow-7b255df3956d217b/e741c1d/src/dataflow/scopes/root.rs:46
  24: <timely::dataflow::scopes::root::Root<A>>::step_while
             at /home/lryzhyk/.cargo/git/checkouts/timely-dataflow-7b255df3956d217b/e741c1d/src/dataflow/scopes/root.rs:59
  25: example::main::{{closure}}
             at examples/example.rs:81
  26: timely::execute::execute_logging::{{closure}}
             at /home/lryzhyk/.cargo/git/checkouts/timely-dataflow-7b255df3956d217b/e741c1d/src/execute.rs:152
  27: timely_communication::initialize::initialize::{{closure}}
             at /home/lryzhyk/.cargo/registry/src/github.1485827954.workers.dev-1ecc6299db9ec823/timely_communication-0.5.0/src/initialize.rs:167
thread 'main' panicked at 'called `Result::unwrap()` on an `Err` value: Any', libcore/result.rs:945:5
stack backtrace:
   0: std::sys::unix::backtrace::tracing::imp::unwind_backtrace
             at libstd/sys/unix/backtrace/tracing/gcc_s.rs:49
   1: std::sys_common::backtrace::print
             at libstd/sys_common/backtrace.rs:71
             at libstd/sys_common/backtrace.rs:59
   2: std::panicking::default_hook::{{closure}}
             at libstd/panicking.rs:207
   3: std::panicking::default_hook
             at libstd/panicking.rs:223
   4: std::panicking::rust_panic_with_hook
             at libstd/panicking.rs:402
   5: std::panicking::begin_panic_fmt
             at libstd/panicking.rs:349
   6: rust_begin_unwind
             at libstd/panicking.rs:325
   7: core::panicking::panic_fmt
             at libcore/panicking.rs:72
   8: core::result::unwrap_failed
             at /checkout/src/libcore/macros.rs:26
   9: <core::result::Result<T, E>>::unwrap
             at /checkout/src/libcore/result.rs:782
  10: <timely_communication::initialize::WorkerGuards<T> as core::ops::drop::Drop>::drop
             at /home/lryzhyk/.cargo/registry/src/github.1485827954.workers.dev-1ecc6299db9ec823/timely_communication-0.5.0/src/initialize.rs:192
  11: core::ptr::drop_in_place
             at /checkout/src/libcore/ptr.rs:59
  12: example::main
             at examples/example.rs:56
  13: std::rt::lang_start::{{closure}}
             at /checkout/src/libstd/rt.rs:74
  14: std::panicking::try::do_call
             at libstd/rt.rs:59
             at libstd/panicking.rs:306
  15: __rust_maybe_catch_panic
             at libpanic_unwind/lib.rs:102
  16: std::rt::lang_start_internal
             at libstd/panicking.rs:285
             at libstd/panic.rs:361
             at libstd/rt.rs:58
  17: std::rt::lang_start
             at /checkout/src/libstd/rt.rs:74
  18: main
  19: __libc_start_main
  20: _start

[frankmcsherry]

Whoa, that is a pretty minimal repro. Let me take a look today and I'll try and track down the issue.

[frankmcsherry]

Ok, I think I have the answer. The high-level bit is that about a year ago differential dataflow changed, and examples/dataflog.rs did not track the change correctly.

You are using timely dataflow loop construction, not differential dataflow loop construction, and there are a few (one? I think) additional things you have to do to maintain differential dataflow's invariants. Specifically, as the data go around the loop, you need to increment the corresponding coordinate of the time component of the (data, time, diff) triple. This field needs to be greater or equal to the associated timely message timestamp, and while that is incremented as you go around the loop, the time field is just data and is not incremented automatically.

In the current iterate.rs this looks like so:

                   .inner
                   .map(|(x,t,d)| (x, Product::new(t.outer, t.inner+1), d))
                   .connect_loop(self.feedback);

If I put this into your example, as

impl<'a, G: Scope, D: Default+Data+Hashable> Drop for Variable<'a, G, D> where G::Timestamp: Lattice+Ord {
    fn drop(&mut self) {
        if let Some(feedback) = self.feedback.take() {
            self.current.distinct()
                        .inner
                        .map(|(x,t,d)| (x, Product::new(t.outer, t.inner+1), d))
                        .connect_loop(feedback);
        }
    }
}

the example seems to run fine.

[frankmcsherry]

Btw, pushed a change to dataflog.rs and marked as "fixed", but totally re-open if you think this is unacceptable. :D

[ryzhyk]

Thank you so much for the quick fix! I confirm that the fix also works in the context of the more complex program where I originally ran into this.