Introduce fate #137

davidselassie · 2022-09-20T20:32:44Z

Previously I made the questionable API design of having
StatefulLogic::snapshot return a StateUpdate. This meant that the
snapshotting process affected the behavior of the stateful operator,
since it could return Reset to signal up to StatefulUnary to
discard the logic.

This unwinds that tangle by introducing a new method with a perhaps
overly poetic name StatefulLogic::fate which should return what
StatefulUnary should do with the logic when it's done processing via
a LogicFate enum. There are three options:

Retain it until a new item for this key comes in.
Retain it and awaken it after a timeout.
Discard it.

fate is attempting to encapsulate the problem of StatefulUnary is
the owner of the logics, so they can't drop themselves. And since
awakening timeouts are part of that process, they're in there too.

This is nice because it simplifies the return value of exec to just
output. The awaken delay is handled in fate.

It also uses this pattern within Windower::fate with WindowerFate
for the same kind of problem: the WindowStatefulLogic is the owner,
and we need to communicate back when it's safe to discard a
Windower. This fixes the bug of never discarding window state if a
key is never seen again: we now discard that state whenever all
windows for a key are closed.

A few other small changes:

Standardises on the language of "awaken" a logic. Still uses
Timely's "activate" for a Timely operator, though. Renames
StatefulLogic::exec to StatefulLogic::awake_with to make more
explicit when it is called. Renames WindowLogic::exec to
WindowLogic::with_next to make explicit when it's called.
All stateful operator tests should test recovery as part of testing
the logic. We should do this to excercise the serde round-trip. I
found three? bugs via this where recovery would panic because the
deserialization wasn't to the correct type. As part of this, I added
explicit type annotations to all StateBytes::ser and
StateBytes::de calls so you can compare them.
Clarifies more comments in the giant chunk of code for
StatefulUnary::stateful_unary. I was also able to optimize it a
little and get rid of two of our temporary buffers. I think it makes
the process slightly clearer.

davidselassie · 2022-09-20T20:40:00Z

This is going to definitely clash with #115 and #127 so I'm happy to wait until those are merged to figure this out.

Although I believe this fixes a bug where any dataflow would panic on recovery, so perhaps worth doing now?

davidselassie · 2022-09-20T23:18:11Z

~~I'm actually going to poke at this a little further because I think persisting awake times per key is pretty closely tied to this. You can review if you want, but I'm going to add some new commits.~~

Actually actually, this is in the right direction, and I need to stop making +3000 line PRs so I do want this reviewed and merged and can keep iterating on it.

blakestier · 2022-09-23T18:15:09Z

pytests/test_operators.py

+    # Recover
+    run_main(flow, epoch_config=epoch_config, recovery_config=recovery_config)
+
+    # But it remembers the first two items in the first window.


Thanks for the comments. So helpful

blakestier · 2022-09-23T18:23:56Z

pytests/test_operators.py

+def test_fold_window(recovery_config):
+    flow = Dataflow()
+
+    # Remember that clocks are built per-key so the `TestingClock` in


Probably a smell but let's table that for now 🤔

This is sort of changed in #139.

Not the per-key thing, but it'll be less confusing? Although I'm having some trouble with the TestingClockConfig in another PR, so the smell does remain...

I figured out what was confusing me. It's that we were recovering and using a half-consumed generator. #144 has the tests out of this working.

src/operators/fold_window.rs

src/recovery/mod.rs

blakestier · 2022-09-23T19:02:57Z

src/recovery/mod.rs

