Datagram sockets do not appear to participate in cooperative multitasking

[NoraCodes]

Version
tokio 1.31.0

Platform
uname -a: Linux enterprise 5.15.0-78-generic #85-Ubuntu SMP Fri Jul 7 15:25:09 UTC 2023 x86_64 x86_64 x86_64 GNU/Linux

Description

It seems like Unix datagram sockets don't correctly participate in cooperative multitasking. When a task continuously recv()s from a Unix datagram socket and there is always a datagram available in the socket, the recv() never blocks, the runtime never forces it to yield and the task never yields. Therefore, other tasks never have an opportunity to run.

In the following small program, I expect to see the debug representation of a std::time::SystemTime printed about every 250ms. When receiving an endless stream of Unix datagrams from 2 or 3 other processes running full tilt, I instead see one every second or every few seconds.

Our example uses a single-threaded runtime, but it also affects a multi-threaded runtime, which we observed as inconsistent but common delays in other tasks being run that happened to be scheduled on the same thread as the datagram recv() task.

use std::time::Duration;

use tokio::net::UnixDatagram;

// Single thread to better demonstrate the issue
#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let args: Vec<_> = std::env::args().collect();

    // Sending process runs only this, sends datagrams forever
    if args[1] == "sender" {
        // Wait until the receiver opens the socket
        tokio::time::sleep(Duration::from_millis(250)).await;
        let bytes = b"hello, tokio!";
        let tx = UnixDatagram::unbound()?;
        tx.connect("socket")?;
        loop {
            tx.send(bytes).await?;
        }
    } else {
        // Delete socket from past run
        let _ = tokio::fs::remove_file("socket").await;
        let rx = UnixDatagram::bind("socket")?;
        // Task that reads forever as fast as the socket will give it datagrams
        let a = tokio::task::spawn(async move {
            let mut buf = [0_u8; 13];
            loop {
                // Uncomment to fix
                //tokio::task::yield_now().await;
                rx.recv(&mut buf).await.unwrap();
                assert_eq!(&buf, b"hello, tokio!");
            }
        });
        // This task shows how often the runtime forces the above task to yield
        // Without the explicit yield, it's starved
        let b = tokio::task::spawn(async move {
            loop {
                tokio::time::sleep(Duration::from_millis(250)).await;
                println!("It's {:?}", std::time::SystemTime::now());
            }
        });
        let (_, _) = tokio::join!(a, b);
    }
    Ok(())
}

This is the output with the explicit yield, and what I would expect:

tokio_brokio on  main [?] is 📦 v0.1.0 via 🦀 v1.66.1 
❯ ./target/release/tokio_brokio recv & ./target/release/tokio_brokio sender & ./target/release/tokio_brokio sender & ./target/release/tokio_brokio sender             
[1] 3297719
[2] 3297720
[3] 3297721
It's SystemTime { tv_sec: 1692908737, tv_nsec: 224419890 }
It's SystemTime { tv_sec: 1692908737, tv_nsec: 475419527 }
It's SystemTime { tv_sec: 1692908737, tv_nsec: 726420228 }
It's SystemTime { tv_sec: 1692908737, tv_nsec: 977419866 }
It's SystemTime { tv_sec: 1692908738, tv_nsec: 228420222 }
It's SystemTime { tv_sec: 1692908738, tv_nsec: 479420486 }
It's SystemTime { tv_sec: 1692908738, tv_nsec: 730419806 }
It's SystemTime { tv_sec: 1692908738, tv_nsec: 981420162 }
It's SystemTime { tv_sec: 1692908739, tv_nsec: 232420042 }
It's SystemTime { tv_sec: 1692908739, tv_nsec: 483419352 }
It's SystemTime { tv_sec: 1692908739, tv_nsec: 734420485 }
It's SystemTime { tv_sec: 1692908739, tv_nsec: 985420387 }
It's SystemTime { tv_sec: 1692908740, tv_nsec: 236419035 }
It's SystemTime { tv_sec: 1692908740, tv_nsec: 487420190 }
It's SystemTime { tv_sec: 1692908740, tv_nsec: 738419797 }
It's SystemTime { tv_sec: 1692908740, tv_nsec: 989420219 }
It's SystemTime { tv_sec: 1692908741, tv_nsec: 240419320 }
It's SystemTime { tv_sec: 1692908741, tv_nsec: 491419883 }
It's SystemTime { tv_sec: 1692908741, tv_nsec: 742422658 }

This is the output without the explicit yield, showing that the timer task is starved of execution time

tokio_brokio on  main [?] is 📦 v0.1.0 via 🦀 v1.66.1
❯ ./target/release/tokio_brokio recv & ./target/release/tokio_brokio sender & ./target/release/tokio_brokio sender & ./target/release/tokio_brokio sender  
[1] 3296885
[2] 3296886
[3] 3296887
It's SystemTime { tv_sec: 1692908693, tv_nsec: 984578213 }
It's SystemTime { tv_sec: 1692908695, tv_nsec: 980149233 }
It's SystemTime { tv_sec: 1692908696, tv_nsec: 980901379 }
It's SystemTime { tv_sec: 1692908700, tv_nsec: 943152097 }
It's SystemTime { tv_sec: 1692908701, tv_nsec: 277230897 }

Co-authored with @jvranish

[Noah-Kennedy]

Thanks, it looks like this codepath was indeed never instrumented with budgeting.

I'll self-assign and set up a PR. I think I know of a few other things that have this issue.

[maminrayej]

@NoraCodes can you confirm the issue is resolved on your machine using the branch in PR #5967?

[NoraCodes]

With a slight modification, I was able to confirm that this issue also affects UDP sockets. The fix proposed in #5967 does resolve the issue on my machine.

[Noah-Kennedy]

This is now in progress.