async-rs / async-rs/parallel-stream

A restrospection on interoperatabiliy with futures crate

Open
#15 1 comment 3 reactions 0 assignees View on GitHub
Dominant language
Rust
Stars
96
Forks
13
PR merge metrics
No merged PRs in 30d

Description

## On Restrospection
I've been looking for parallelism combined with async/await and thanks for your decent work.
I'd like to share to retrospection on pulling this crate in my existing project.

My project is filled with stream combinators from futures' [StreamExt](https://docs.rs/futures/0.3.5/futures/stream/trait.StreamExt.html) and async\_std's [StreamExt](https://docs.rs/async-std/1.6.2/async_std/stream/trait.StreamExt.html). They are extension traits of futures' [Stream](https://docs.rs/futures/0.3.5/futures/stream/trait.Stream.html). In this way, any streams with `Stream` trait equip with extended combinators automatically. It's great convenience when you're writing your own stream type.

On the contrary, the parallel-stream's [ParallelStream](https://docs.rs/parallel-stream/2.1.1/parallel_stream/trait.ParallelStream.html) is alien to above extension traits. It's a standalone trait with a family of implemented types. That is, when a stream is turned into a ParallelStream, it loses all combinators from those extension traits. Also, things get more complex when writing your own stream type. I think the `limit()` of `ParallelStream` is the root of evil.

I noticed aggressive trait bounds that is hard to satisfy. For example, the [map](https://docs.rs/parallel-stream/2.1.1/parallel_stream/trait.ParallelStream.html#method.map) method requires the `f` to have `Send`, `Sync` and `Copy`. Only few and very special types have both `Sync` and `Copy`. It makes `map` useless because it restricts the closure cannot have a local variable lacking one of the traits.

```rust
fn map(self, f: F) -> Map where
F: FnMut(Self::Item) -> Fut + Send + Sync + Copy + 'static,
T: Send + 'static,
Fut: Future + Send,
```

## On Alternative Design

I gather above thoughts and attempted an alternative based on your work, and it comes the [par-stream](https://github.com/jerry73204/par-stream). Basically it provides an extension trait `ParStreamExt` to futures's `Stream` and solve the trait bound issue. The `limit` is given on demand. It sets the # of workers only for that stage. It's not 100% equal to your design because the `limit` only applies to one stage rather than a group of stages. Instead, I moved your design to another particular API.

```rust
let shared = Arc::new(AtomicUsize::new(0));
stream.par_then(None, |item| { // None sets the limit to the number of cores.
let shared = shared.clone(); // Clone a variable without `Copy` trait
async move {
let new_item = compute(item, shared);
new_item
}
})
.collect::>(); // from futures' StreamExt
```

To the limit the workers of a group of combinators, I suggest the builder patten. So far, it's implemented yet in my crate, but we can see how it would become here.

```rust
stream
.enumerate()
/* start of group */
.into_par_group(ParGroupConfig { // turn to a parallel group builder
limit: Some(4),
..Default::default() // using default runtime and other default options, etc
})
.then(|item| { async move { /* omit */ } }) // first stage
.filter_map(|item| { async move { /* omit */ } }) // second stage
.build_stream() // build a stream from the group builder
/* end of group */
. collect::>(); // combinator from futures' StreamExt
```

In this way, it lets users to customize the whole parallel group in one config. It would work seamlessly with existing futures' combinators.

Here we may move to more thorough discussion to help the design to evolve. It's fine for me to look more carefully to my work and find a way to combine them.

Contributor guide

Open the contributing guide

Research direction

Start by reading the existing ParallelStream API, especially limit() and map(), then compare it with the proposed ParStreamExt and builder-style into_par_group design. The issue does not identify a repository file or test; it needs an agreed API direction before implementation can be considered done.

Written by the indexing model from the issue text.

Assessment

Tech stack
rust
Domain
backend-api-design
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Needs clarification
Newbie friendliness
20/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.