Ref #22

Remove the need for MaybeDone on Join for tuples.

This is a spiritual successor of #74, thank you @yoshuawuyts and @poliorcetics for your reviews and ideas!

The perf results are really nice now:

>> critcmp main maybegone -f tuple::join
group             main                                   maybegone
-----             ----                                   ---------
tuple::join 10    1.00    246.2±0.79ns        ? ?/sec    2.86    703.8±1.49ns        ? ?/sec

edit: removes reference to #21, as we're no longing implementing "perfect" waking

This mimics the implementation for array::merge.

I used that hack I introduced on #74 in a more generalized form, we can rewrite the gen_conditions in terms of this new macro (basically the same way I do here)

Ref #21

Tries to implement #21 for array and vec Join.

I’m a bit surprised as I have a ~400% regression in the benchmarks. Though I had a hard time figuring out how the benchmarks are working, and what they are even benchmarking.

Looks like the CountdownFuture used in benchmarks might not be ideal to test for this? To properly test this, we could have a list of futures that wake from back to front, and count the number of times they were polled. With "perfect" waking, they should only be polled twice (initially, and a second time once they are woken). By making the poll "expensive", the wins should be visible in the benchmarks as well.

Duplicate changes from https://github.com/yoshuawuyts/futures-concurrency/pull/94/files for RaceOk impl.

Should resolve the second item on https://github.com/yoshuawuyts/futures-concurrency/issues/98

Hey all,

Just wanted to share that I'll be out of office starting tomorrow until December 19th. In the interim: please feel free to keep filing issues, make PRs, and merge them too. I trust folks here to exercise their judgement on which things to merge, and make sure changes are reviewed, etc. So far things have been going pretty steady, so please feel free to things going the way they are.

Both @matheus-consoli and @eholk have commit rights and can merge PRs. And @eholk also has the ability to publish new releases in a pinch. I don't expect a new release to have to be made, but should there be some critical issue that needs a hot fix: it's at least possible if I'm not around.

For anyone new to the project: welcome! I ask you to please be respectful of people's time. Other than me are folks here are really just volunteering their time to help out on this. While they may be kind enough to help in a pinch, they are under absolutely no obligation to do so. Should there be a larger issues or design questions: I'll be back in a few weeks and I'll be happy to discuss them then.

Anyway, I figured I'd let people know so nobody wonders why I'm suddenly radio-silent. See y'all in a few weeks!

By using named fields instead of anonymous one in the $mod_name::Streams structs, we can completely remove the need for the very tuple_for_each! macros, which makes the code more maintainable and the crate faster to compile (though that wasn't a limitation at this point).

I also took the opportunity to remove the tuple_len! macro because recursivity is heavy on the compiler. The generated code for this part doesn't slow down at all because it's all const-computed anyway (but instead of producing an imbalanced binary tree, we produce a more "normal" one which is much easier to deal with for rustc).

Finally, I greatly simplified the gen_conditions! macro by removing the recursion by using the Indexes enum again (that trick is awesome, I'm glad I finally found a crate to share it with 😄), which (again) removes a very unbalanced tree of 0 + 1 + 1 + 1 + ... in the generated code.

Ref #22

Remove the need for MaybeDone on tuple::join by creating for each future three fields: the Future itself; its result; its state.

I had to add a dependency on paste to create the $F_fields.

Bench:

>> critcmp main patch -f tuple::join
group             main                                   patch
-----             ----                                   -----
tuple::join 10    1.00    232.6±9.13ns        ? ?/sec    1.14    266.0±3.09ns        ? ?/sec

Continues the work started in https://github.com/yoshuawuyts/futures-concurrency/pull/50, removing the final set of allocations. The strategy used for this is to create an initial WakerList, which we populate with intermediate wakers on the first call to poll. In a future patch we may want to create a static variant of this as well for use in our fixed-length combinators.

I also took the opportunity to rename StreamWaker to InlineWaker, since we may want to use this mechanism for the other combinators too.

Benchmarks

Performance seems to be about 1.5-2x as good as before, which is expected since we no longer allocate on each call to poll:

❯ critcmp main patch -f vec::merge
group              main                                   patch
-----              ----                                   -----
vec::merge 10      1.89      5.3±0.15µs        ? ?/sec    1.00      2.8±0.08µs        ? ?/sec        
vec::merge 100     1.68     75.3±0.56µs        ? ?/sec    1.00     44.8±1.11µs        ? ?/sec        
vec::merge 1000    1.46      2.8±0.02ms        ? ?/sec    1.00  1946.8±40.07µs        ? ?/sec   

In this function you use the unsafe keyword for almost the entrie function body.

We need to mark unsafe operations more precisely using unsafe keyword. Keeping unsafe blocks small can bring many benefits. For example, when mistakes happen, we can locate any errors related to memory safety within an unsafe block. This is the balance between Safe and Unsafe Rust. The separation is designed to make using Safe Rust as ergonomic as possible, but requires extra effort and care when writing Unsafe Rust.

Hope this PR can help you. Best regards. References https://doc.rust-lang.org/nomicon/safe-unsafe-meaning.html https://doc.rust-lang.org/book/ch19-01-unsafe-rust.html

In https://blog.yoshuawuyts.com/futures-concurrency-3/ there is mention of Future::race extension method in async_std.

There is also a suggestion that there could be an extension trait that's implemented for tuples, similar to how Merge works currently in this crate: https://docs.rs/futures-concurrency/latest/futures_concurrency/trait.Merge.html#impl-Merge-for-(S0%2C%20S1%2C%20S2%2C%20S3)

The post also says:

In this post we're going to take a look at how this mode of concurrency works, take a closer look at the issues select! {} has, discuss Stream::merge as an alternative, and finally we'll look at what the ergonomics might look like in the future. This post is part of the "Futures Concurrency" series. You can find all library code mentioned in this post as part of the futures-concurrency crate.

I find the idea of a race extension method really compelling, and have told all of my workmates about the blog post, but there doesn't seem to be an implementation of it yet. If someone wanted to implement this trait, would this repo be a reasonable place to put it?

Initial impl of RaceOk for tuples

something to look at:

I'm using an array as the internal representation of AggregateError, which can be surprising as we tend to expect a tuple-ish.

I opted to use the array because 1) arrays are much easier to work on compared to macro-based-variadic-tuples; 2) as of this implementation, we're restricting every Future to have the same Output, ergo every error has the same type.

