Starvation with UdpFramed stream

[aegooby]

When using a UdpFramed with split(), I am looking to have a loop to send out broadcast pings and another loop to listen for them, on separate tasks. The read loop is pretty similar to the example in the docs, along with the write loop,

let (sink, stream) = UdpFramed::new(...);
let sink_fut = spawn(async move {
    loop {
        sink.send(...).await;
        sleep(Duration::from_millis(1500)).await;
    }
});
let stream_fut = spawn(async move {
    while let Some(item) = stream.next().await {
        ...
    }
});
sink_fut.await;
stream_fut.await;

The docs say that the stream task would cause starvation, but then say that "To account for this, Tokio has explicit yield points in a number of library functions, which force tasks to return to the executor periodically." I can't tell if I am doing something wrong or if UdpFramed doesn't have these yield points, because the stream task seems to hog the IO and prevent everything else running.

[Darksonn]

There is no problem in the code you have posted. Perhaps you removed the line with the problem when you simplified the example?

[aegooby]

Yeah I did oversimplify it there. This example pasted into main.rs causes issues when I run it:

use std::{
    net::SocketAddr,
    time::Duration,
};

use anyhow::{Ok, Result};
use bytes::BytesMut;
use futures_util::{SinkExt, StreamExt};
use socket2::{Domain, Protocol, SockAddr, Socket, Type};
use tokio::{main, net::UdpSocket, spawn, time::sleep};
use tokio_util::{codec::BytesCodec, udp::UdpFramed};

#[main(flavor = "multi_thread")]
pub async fn main() -> Result<()> {
    let socket_2 = Socket::new(Domain::IPV4, Type::DGRAM, Some(Protocol::UDP))?;
    socket_2.set_reuse_address(true)?;
    socket_2.set_reuse_port(true)?;
    socket_2.bind(&SockAddr::from("0.0.0.0:8080".parse::<SocketAddr>()?))?;
    socket_2.set_broadcast(true)?;

    let framed = UdpFramed::new(UdpSocket::from_std(socket_2.into())?, BytesCodec::new());
    let (mut sink, mut stream) = framed.split();
    let sink_fut = spawn(async move {
        loop {
            sink.send((BytesMut::from_iter([1]), "255.255.255.255:8080".parse()?))
                .await?;
            sleep(Duration::from_millis(1500)).await;
            if false {
                break Ok(());
            }
        }
    });
    let stream_fut = spawn(async move {
        while let Some(item) = stream.next().await {
            let (data, _) = item?;
            println!("{data:#?}");
        }
        Ok(())
    });
    sink_fut.await;
    stream_fut.await;
    Ok(())
}

[luben]

You have to set the socket in non-blocking mode before giving it to Tokio: https://docs.rs/socket2/latest/socket2/struct.Socket.html#method.set_nonblocking

[aegooby]

That worked. Thanks for the help I hope you all become very rich billionaires (and maybe give me a couple million).