Add minimal genarator-like stream implementation

[pepijnve]

Which issue does this PR close?

None, related to PR #23407

Rationale for this change

Manually writing Stream implementations can be quite tedious since it requires implementing the state machine of the stream yourself. This often results in hard to read code. The same problem exists with manual Future implementations and async was added to the Rust language to mitigate exactly this problem. Unfortunately there's no language level support yet to help with writing streams/generators.

There are quite a few projects that attempt to fill this gap:

• Tokio's async_stream provides a proc macro that recognises a yield keyword. This is a good implementation, but proc macros don't play nice with code formatting, increase compile time, and in this particular case prevent decomposition into smaller functions.
• async_fn_stream provides an implementation of the same concept, but without the proc macro. Unfortunately this implementation chooses a heavier mechanism to communicate generated values back to the stream via a SmallVec.
• genawaiter is a more general purpose generator library, but this can also be used to implement Streams. This project does miss some of the ergonomics provided by the other two in the form of try_ variants that help in implementing fallible streams.

Since none of these variants seems like the ideal candidate, the best option might be to have a custom implementation of the concept tailored to the needs of the DataFusion project. This PR provides an initial draft of exactly that.

What changes are included in this PR?

• Adds async_stream and async_try_stream functions that create Stream implementations based on an async generator function.

The initial implementation was inspired by the macro expansion produced by async_stream. The code was then adapted further taking inspiration from the two other libraries. Specifically, the Emitter terminology was taken from async_fn_stream and the Arc<Mutex<Option<T>>> value transfer mechanism was taken from genawaiter.

async_stream uses a very light weight thread-local storage based mechanism to handle value transfer, but this felt too risky to use when the emitter is exposed. It makes sense in the Tokio implementation since the proc macro can manage control flow in a stricter way.

I wasn't entirely sure if taking some inspiration from has code licensing implications. I applied ASLv2 for now on the added code.

Are these changes tested?

Test code was mainly adapted from the tokio implementation.

Are there any user-facing changes?

Yes, new public function async_stream and async_try_stream

[rluvaton]

Can you please add comment on what is the mutex here is for? same in the try emitter and receiver

[pepijnve]

I'm not sure at what level of detail to explain this or if it's actually strictly required. The emitter needs to be Send for now to be able to do the async move, but I don't think we really need Sync do we?

[rluvaton]

the mutex is required since you modify the value and Arc does not let you modify the value (unless as_mut), no?

[pepijnve]

We could probably get away with an UnsafeCell here as well. I'm experimenting a bit to see what's possible.

I hit compile errors and gave up on this. Sticking with the mutex for now.

[rluvaton]

Thanks for the PR, could you try locally if modifying the code in #23407 to use this new generator compile?

[pepijnve]

Thanks for the PR, could you try locally if modifying the code in #23407 to use this new generator compile?

See https://github.com/pepijnve/datafusion/tree/use_yield_alternative

[rluvaton]

LGTM, thanks!