Windows named pipes: process faults with STATUS_HEAP_CORRUPTION when clients are dropped repeatedly on a contended multi-threaded runtime

[dependencies]
tokio = { version = "1.53.1", features = ["rt-multi-thread", "macros", "time", "io-util", "net"] }
windows-sys = { version = "0.60", features = ["Win32_Foundation", "Win32_System_Threading"] }

//! Repeatedly connecting and dropping a `NamedPipeClient` on a multi-threaded runtime faults the
//! process (STATUS_HEAP_CORRUPTION / ACCESS_VIOLATION) when the machine is heavily oversubscribed.
//! Run with no arguments; it starts the load, the servers and the clients itself.
use std::{
    env,
    os::windows::io::AsRawHandle,
    process::{Child, Command, Stdio},
    time::{Duration, Instant},
};

const CLIENTS: usize = 12;
const CLIENTS_PER_PIPE: usize = 2;
const SPINNERS: usize = 8;
const SECONDS: u64 = 120;
const ROUNDS: usize = 4;
const IDLE: Duration = Duration::from_millis(500);

fn pipe(slot: usize) -> String {
    format!(r"\\.\pipe\np-crash-repro-{slot}")
}

fn main() {
    let slot: usize = env::args()
        .nth(2)
        .and_then(|value| value.parse().ok())
        .unwrap_or(0);
    match env::args().nth(1).as_deref() {
        Some("server") => server(slot),
        Some("client") => client(slot),
        Some("spin") => loop {
            std::hint::black_box(1);
        },
        _ => supervise(),
    }
}

fn affinity() -> usize {
    env::var("NP_AFFINITY")
        .ok()
        .and_then(|value| value.parse().ok())
        .unwrap_or(0b11)
}

fn spawn(mode: &str, slot: usize, pin: bool) -> Child {
    let child = Command::new(env::current_exe().unwrap())
        .arg(mode)
        .arg(slot.to_string())
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .spawn()
        .unwrap();
    if pin {
        // Two logical processors, so the runtime's worker threads and its I/O driver preempt each
        // other constantly. Any pair of busy cores reproduces this.
        let handle = child.as_raw_handle() as windows_sys::Win32::Foundation::HANDLE;
        unsafe { windows_sys::Win32::System::Threading::SetProcessAffinityMask(handle, affinity()) };
    }
    child
}

fn supervise() {
    let count: usize = env::var("NP_CLIENTS")
        .ok()
        .and_then(|value| value.parse().ok())
        .unwrap_or(CLIENTS);
    let servers = count / CLIENTS_PER_PIPE;
    let spinners: usize = env::var("NP_SPINNERS")
        .ok()
        .and_then(|value| value.parse().ok())
        .unwrap_or(SPINNERS);
    // Pinned to the same two processors as the servers and clients, so the starvation does not
    // depend on what else the machine happens to be doing.
    println!("starting {spinners} spinners, {servers} servers and {count} clients");
    let mut load: Vec<Child> = (0..spinners).map(|_| spawn("spin", 0, true)).collect();
    let mut hosts: Vec<Child> = (0..servers).map(|slot| spawn("server", slot, true)).collect();
    std::thread::sleep(Duration::from_millis(500));
    let mut clients: Vec<Child> = (0..count)
        .map(|index| spawn("client", index / CLIENTS_PER_PIPE, true))
        .collect();
    let seconds: u64 = env::var("NP_SECONDS")
        .ok()
        .and_then(|value| value.parse().ok())
        .unwrap_or(SECONDS);
    let deadline = Instant::now() + Duration::from_secs(seconds);
    let mut faulted = vec![];
    while Instant::now() < deadline && faulted.is_empty() {
        std::thread::sleep(Duration::from_millis(250));
        for client in &mut clients {
            if let Some(status) = client.try_wait().unwrap() {
                faulted.push(format!("client {} exited {status}", client.id()));
            }
        }
    }
    for child in clients.iter_mut().chain(load.iter_mut()).chain(hosts.iter_mut()) {
        let _ = child.kill();
    }
    if faulted.is_empty() {
        println!("no fault in {seconds}s");
    } else {
        // 0xC0000374 = STATUS_HEAP_CORRUPTION, 0xC0000005 = ACCESS_VIOLATION.
        for line in &faulted {
            println!("FAULT: {line}");
        }
    }
}

/// Several accept loops, so a client always finds a free instance and spends its time idling
/// between connections rather than retrying a busy pipe.
#[tokio::main]
async fn server(slot: usize) {
    use tokio::{
        io::{AsyncReadExt as _, AsyncWriteExt as _},
        net::windows::named_pipe::ServerOptions,
    };
    let mut loops = tokio::task::JoinSet::new();
    for _ in 0..4 {
        let name = pipe(slot);
        loops.spawn(async move {
            loop {
                let Ok(mut host) = ServerOptions::new().create(&name) else {
                    tokio::time::sleep(Duration::from_millis(5)).await;
                    continue;
                };
                if host.connect().await.is_err() {
                    continue;
                }
                tokio::spawn(async move {
                    let mut buffer = [0_u8; 8];
                    for _ in 0..ROUNDS {
                        if host.read_exact(&mut buffer).await.is_err() {
                            return;
                        }
                        if host.write_all(&buffer).await.is_err() {
                            return;
                        }
                    }
                });
            }
        });
    }
    loops.join_all().await;
}

#[tokio::main]
async fn client(slot: usize) {
    use tokio::{
        io::{AsyncReadExt as _, AsyncWriteExt as _},
        net::windows::named_pipe::ClientOptions,
    };
    loop {
        let mut client = match ClientOptions::new().open(pipe(slot)) {
            Ok(client) => client,
            // ERROR_PIPE_BUSY
            Err(error) if error.raw_os_error() == Some(231) => {
                tokio::time::sleep(Duration::from_millis(1)).await;
                continue;
            }
            Err(_) => {
                tokio::time::sleep(Duration::from_millis(10)).await;
                continue;
            }
        };
        let mut buffer = [0_u8; 8];
        for round in 0..ROUNDS {
            if client.write_all(&[round as u8; 8]).await.is_err() {
                break;
            }
            if client.read_exact(&mut buffer).await.is_err() {
                break;
            }
        }
        drop(client);
        // The I/O driver parks between connections; without this gap the fault does not appear.
        tokio::time::sleep(IDLE).await;
        let noise: Vec<Vec<u8>> = (0..16).map(|size| vec![7_u8; 128 + size * 91]).collect();
        std::hint::black_box(&noise);
    }
}

starting 8 spinners, 6 servers and 12 clients
FAULT: client 173160 exited exit code: 0xc0000374