Início / coding / Perguntas / 79156647
Accepted
narumi
Asked: 2024-11-05 02:51:24 +0800 CST

Por que `while let` em uma implementação Future aciona o Waker, mas `if let` não?

  • 772

Tenho trabalhado em uma implementação Future em Rust, e encontrei um comportamento que não entendo completamente. Especificamente, estou usando std::sync::mpsc::Receiverdentro do pollmétodo, e estou tentando decidir entre while lete if letpara receber mensagens.

Aqui está a versão simplificada do meu método de enquete:

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
}

O que eu observei

Quando uso while let, tudo parece funcionar como esperado: o Future é re-polled quando novas mensagens chegam. No entanto, quando uso if let, o Future parece travar e nunca mais acorda, mesmo quando novas mensagens estão disponíveis.

Meu Entendimento

Eu sei que o Waker deve ser usado para notificar o executor de que o Future deve ser re-pesquisado. Dado que minha função de pesquisa sempre retorna Poll::Pending, eu esperaria que o Future continuasse sendo pesquisado repetidamente, a menos que o próprio processo pare. No entanto, não entendo por que while let garante que o Waker seja acionado corretamente, enquanto if let não parece fazer o mesmo.

Contexto adicional

O Futuro é gerado assim:

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

A função spawn_critical_blocking funciona da seguinte maneira:

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)
}

E internamente:

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)
}

O spawn_on_rtmétodo então usa spawn_blocking.

Minha pergunta

  • Por que usar while let garante que o Waker seja acionado, mas usar if let
  • faz

não?

  • Qual é o mecanismo subjacente que causa essa diferença de comportamento?

Quaisquer informações ou explicações serão muito apreciadas!

asynchronous

1 respostas

  1. Best Answer

    Antes de responder às suas perguntas, gostaria de desafiar uma de suas suposições:

    Dado que minha função de pesquisa sempre retorna Poll::Pending, eu esperaria que ela Futurecontinuasse sendo pesquisada repetidamente

    Isso, embora tecnicamente possível, não é de se esperar. asyncTempos de execução normais (como tokio) tentam ser eficientes apenas sondando quando o progresso foi feito, conforme sinalizado usando o Wakertransmitido pelo Context.

    Observe que você nunca usa o Contextargumento which, a menos que seu future sempre retorne Poll::Readyimediatamente, é um erro, pois não há razão confiável para que sua pollfunção seja chamada novamente. (Tecnicamente, é possível que ela seja chamada novamente, porque asyncos tempos de execução podem criar futures espúrios poll, mas você nunca deve confiar nisso!)

    Por que usar while let garante que o Wakerseja acionado, mas usar if let não?

    Não há nada sobre while letisso que seja garantido para disparar oWaker . É por esse motivo que o código a seguir não funciona:

    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();
}

    Saída:

    before
poll

    Qual é o mecanismo subjacente que causa essa diferença de comportamento?

    A única diferença entre if lete while leté que o último continuará a ler do canal se houver mais mensagens armazenadas em buffer. Uma explicação alternativa, mais plausível, para o que você está vendo é que while letpermite uma única invocação de pollpara processar várias mensagens sem fazer nada para agendar uma invocação subsequente depoll .

    Como consertar

    O principal problema com seu código é que você está tentando usar uma implementação de canal síncrono da biblioteca padrão em um asynccontexto; Eles são incompatíveis, e misturá-los dessa forma é incorreto e não confiável. Você deve mudar para um asynccanal baseado em - como 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();
}

    Observe como receber no canal agora envolve uma pollchamada interna do tipo - que toma o Contextcomo argumento. Se não houver uma mensagem para receber no momento, o Wakerserá associado ao canal (de uma forma específica da implementação) de modo que pode fazer com que o future seja pesquisado quando uma mensagem chegar.

    Saída:

    before
poll
poll
processsed
after

    Observe que há duas pesquisas. Presumivelmente, a primeira não encontrou nenhuma mensagem para receber. O uso adequado da asyncpesquisa significou que o futuro foi pesquisado produtivamente novamente.

    • 5

relate perguntas