Performance improvements

WakerArray/WakerVec now keeps track of what futures were woken in a list, so that no O(n) search is needed when polling. This makes polling O(woken) instead of O(n). The impact is extremely significant for large combinators.

Also, the new implementation avoids locking and unlocking the Readiness Mutex in a loop. It copies out the data necessary for iteration once at the beginning of each poll.

I've also made WakerArray/WakerVec use a single Arc shared between all wakers (without giving up perfect waking) instead of needing a separate Arc for each. This change involves some unsafe work with RawWaker.

API changes

Race, TryJoin, and RaceOk for tuples now supports futures with heterogeneous results. Racing futures with different output types would return an enum whose variants are the possible outputs. If all outputs are the same type then there is a function to convert the enum to that type.

RaceOk error types are simplified to be just array/tuple/vec of the errors. I removed the wrapping AggregateError because it can be confusing for tuples (since error types are not necessarily homogeneous anymore).

Organizational changes

As part of rewriting the various combinators, I've merged the code for join/try_join/race/race_ok. There is now a crate-private futures::common module with a generic combinator whose behaviors can be controlled to match join/try_join/race/race_ok by a generic type parameter. For tuples, I basically implement try join and make every other combinator delegate to that.

I've also upped the edition to Rust 2021. This is not really necessary but the disjoint closure capture saves a few lines of code.

I renamed "Readiness" to "Awakeness" because the former gets confusing since Poll/PollState::Ready means the future is complete rather than awake.

Benchmark

Currently, CountdownFuture/Stream wake and complete in perfect order. This is unrealistic. I added shuffling (with a fixed seed) so that they better represent real workload.

Below: after = this PR before = origin/main with countdown shuffling commit cherry-picked

