Tokio :: Тайм-аут на Stream всегда происходит - PullRequest
Новая
       71

Tokio :: Тайм-аут на Stream всегда происходит

1 голос
/ 15 февраля 2020

Я пытаюсь принять UDP-сообщение, но только если это происходит в течение 5 секунд, у меня есть абстракция Stream, построенная как с помощью ручного внедрения Stream, так и с использованием комбинаторов в библиотеке futures. Каждый раз после разрешения будущего recv_from продолжительность истекает, а поток возвращает Err(Elapsed(())). Это не ожидаемое поведение, если возвращается значение, ошибка не должна возвращаться.

Ожидаемое поведение: поток разрешит или тайм-аут или Ve c, но не один, а остальные 5 секунд спустя. 

use futures::{pin_mut, ready, stream::unfold, FutureExt};
use tokio::{
    net::{udp, UdpSocket},
    stream::{Stream, StreamExt},
    time::{self, Duration},
};

use std::{
    io,
    net::SocketAddr,
    pin::Pin,
    task::{Context, Poll},
};

#[derive(Debug)]
pub(crate) struct UdpStream {
    stream: udp::RecvHalf,
}

impl UdpStream {
    fn new(stream: udp::RecvHalf) -> Self {
        Self { stream }
    }

    fn stream(self) -> impl Stream<Item = io::Result<(Vec<u8>, SocketAddr)>> {
        unfold(self.stream, |mut stream| async move {
            let mut buf = [0; 4096];
            match time::timeout(Duration::from_secs(5), stream.recv_from(&mut buf)).await {
                Ok(Ok((len, src))) => {
                    Some((Ok((buf.iter().take(len).cloned().collect(), src)), stream))
                }
                e => {
                    println!("{:?}", e);
                    None
                }
            }
        })
    }
}

impl Stream for UdpStream {
    type Item = io::Result<(Vec<u8>, SocketAddr)>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let socket = &mut self.stream;
        pin_mut!(socket);
        let mut buf = [0u8; 4096];
        let (len, src) = ready!(Box::pin(socket.recv_from(&mut buf)).poll_unpin(cx))?;
        Poll::Ready(Some(Ok((buf.iter().take(len).cloned().collect(), src))))
    }
}

async fn listen_udp(addr: SocketAddr) -> io::Result<()> {
    let udp = UdpSocket::bind(addr).await?;
    let (mut udp_recv, mut udp_send) = udp.split();

    let mut msg_stream = Box::pin(UdpStream::new(udp_recv).stream());
    // use the manually implemented stream with this:
    // let mut msg_stream = UdpStream::new(udp_recv).timeout(Duration::from_secs(5));

    while let Some(msg) = msg_stream.next().await {
        match msg {
            Ok((buf, src)) => {
                udp_send.send_to(&buf, &src).await?;
                println!("Message recv: {:?}", buf);
            }
            Err(e) => {
                eprintln!("timed out: {:?}", e);
            }
        }
    }
    Ok(())
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync + 'static>> {
    listen_udp("127.0.0.1:9953".parse()?).await?;
    Ok(())
}

Вы можете попробовать запустить этот код и сделать запросы udp с echo "foo" | nc 127.0.0.1 9953 -u или с dig

car go .toml 

[package]
name = "udp_test"
version = "0.1.0"
edition = "2018"

# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

[dependencies]
tokio = { version = "0.2", features = ["full"] }
futures = "0.3"

1 Ответ

0 голосов
/ 15 февраля 2020

После того, как ваш поток возвращает первый (и единственный) элемент, он возвращается к ожиданию следующего; оно никогда не заканчивается, пока не истечет время ожидания.

В принципе, абстракция потока здесь не нужна. future, завернутый в тайм-аут, будет: 

use std::{io, net::SocketAddr};
use tokio::{
    net::UdpSocket,
    time::{self, Duration},
};

async fn listen_udp(addr: SocketAddr) -> io::Result<()> {
    let mut udp = UdpSocket::bind(addr).await?;
    let mut buf = [0; 4096];

    match time::timeout(Duration::from_secs(5), udp.recv_from(&mut buf)).await? {
        Ok((count, src)) => {
            udp.send_to(&buf[..count], &src).await?;
            println!("Message recv: {:?}", &buf[..count]);
        }
        Err(e) => {
            eprintln!("timed out: {:?}", e);
        }
    }
    Ok(())
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    listen_udp("127.0.0.1:9953".parse()?).await?;
    Ok(())
}
...