Skip to content

Conversation

petrosagg
Copy link
Contributor

@petrosagg petrosagg commented Feb 7, 2021

Example usage:

use timely::dataflow::operators::{Event, ToStreamAsync};
use timely::dataflow::operators::Inspect;

#[tokio::main]
async fn main() {
    let iter = (0..4)
        .map(|x| Event::Message(0, x))
        .chain(Some(Event::Progress(1)));
    let stream = futures::stream::iter(iter);

    timely::example(move |scope| {
        let (future, stream) = stream.to_stream(scope);
        tokio::spawn(future);
        stream.inspect(|x| println!("seen: {:?}", x));
    });
}

@frankmcsherry
Copy link
Member

Thanks! Can you say a bit about futures-util and how important it is? It looks like it pulls in a fair bit non-optionally.

Comment on lines 77 to 81
/// Data and progress events of the native stream.
pub enum Event<T, D> {
/// Indicates that timestamps have advanced to time T
Progress(T),
/// Indicates that event D happened at time T
Message(T, D),
}
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We should talk through this! There is already a similar enum in the capture module, and it is a bit more rich than this. In particular, this enum isn't able to capture partially ordered times, where there may be a set of capabilities rather than just one.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I did see that one but wasn't sure what the Vec is for. Let's talk about it tomorrow

@petrosagg
Copy link
Contributor Author

Thanks! Can you say a bit about futures-util and how important it is? It looks like it pulls in a fair bit non-optionally.

It's still smaller than the full futures crate but I just pushed a change that reduces it to futures-core. The only reason I needed futures-util is for the .fuse() method to match the implementation on iterators but turns out I don't need that

@petrosagg petrosagg force-pushed the async-stream branch 7 times, most recently from b1a39d0 to e578886 Compare February 8, 2021 17:43
@frankmcsherry
Copy link
Member

Thanks! I'll get cracking on this promptly. The main ? for me is getting my head around Future so I can own some of that technical debt. I'm familiar with the C# version of it, but I'll need to get up to speed on Rust's version.

Comment on lines +376 to +377
B: std::borrow::Borrow<T>,
F: IntoIterator<Item=B>,
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Minor, but can this be

F: IntoIterator,
F::Item: std::borrow::Borrow<T>,

or is the generic argument B necessary / helpful?

@petrosagg
Copy link
Contributor Author

ok, I did some minor cleanups and fixed the example code. It should be good to merge I think!

Copy link
Contributor

@benesch benesch left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The futures goop looks great to me. No opinion on the Event protocol!

cap_set.downgrade(time);
}
Some(Event::Message(time, data)) => {
output.session(&cap_set.delayed(&time)).give(data);
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Minor, but this will eventually have throughput issues. At least, timely can send data around much faster than it can mint capabilities. It may eventually be smart to have the D of the enum implement IntoIterator<Item = D2>, though perhaps .. that is just a flatmap after this. Idk. Doesn't block anything here, but wanted to call out!

Copy link
Member

@frankmcsherry frankmcsherry left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looks good to me. Various comments that are good for future discussion (lol not intended nvm sry).

antiguru added a commit to antiguru/timely-dataflow that referenced this pull request Feb 5, 2024
@github-actions github-actions bot mentioned this pull request Oct 29, 2024
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

Successfully merging this pull request may close these issues.

3 participants