异步：如何在选择（no_std 环境）的循环中继续使用相同的未来？

Question

我有两个异步函数：get_message 和 get_event。我想在消息到达或事件发生时执行一个操作，并在无限循环中永远执行该操作。

简化的设置如下所示：

use futures::{future::select, future::Either, pin_mut};

impl MsgReceiver {
    async fn get_message(&mut self) -> Message { /* ... */ }
}

impl EventListener {
    async fn get_event(&mut self) -> Event { /* ... */ }
}

async fn eternal_task(receiver: MsgReceiver, listener: EventListener) -> ! {
    let get_msg_fut = receiver.get_message();
    pin_mut!(get_msg_fut);
    loop {
        let get_event_fut = listener.get_event();

        pin_mut!(get_event_fut);

        match select(get_event_fut, get_msg_fut).await {
            Either::Left((ev, r_get_msg_fut)) => {
                /* react to the event */

                // r_get_msg_fut is not done, how to reuse it in the next iteration?
            }
            Either::Right((msg, r_get_event_fut)) => {
                /* react to the message */

                // it's fine to drop get_event_fut here

                // the following line causes a double-mut-borrow error on receiver,
                // despite receiver isn't borrowed anymore (the old future is completed and dropped)
                let new_future = receiver.get_message();
            }
        };
    }
}

我在这里有三个主要问题：

1. 当一个事件首先发生时，如何告诉 rust 我想在下一个循环迭代中重用未完成的 get_message future？
2. 当一个消息先来时，如何构造一个没有借用错误的新未来？
3. 当（2）解决后，如何将新的未来放入同一个固定内存位置并在下一次循环迭代中使用它？

Answer 1

我成功地使用了它，但无法摆脱 Box::pin

use futures::{future::select, future::Either, pin_mut};
use std::sync::Mutex;
#[derive(Debug)]
struct MsgReceiver;
#[derive(Debug)]
struct EventListener;
#[derive(Debug)]
struct Message;
#[derive(Debug)]
struct Event;
impl MsgReceiver {
    async fn get_message(&mut self) -> Message {
        Message
    }
}

impl EventListener {
    async fn get_event(&mut self) -> Event { 
        Event
    }

}

async fn eternal_task(receiver: MsgReceiver, mut listener: EventListener) -> ! {
    let receiver = Mutex::new(receiver);
    let mut f = None;
    loop {
        let get_msg_fut = match f.take() {
            None => {
                let mut l = receiver.lock();
                Box::pin(async move {
                    l.get_message().await
                })
            }
            Some(f) => f,

        };
        let get_event_fut = listener.get_event();
        pin_mut!(get_event_fut);

        match select(get_event_fut, get_msg_fut).await {
            Either::Left((ev, r_get_msg_fut)) => {
                /* react to the event */
                // store the future for next iteration
                f = Some(r_get_msg_fut);
            }
            Either::Right((msg, r_get_event_fut)) => {
                /* react to the message */
            }
        };
    }
}

#[tokio::main]
async fn main() {
    eternal_task(MsgReceiver, EventListener).await;
}

Answer 2

我认为这很难做到正确，即使使用 unsafe 也可能需要完成此操作。坚持和重用相同的变量并不太难，它实际上是最难的#2（至少对于当前的借用检查器）。

我找到了一个解决方案，通过使用 async-stream 板条箱提供中介来完全规避问题：

async fn eternal_task(mut receiver: MsgReceiver, mut listener: EventListener) -> ! {
    let combined = futures::stream::select(
        stream! { loop { yield Either::Left(receiver.get_message().await); } },
        stream! { loop { yield Either::Right(listener.get_event().await); } },
    );

    pin_mut!(combined);
    while let Some(msg_or_evt) = combined.next().await {
        match msg_or_evt {
            Either::Left(msg) => {
                // do something with msg
            }
            Either::Right(evt) => {
                // do something with evt
            }
        };
    }

    unreachable!()
}

它使用 stream! 宏来生成一个类型，该类型不断调用并从 .get_message() 和 .get_event() 生成值。然后它使用 futures::stream::select 和 Either 来组合它们。然后它只是循环结果的问题。它适用于#![no_std]。