timely-util: no nonsense timely/async bridge #14630

petrosagg · 2022-09-05T09:16:58Z

Motivation

This PR provides a new async/timely integration that as far as I can tell has no weird invariants that the user needs to now about and so it's hopefully much harder to misuse than the previous build_async method.

The piece that makes this possible is the AsyncInputHandle type that correctly wires up async Wakers and a manual Future implementation that allows users to naturally call input.next().await on a handle and/or safely use these future in a select! loop.

The handle's next() method behaves like an async iterator, resolving to None when there are no more notifications to be received, and gives out an enum of Event::Data(capability, data) when there is data in the input or Event::Progress(frontier) when the frontier has advanced.

On top of the generic builder_async::OperatorBuilder this PR also implements Operator::unary_async, Operator::binary_async, Operator::sink_async and a freestanding function operator::source_async.

src/timely-util/src/builder_async.rs

guswynn

I need to read this more closely and ask more questions, i will reread it tomorrow morning

guswynn · 2022-09-07T19:10:48Z

src/timely-util/src/builder_async.rs

+
+/// An async Waker that activates a specific operator when woken and marks the task as ready
+struct TimelyWaker {
+    activator: SyncActivator,


my understanding is that if Waker didnt require Send then this could be a normal activator?

Yes, exactly. We do some trickery in the future were we also carry around a normal activator plus the thread id that it belongs to and dispatch to the right one based on thread id. This is what gloomio does

src/timely-util/src/builder_async.rs

guswynn · 2022-09-07T19:27:00Z

src/timely-util/src/builder_async.rs

+        let connection =
+            vec![Antichain::from_elem(Default::default()); self.builder.shape().inputs()];
+        self.new_output_connection(connection)


just call self.builder.new_output(), to make it more clear that for output we just forward on

guswynn · 2022-09-07T19:30:04Z

src/timely-util/src/builder_async.rs

+                // The timely operator needs to be kept alive if the task is pending or if there
+                // are live input handles, perhaps in an another executor


if the logic future is dead, then what happens if their are live input handles? nothing will be able to read them, right?

It was supposed to support a usecase where you've sent the handles to another executor but the original operator has terminated. I removed the input_handle logic and now just panic if a handle is used after the operator has died

guswynn · 2022-09-08T20:03:40Z

src/storage/src/source/persist_source.rs

                    let consumed_part_cap = cap.delayed_for_output(cap.time(), consumed_part_port);
-                    let mut consumed_part_session =
-                        consumed_part_output_handle.session(&consumed_part_cap);
+                    drop(cap);


is this strictly required?

guswynn · 2022-09-08T20:04:42Z

src/storage/src/source/source_reader_pipeline.rs

                    new_ts_upper
                );

+                let cap = cap_set.delayed(cap_set.first().unwrap());


similarly, is this just to avoid holding onto the cap as we wait for the advance future to complete?

src/timely-util/src/operator.rs

src/timely-util/src/builder_async.rs

guswynn

Ok I think this is reasonable, but 3 things need to be documented:

Why do we need shared_frontiers? Its so we can hold frontierref's across await points, i think
Why are the transmute's okay? Link to the tokio docs
- Also followup: make a miri test if that even possible at all
How the reactor works

As well as:

follow-up on where we actually need SeqCst
and I have a few more nits i added!

src/timely-util/src/builder_async.rs

guswynn · 2022-09-09T20:31:18Z

src/timely-util/src/builder_async.rs

+    /// The index of this handle into the shared frontiers vector
+    index: usize,
+    /// Reference to the reactor queue of this input handle where Wakers can be registered
+    reactor_registry: Weak<RefCell<Vec<Waker>>>,


name this the same as registered_wakers

Also, we should document why it needs to be a Weak

src/timely-util/src/builder_async.rs

Signed-off-by: Petros Angelatos <[email protected]>

petrosagg commented Sep 5, 2022

View reviewed changes

src/timely-util/src/builder_async.rs Outdated Show resolved Hide resolved

petrosagg force-pushed the timely-async-bridge branch 3 times, most recently from fd69010 to be0c305 Compare September 5, 2022 11:34

petrosagg marked this pull request as ready for review September 5, 2022 11:37

petrosagg requested a review from aljoscha September 5, 2022 11:37

petrosagg force-pushed the timely-async-bridge branch 5 times, most recently from cf16ed3 to 88f26f0 Compare September 7, 2022 13:50

guswynn reviewed Sep 7, 2022

View reviewed changes

petrosagg force-pushed the timely-async-bridge branch from 88f26f0 to 9f17445 Compare September 8, 2022 20:42

guswynn reviewed Sep 9, 2022

View reviewed changes

guswynn approved these changes Sep 9, 2022

View reviewed changes

petrosagg force-pushed the timely-async-bridge branch from 9f17445 to f54fb16 Compare September 12, 2022 14:43

petrosagg enabled auto-merge September 12, 2022 15:11

timely-util: no nonsense timely/async bridge

fa8a719

Signed-off-by: Petros Angelatos <[email protected]>

petrosagg force-pushed the timely-async-bridge branch from f54fb16 to fa8a719 Compare September 12, 2022 15:11

petrosagg merged commit c3a5ba7 into MaterializeInc:main Sep 12, 2022

petrosagg deleted the timely-async-bridge branch September 12, 2022 16:06

		// The timely operator needs to be kept alive if the task is pending or if there
		// are live input handles, perhaps in an another executor

timely-util: no nonsense timely/async bridge #14630

timely-util: no nonsense timely/async bridge #14630

Uh oh!

Conversation

petrosagg commented Sep 5, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Motivation

Uh oh!

Uh oh!

guswynn left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

guswynn left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

petrosagg commented Sep 5, 2022 •

edited

Loading