讨论区：在 Rust 中用 GAT 手动实现零开销 async trait

[skyzh]

No description provided.

[wangrunji0408]

对于文中举的 async iterator 的例子，这里再给出一种基于 Stream 和 generator 的更简单写法：

# Cargo.toml
[dependencies]
futures = "0.3"
futures-async-stream = "0.2"
tokio = { version = "1", features = ["rt", "macros"] }

// lib.rs
#![feature(generators)]

use futures::{pin_mut, stream::Stream, StreamExt};
use futures_async_stream::stream;
use std::ops::Range;

#[stream(item = (String, String))]
async fn gen(idx: Range<usize>) {
    for i in idx {
        let key = format!("key_{:05}", i);
        let value = format!("value_{:05}", i);
        yield (key, value);
    }
}

#[stream(item = (String, String))]
async fn concat(iters: Vec<impl Stream<Item = (String, String)>>) {
    for iter in iters {
        #[for_await]
        for item in iter {
            yield item;
        }
    }
}

#[tokio::test]
async fn test_gen() {
    let iter = gen(0..10);
    pin_mut!(iter);
    while let Some((key, value)) = iter.next().await {
        println!("{:?} {:?}", key, value);
    }
}

#[tokio::test]
async fn test_concat() {
    let iter1 = gen(0..5);
    let iter2 = gen(5..10);
    let concat_iter = concat(vec![iter1, iter2]);
    pin_mut!(concat_iter);

    while let Some((key, value)) = concat_iter.next().await {
        println!("{:?} {:?}", key, value);
    }
}

这种写法不需要自己定义 async trait，而是使用标准的 Stream trait，从而避免使用 GAT，并且可以使用 generator 语法来生成，不需要自己维护迭代器状态，所以代码非常精简。

这种写法和文中的方法一样是零开销的，但是有一个很大的限制是：

目前 generator 不允许 yield 引用内部状态的变量，只能输出 String 而不能是 &str。

如果我们尝试这么写：

#[stream(item = (&'a str, &'a str))]
async fn gen<'a>(idx: Range<usize>) {
    for i in idx {
        let key = format!("key_{:05}", i);
        let value = format!("value_{:05}", i);
        yield (key.as_str(), value.as_str());
    }
}

编译器会报错：

error[E0515]: cannot yield value referencing local variable `value`
  --> src/lib.rs:7:1
   |
7  | #[stream(item = (&'a str, &'a str))]
   | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ yields a value referencing data owned by the current function
...
12 |         yield (key.as_str(), value.as_str());
   |                              -------------- `value` is borrowed here

最后就是这种写法只适用于迭代器这一种模式，对于其它场景还是要用文中的更为通用的 GAT 写法。

[skyzh]

本来写这篇文章的时候想顺便写个 AsyncIterator (Stream) 的实现，后来发现没法返回内部的变量（not zero-copy）🤪🤪

我猜想如果要做成 async iterator 要自己实现 Stream 了（

[jiangzhe]

ConcatIterator实现中borrow check报错似乎用pattern match解构一下可以解决，见下面的代码。

如果是在stable Rust中不使用Box::pin实现async，是不是只能手动实现Future了？
但是状态维护以及和其他async代码整合确实是大问题...

struct ConcatIterator<F>{
    iters: Vec<F>,
    idx: usize,
}

impl<F> ConcatIterator<F> {
    fn new(iters: Vec<F>) -> Self {
        ConcatIterator{iters, idx: 0}
    }
}

impl<T, F> Future for ConcatIterator<F>
where
    F: Future<Output=Option<T>> + Unpin,
{
    type Output = Option<T>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<T>> {
        while self.idx < self.iters.len() {
            let ConcatIterator{iters, idx} = &mut *self;
            let iter = &mut iters[*idx];
            match Pin::new(iter).poll(cx) {
                Poll::Pending => return Poll::Pending,
                Poll::Ready(v) => {
                    if v.is_none() {
                        // exhausted current iterator
                        self.idx += 1;
                    } else {
                        return Poll::Ready(v)
                    }
                }
            }
        }
        Poll::Ready(None)
    }
}

[xiaguan]

Update 一下，async_fn_trait

#![feature(async_fn_in_trait)]
pub trait MyIteratorVersion3 {
    async fn next3(&mut self) -> Option<(&[u8], &[u8])>;
}

可以不用手写了 😭