$ critcmp  after before
group                after                                  before
-----                -----                                  ------
array::join 10       1.00  1854.5±12.26ns        ? ?/sec    1.20      2.2±0.02µs        ? ?/sec
array::join 100      1.00     20.6±0.16µs        ? ?/sec    1.17     24.1±0.28µs        ? ?/sec
array::join 1000     1.00    219.4±3.13µs        ? ?/sec    5.67  1242.9±19.53µs        ? ?/sec
array::merge 10      1.00  1828.5±29.31ns        ? ?/sec    1.32      2.4±0.05µs        ? ?/sec
array::merge 100     1.00     19.9±0.26µs        ? ?/sec    2.15     42.8±3.00µs        ? ?/sec
array::merge 1000    1.00    225.9±3.09µs        ? ?/sec    8.98      2.0±0.04ms        ? ?/sec
array::race 10       1.07  1138.1±18.48ns        ? ?/sec    1.00  1061.1±43.17ns        ? ?/sec
array::race 100      1.00      7.9±0.09µs        ? ?/sec    1.41     11.1±0.17µs        ? ?/sec
array::race 1000     1.00     82.5±1.13µs        ? ?/sec    1.58    130.6±2.91µs        ? ?/sec
tuple::join 10       1.00  1912.5±29.00ns        ? ?/sec    1.14      2.2±0.03µs        ? ?/sec
tuple::merge 10      1.00      2.2±0.03µs        ? ?/sec    1.23      2.7±0.06µs        ? ?/sec
tuple::race 10       1.15  1134.5±20.99ns        ? ?/sec    1.00   987.5±13.09ns        ? ?/sec
vec::join 10         1.00      2.3±0.06µs        ? ?/sec    1.08      2.5±0.05µs        ? ?/sec
vec::join 100        1.00     18.0±0.19µs        ? ?/sec    2.05     36.9±0.47µs        ? ?/sec
vec::join 1000       1.00    202.1±1.68µs        ? ?/sec    9.45  1909.9±161.66µs        ? ?/sec
vec::merge 10        1.00      2.4±0.04µs        ? ?/sec    1.18      2.8±0.05µs        ? ?/sec
vec::merge 100       1.00     20.8±0.25µs        ? ?/sec    2.57     53.3±1.27µs        ? ?/sec
vec::merge 1000      1.00    222.8±3.95µs        ? ?/sec    12.86     2.9±0.05ms        ? ?/sec
vec::race 10         1.33  1349.1±15.53ns        ? ?/sec    1.00  1011.6±13.01ns        ? ?/sec
vec::race 100        1.00      7.3±0.10µs        ? ?/sec    1.45     10.6±0.14µs        ? ?/sec
vec::race 1000       1.00     72.0±1.64µs        ? ?/sec    1.70    122.7±1.75µs        ? ?/sec

https://github.com/yoshuawuyts/futures-concurrency/pull/109 implemented RaceOk for tuples, but is missing a PinnedDrop implementation. This is needed since we have a MaybeUninit structure which holds the errors. If we get an error once, store it, and then cancel the RaceOk call - we've now leaked data - since MaybeUninit types need to be manually de-initialized on drop.

Now that #104 exists to close https://github.com/yoshuawuyts/futures-concurrency/issues/85, there is a real question about fairness and chaining. I've made the case before that in order to guarantee fairness, the scheduling algorithm needs to know about all types it operates on. When we were still using permutations and even just rng-based starting points, I believe this to be true. But I'm slowly coming around to @eholk's idea that this may not be the case.

Benefits

If we resolve this, I think we may be able to improve our ergonomics. Take for example the following code, which I believe to be quite representative of futures-concurrency's ergonomics:

let streams = (
    socket_stream.map(Either::Response),
    route_rx.stream().map(Either::Request),
    pinger.map(|_| Either::Ping),
);
let mut merged = streams.merge();
while let Some(either) = merged.next().await { ... }

The tuple instantiation imo looks quite foreign. In this repo's style, we'd probably instead choose to name the futures, flattening the operation somewhat:

let a = socket_stream.map(Either::Response);
let b = route_rx.stream().map(Either::Request);
let c = pinger.map(|_| Either::Ping);

let mut merged = (a, b, c).merge();
while let Some(either) = merged.next().await { ... }

But while I'd argue this is more pleasant to read, we can't expect people to always do this. The earlier example is often to easier to write, and thus will be written as such. But a chaining API could probably be even easier to author as well:

let mut merged = socket_stream
    .map(Either::Response)
    .merge(route_rx.stream().map(Either::Request))
    .merge(pinger.map(|_| Either::Ping));

while let Some(either) = merged.next().await { ... }

We used to have this API, but we ended up removing it. And I think there's definitely a case to be made to add this back. Just like we'd be expected to have both: async_iter::AsyncIterator::chain and impl async_iter::Chain for tuple, so could we have this for both variants of merge.

Implementation

I'd love to hear more from @eholk here. But my initial hunch is that perhaps something like ExactSizeIterator could help us. But rather than return how many items are contained in an iterator, it'd return the number of iterators contained within. That way outer iterators can track how often they should call inner iterators before moving on. I think this may need specialization to work though?

I think even if we can't make the API strictly fair, it might still be worth adding the chaining API - and we can possibly resolve the fairness issues in the stdlib? Or maybe we can add a nightly flag with the specializations on it as part of this lib? Do folks have thoughts on this?

We should duplicate the pretty printing logic of https://github.com/yoshuawuyts/futures-concurrency/pull/94 for the remainder of the race_ok types.

Tasks

• [x] impl RaceOk for Vec
• [X] impl RaceOk for array
• [ ] impl RaceOk for tuple