@@ -778,7 +797,7 @@ pub(crate) fn build_state_loading_dataflow<A: Allocate>(

                                match update {
                                    StateUpdate::Upsert(state) => resume_state.insert(key, state),
-                                    StateUpdate::Reset => resume_state.remove(&key),
+                                    StateUpdate::Discard => resume_state.remove(&key),


When i read this, I keep expecting the logic to be something like "Is there an updated state? great, add it to the stateful collection. Otherwise, discard it". Otherwise, I'm thinking there would be a 3rd possibility of no state update? I think it's the Update that's confusing me. Like it's either a State::Upsert(state) or a State::Discard but maybe the word State is too loaded to stand alone

On the writing side, we could eventually add a StateUpdate::Unchanged return type and then not do the write, but that would be an optimization. The trade-off is that now each logic doesn't just need to remember its state, it also needs to remember the last state it wrote (or some proxy for it) so it can look at itself when asked for a snapshot and actually determine there was no state change relative to the last snapshot. That really increases the bug surface area here as you're introducing the concept of epochs (in a minor way) into each stateful logic, whereas the more basic "just snapshot me!" approach lets the logic writer ignore that.

FWIW on the reading side (since this linked snippet is the loading during recovery), that situation is already handled: if we didn't write the update, then there wasn't an update, so don't do anything.

blakestier · 2022-09-23T19:20:17Z

src/window/tumbling_window.rs

-            .map(|t| t - watermark)
-            .min()
+    fn fate(&self) -> WindowerFate {
+        if let Some(next_close) = self.close_times.values().cloned().min() {


why did i think these were sorted already? 🤔

Maybe you're remembering that we use BTreeMap elsewhere? We can't here because we need to order by value, not key. We could cache the min close time, but I don't know that it would help that much.

blakestier

I left some comments around names to highlight places that felt a little sticky for me, but I think this is a great change and I appreciate the tests, the untangling, and the dramatic introduction of fate

Makes all types for recovery serialization and deserialization explicit and fixes them to match each other for each recoverable operator. This will fix panics during recovery loading. This was also fixed in #137 but I'm breaking it out here separately because I keep discoverying smaller recovery bugs and am fixing them separately.

Previously I made the questionable API design of having `StatefulLogic::snapshot` return a `StateUpdate`. This meant that the snapshotting process affected the behavior of the stateful operator, since it could return `Reset` to signal up to `StatefulUnary` to discard the logic. This unwinds that tangle by introducing a new method with a perhaps overly poetic name `StatefulLogic::fate` which should return what `StatefulUnary` should do with the logic when it's done processing via a `LogicFate` enum: either `Retain` it or `Discard` it. `fate` is attempting to encapsulate the problem of `StatefulUnary` is the owner of the logics, so they can't drop themselves. This is nice because it simplifies the return value of `exec` to just output. The awaken delay is handled in `fate`. The other part of the logic return value was "time to next awake". Which is now encapsulated in `StatefulLogic::next_awake`. It also breaks apart the results of `Windower` due to the same kind of problem: the `WindowStatefulLogic` is the owner, and we need to communicate back when it's safe to discard a `Windower`. Adds `Windower::is_empty` and `Windower::next_close` to handle this. This fixes the bug of never discarding window state if a key is never seen again: we now discard that state whenever all windows for a key are closed. A few other small changes: - Standardises on the language of "awaken" a logic. Still uses Timely's "activate" for a Timely operator, though. Renames `StatefulLogic::exec` to `StatefulLogic::on_awake` to make more explicit when it is called. Renames `WindowLogic::exec` to `WindowLogic::with_next` to make explicit when it's called. - Clarifies more comments in the giant chunk of code for `StatefulUnary::stateful_unary`. I was also able to optimize it a little and get rid of two of our temporary buffers. I think it makes the process slightly clearer.

davidselassie · 2022-09-27T21:53:09Z

Rebased this.

The changes from the #143 and #144 are thus no longer in this PR.

Broke out LogicFate::AwakeAfter into its own function StatefulLogic::next_awake so that the "do I keep this around" question is totally separate from the "when do I wake up next" question.

In debugging that I think we're going to have to take an overhaul of how system time works: currently because there are interactions between the times returned by the clock in the window operators and the system time used by the Timely scheduler, we can't really deterministically unit test system time still. This is tough because the behavior is determined by when some code runs / windows close. I think this looks like using the Clock in StatefulUnary for run time and assuming the Timely scheduler activation delays will match up, but I'm not sure.

davidselassie requested review from blakestier, Psykopear and whoahbot and removed request for blakestier, Psykopear and whoahbot September 20, 2022 20:51

whoahbot force-pushed the serde-separate branch from 5656db9 to 57138f8 Compare September 20, 2022 23:09

davidselassie marked this pull request as draft September 20, 2022 23:17

davidselassie marked this pull request as ready for review September 21, 2022 19:03

whoahbot approved these changes Sep 21, 2022

View reviewed changes

blakestier reviewed Sep 23, 2022

View reviewed changes

src/operators/fold_window.rs Show resolved Hide resolved

blakestier reviewed Sep 23, 2022

View reviewed changes

src/recovery/mod.rs Outdated Show resolved Hide resolved

blakestier reviewed Sep 23, 2022

View reviewed changes

src/recovery/mod.rs Show resolved Hide resolved

blakestier reviewed Sep 23, 2022

View reviewed changes

blakestier approved these changes Sep 23, 2022

View reviewed changes

This was referenced Sep 26, 2022

Fixes recovery serde #143

Merged

Better recovery tests #144

Merged

davidselassie force-pushed the serde-separate branch from 57138f8 to e7e4a12 Compare September 27, 2022 21:38

davidselassie force-pushed the serde-separate branch from e7e4a12 to a26045a Compare September 27, 2022 21:43

davidselassie merged commit 8da9636 into main Sep 27, 2022

davidselassie deleted the serde-separate branch September 27, 2022 23:40

davidselassie mentioned this pull request Sep 27, 2022

Convert window tests to use event time #145

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Introduce fate #137

Introduce fate #137

davidselassie commented Sep 20, 2022

davidselassie commented Sep 20, 2022 •

edited

davidselassie commented Sep 20, 2022 •

edited

blakestier Sep 23, 2022

blakestier Sep 23, 2022

davidselassie Sep 26, 2022

davidselassie Sep 26, 2022

blakestier Sep 23, 2022

davidselassie Sep 26, 2022

davidselassie Sep 26, 2022

blakestier Sep 23, 2022

davidselassie Sep 26, 2022

blakestier left a comment

davidselassie commented Sep 27, 2022

Introduce fate #137

Introduce fate #137

Conversation

davidselassie commented Sep 20, 2022

davidselassie commented Sep 20, 2022 • edited

davidselassie commented Sep 20, 2022 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

blakestier left a comment

Choose a reason for hiding this comment

davidselassie commented Sep 27, 2022

davidselassie commented Sep 20, 2022 •

edited

davidselassie commented Sep 20, 2022 •

edited