Почему я не получаю пробуждение по нескольким фьючерсам, когда они используют один и тот же базовый сокет? - PullRequest
Новая
       75

Почему я не получаю пробуждение по нескольким фьючерсам, когда они используют один и тот же базовый сокет?

1 голос
/ 01 апреля 2020

У меня есть код, который отправляет данные на несколько конечных точек UDP с использованием одного и того же локального UdpSocket: 

use futures::stream::FuturesUnordered;
use futures::StreamExt;
use std::{
    future::Future,
    net::{Ipv4Addr, SocketAddr},
    pin::Pin,
    task::{Context, Poll},
};
use tokio::net::UdpSocket;

#[tokio::main]
async fn main() {
    let server_0: SocketAddr = (Ipv4Addr::UNSPECIFIED, 12000).into();
    let server_2: SocketAddr = (Ipv4Addr::UNSPECIFIED, 12002).into();
    let server_1: SocketAddr = (Ipv4Addr::UNSPECIFIED, 12001).into();

    tokio::spawn(start_server(server_0));
    tokio::spawn(start_server(server_1));
    tokio::spawn(start_server(server_2));

    let client_addr: SocketAddr = (Ipv4Addr::UNSPECIFIED, 12004).into();
    let socket = UdpSocket::bind(client_addr).await.unwrap();

    let mut futs = FuturesUnordered::new();
    futs.push(Task::new(0, &socket, &server_0));
    futs.push(Task::new(1, &socket, &server_1));
    futs.push(Task::new(2, &socket, &server_2));

    while let Some(n) = futs.next().await {
        println!("Done: {:?}", n)
    }
}

async fn start_server(addr: SocketAddr) {
    let mut socket = UdpSocket::bind(addr).await.unwrap();
    let mut buf = [0; 512];
    loop {
        println!("{:?}", socket.recv_from(&mut buf).await);
    }
}

struct Task<'a> {
    value: u32,
    socket: &'a UdpSocket,
    addr: &'a SocketAddr,
}

impl<'a> Task<'a> {
    fn new(value: u32, socket: &'a UdpSocket, addr: &'a SocketAddr) -> Self {
        Self {
            value,
            socket,
            addr,
        }
    }
}

impl Future for Task<'_> {
    type Output = Option<u32>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        println!("Polling for {}", self.value);
        let buf = &self.value.to_be_bytes();

        match self.socket.poll_send_to(cx, buf, self.addr) {
            Poll::Ready(Ok(_)) => {
                println!("Got Ok for {}", self.value);
                Poll::Ready(Some(self.value))
            }
            Poll::Ready(Err(_)) => {
                println!("Got err for {}", self.value);
                Poll::Ready(None)
            }
            Poll::Pending => {
                println!("Got pending for {}", self.value);
                Poll::Pending
            }
        }
    }
}

Иногда он застревает после записи только одного из данных, печатая: 

Polling for 0
Got pending for 0
Polling for 1
Got pending for 1
Polling for 2
Got pending for 2
Polling for 2
Got Ok for 2
Done: Some(2)
Ok((4, V4(127.0.0.1:12004)))

В этом случае задачи со значениями 0 и 1 никогда не пробуждаются. Как мне надежно дать им сигнал разбудить их?

Я пытался дозвониться до cx.waker().wake_by_ref() при получении Poll::Ready, так как думал, что это может разбудить и других, но это не так.

1 Ответ

1 голос
/ 01 апреля 2020

Когда poll_send_to возвращает Poll::Pending, это гарантирует отправку сигнала вызова Waker, указанного в контексте, с которым был опрошен. Однако требуется только отправить пробуждение до последнего Waker, с которым он был опрошен. Это означает, что, поскольку вы вызываете poll_send_to в одном и том же сокете из нескольких задач, сокет только пообещал отправить пробуждение тому, который его опрашивал последним.

Это также объясняет, почему это работает: 

let mut futs = Vec::new();
futs.push(Task::new(0, &socket, &server_0));
futs.push(Task::new(1, &socket, &server_1));
futs.push(Task::new(2, &socket, &server_2));

for n in join_all(futs).await {
    println!("Done: {:?}", n)
}

В отличие от FuturesUnordered, комбинатор join_all будет опрашивать каждое внутреннее будущее каждый раз, когда он опрашивается, но FuturesUnordered отслеживает, из какого базового будущего произошло пробуждение.

См. Также эту тему .

...