chore(deps): update crates #6424

renovate · 2024-04-30T16:17:03Z

This PR contains the following updates:

Package	Type	Update	Change
anyhow	workspace.dependencies	patch	`1.0.81` -> `1.0.82`
async-recursion	workspace.dependencies	patch	`1.1.0` -> `1.1.1`
async-trait	workspace.dependencies	patch	`0.1.79` -> `0.1.80`
data-encoding	dependencies	minor	`2.5.0` -> `2.6.0`
either	dependencies	minor	`1.9.0` -> `1.11.0`
futures-concurrency	dependencies	minor	`7.5.0` -> `7.6.0`
proc-macro2	workspace.dependencies	patch	`1.0.79` -> `1.0.81`
quote	workspace.dependencies	patch	`1.0.35` -> `1.0.36`
schemars (source)	workspace.dependencies	patch	`0.8.16` -> `0.8.17`
serde (source)	workspace.dependencies	patch	`1.0.198` -> `1.0.200`
smol_str	dependencies	patch	`0.2.0` -> `0.2.1`
syn	workspace.dependencies	patch	`2.0.58` -> `2.0.60`
termcolor	dependencies	minor	`1.2.0` -> `1.4.1`
trybuild	dev-dependencies	patch	`1.0.91` -> `1.0.93`
unicode-width	dependencies	patch	`0.1.11` -> `0.1.12`
wasmparser (source)	dependencies	minor	`0.202.0` -> `0.206.0`

Release Notes

dtolnay/anyhow (anyhow)

`v1.0.82`

Compare Source

Documentation improvements

dcchut/async-recursion (async-recursion)

`v1.1.1`

Compare Source

What's Changed

Fix incorrect syn features by @dcchut in https://github.com/dcchut/async-recursion/pull/44

Full Changelog: dcchut/async-recursion@v1.1.0...v1.1.1

dtolnay/async-trait (async-trait)

`v0.1.80`

Compare Source

Fix unreachable code warning for async functions that return ! (#265, thanks @de-vri-es)

rayon-rs/either (either)

`v1.11.0`

Compare Source

`v1.10.0`

Compare Source

yoshuawuyts/futures-concurrency (futures-concurrency)

`v7.6.0`

Compare Source

✨ Portable Concurrent Async Iteration ✨

This release introduces the ConcurrentStream API: a concurrent async/.await adaptation of Rayon's ParallelStream API. We've been trying to implement this API for the past six or so years, and we've happy to announce we finally have a working implementation!

The main feature that sets this implementation apart from other, similar attempts is that it works for any existing Stream impl. All you need to do is call the .co() method to obtain a ConcurrentStream, and from that point all Stream combinators should just work as expected:

use futures_concurrency::prelude::*;

let v: Vec<_> = stream::repeat("chashu")
    .co()   // ← call this to convert any `Stream` into a `ConcurrentStream`
    .take(2)
    .map(|msg| async move { format!("hello {msg}") })
    .collect()
    .await;

assert_eq!(v, &["hello chashu", "hello chashu"]);

See the call to collect at the end there? That's right: that's concurrent async iteration, collecting back into a single structure. This makes writing fan-out/fan-in pipelines trivial to author. But that's not all: in addition to converting into collections, we can also directly convert collections into ConcurrentStream implementations too:

use futures_concurrency::prelude::*;

let v: Vec<_> = vec!["chashu", "nori"]
    .into_co_stream()  // ← call this to convert collections into concurrent async iterators
    .map(|msg| async move { format!("hello {msg}") })
    .collect()
    .await;

The amount of concurrency by default is unbounded, but can be bounded by calling the limit method. This will apply backpressure should the stream produce items faster than the concurrent iterator can process them.

This API also resolves the buffered streams problem. ConcurrentStream removes the need for the dreaded combination of mapping to futures and then calling the futures-rs buffered method. Instead it ensures that the processing of items in a loop always happens in concert with the execution of the concurrent futures [^queuing-theory].

[^queuing-theory]: There is a more elaborate version of this problem we don't have a user story for yet. Rust's futures model couples "liveness" to "backpressure". In theory we might fail to send keepalive messages if we apply backpressure for too long; this is a direct artifact of queueing theory, and would be a reason to add some form of async Future::pending / Future::poll_pending method for. This is a more subtle / niche issue than "barbara battles buffered streams". But it's worth explicitly calling out the limits of our solutions.

Notably this API will work with any async runtime or async framework, because it makes no assumptions about the underlying runtime. The only assumption is makes is that an allocator is available. This means that at this time, unlike most other APIs in futures-concurrency this will not work on #[no_std] environments. This, however, is not an inherent restriction but merely an artifact of the implementation. In the future we may explore porting this to a #[no_std] compatible version - this will require some minor API changes, but should, as a system, likely work.

In order to make this system work with parallel execution it should be possible to write a custom adapter. We encourage async runtimes to wrap the ConcurrentStream trait exposed in this crate to create their own ParallelStream system. This can depend on the runtime, and ensure that all execution not only happens concurrently, but can also be scheduled on multiple cores.

We're excited for people to give this a try. We certainly hope this lowers the bar for correctly applying structured, asynchronous, concurrent streaming processing in Rust!