主页 / coding / 问题 / 79156647
Accepted
narumi
Asked: 2024-11-05 02:51:24 +0800 CST

为什么 Future 实现中的 `while let` 会触发 Waker，而 `if let` 却不会？

  • 772

我一直在 Rust 中研究 Future 实现，但遇到了一些我无法完全理解的行为。具体来说，我在方法std::sync::mpsc::Receiver内部使用poll，并尝试在while let和之间做出选择if let以接收消息。

这是我的轮询方法的简化版本：

fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
    let this = self.get_mut();

    // Option 1: Using `while let`
    while let Ok(new_attributes) = this.receiver.try_recv() {
        // Process the message
    }

    // Option 2: Using `if let`
    if let Ok(new_attributes) = this.receiver.try_recv() {
        // Process the message
    }

    Poll::Pending
}

我观察到的情况

当我使用 while let 时，一切似乎都按预期工作：当新消息到达时，Future 会重新轮询。但是，当我使用 if let 时，Future 似乎挂起了，并且永远不会再次唤醒，即使有新消息可用。

我的理解

我知道 Waker 应该用于通知执行者应该重新轮询 Future。鉴于我的轮询函数始终返回 Poll::Pending，我预计 Future 将继续被重复轮询，除非进程本身停止。但是，我不明白为什么 while let 确保 Waker 被正确触发，而 if let 似乎没有这样做。

其他背景信息

未来是这样产生的：

ctx.task_executor()
    .spawn_critical_blocking("transaction execution service", Box::pin(fut_struct));

spawn_critical_blocking 函数的工作原理如下：

pub fn spawn_critical_blocking<F>(&self, name: &'static str, fut: F) -> JoinHandle<()>
where
    F: Future<Output = ()> + Send + 'static,
{
    self.spawn_critical_as(name, fut, TaskKind::Blocking)
}

在内部：

fn spawn_critical_as<F>(
    &self,
    name: &'static str,
    fut: F,
    task_kind: TaskKind,
) -> JoinHandle<()>
where
    F: Future<Output = ()> + Send + 'static,
{
    // Wrapping the future and handling task errors
    let task = std::panic::AssertUnwindSafe(fut)
        .catch_unwind()
        .map_err(...);

    let task = async move {
        // Handling task shutdown and execution
        let task = pin!(task);
        let _ = select(on_shutdown, task).await;
    };

    self.spawn_on_rt(task, task_kind)
}

然后该spawn_on_rt方法使用spawn_blocking。

我的问题

  • 为什么使用 while let 可以确保 Waker 被触发，但使用 if let
  • 做

不是？

  • 导致这种行为差异的根本机制是什么？

如有任何见解或解释我将不胜感激！

asynchronous

1 个回答

  1. Best Answer

    在回答你的问题之前，我想先质疑一下你的一个假设：

    鉴于我的轮询函数始终返回Poll::Pending，我预计会Future继续被反复轮询

    虽然从技术上讲这是可行的，但根本无法预料。正常async运行时间（如tokio）尝试仅在取得进展时进行轮询，以提高效率，如使用Waker传达的信号Context。

    请注意，您永远不会使用这个Context参数，除非您的未来总是Poll::Ready立即返回，否则这是一个错误，因为没有可靠的理由让您的poll函数再次被调用。（从技术上讲，它有可能被再次调用，因为async运行时可能会虚假地poll返回未来，但您永远不应该依赖它！）

    为什么使用 while let 可以确保触发Waker，而使用 if let 却不能？

    没有任何内容while let可以保证触发Waker。 因此，以下代码不起作用：

    use std::task::{Context, Poll};
use std::pin::Pin;
use std::future::Future;
use std::sync::mpsc::{channel, Receiver};

struct Demo{
    receiver: Receiver<()>,
}

impl Future for Demo {
    type Output = ();
    fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.get_mut();
        println!("poll");
    
        // Option 1: Using `while let`
        while let Ok(_) = this.receiver.try_recv() {
            println!("processing");
            // Process the message
        }
    
        Poll::Pending
    }
}

#[tokio::main]
pub async fn main() {
    let (sender, receiver) = channel();
    let demo = Demo{receiver};
    let thread = std::thread::spawn(move || {
        sender.send(()).unwrap();
    });
    println!("before");
    demo.await;
    println!("after");
    thread.join().unwrap();
}

    输出：

    before
poll

    导致这种行为差异的根本机制是什么？

    if let和之间的唯一区别while let在于，如果有更多缓冲消息，后者将继续从通道读取。对于您所看到的情况，另一种更合理的解释是，while let允许单次调用poll来处理多条消息，同时不执行任何操作来安排 的后续调用poll。

    如何修复

    您的代码的主要问题是您尝试在上下文中使用标准库中的同步通道实现async；它们不兼容，以这种方式混合它们是不正确且不可靠的。您应该切换到async基于的通道，例如tokio::sync::mpsc：

    use std::task::{Context, Poll};
use std::pin::Pin;
use std::future::Future;
use tokio::sync::mpsc::{channel, Receiver};

struct Demo{
    receiver: Receiver<()>,
}

impl Future for Demo {
    type Output = ();
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.get_mut();
        
        println!("poll");
    
        if let Poll::Ready(result) = this.receiver.poll_recv(cx) {
            if let Some(_message) = result {
                println!("processed");
            } else {
                println!("channel closed");
            }
            return Poll::Ready(());
        }
    
        Poll::Pending
    }
}

#[tokio::main]
pub async fn main() {
    let (sender, receiver) = channel(10);
    let demo = Demo{receiver};
    let task = tokio::spawn(async move {
        sender.send(()).await.unwrap();
    });
    println!("before");
    demo.await;
    println!("after");
    task.await.unwrap();
}

    请注意，现在在通道上接收消息涉及一个类似 innerpoll的调用，该调用将Context作为参数。如果当前没有消息要接收，则将Waker与通道相关联（以特定于实现的方式），这样当消息到达时，它可能会导致未来被轮询。

    输出：

    before
poll
poll
processsed
after

    注意有两个轮询。据推测，第一个轮询没有收到任何消息。正确使用async轮询意味着可以再次对未来进行有效的轮询。

    • 5

相关问题