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

[rainyflash]

Summary

On Windows, a process that repeatedly opens a NamedPipeClient, exchanges a few messages and drops it faults with 0xC0000374 (STATUS_HEAP_CORRUPTION) or 0xC0000005 (ACCESS_VIOLATION) when the machine is heavily oversubscribed. There is no Rust panic and no output — the process simply disappears.

The same workload never faults when the pipes live on a current_thread runtime, and never faults when everything is pinned to a single logical processor. Both point at the I/O driver processing a pipe's completions on one thread while another thread drops that pipe.

Everything in the reproducer is safe code using only tokio::net::windows::named_pipe.

Environment

• Windows 11 Pro 26200, x86-64, 32 logical processors
• rustc 1.97.1 MSVC, dev profile
• tokio 1.53.1, mio 1.2.2

Reproducer

Cargo.toml:

[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"] }

src/main.rs (windows-sys is only used by the harness, to pin the child processes):

//! 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);
    }
}

Run cargo run and wait. On a faulting run:

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

It is probabilistic: roughly one faulting client per three to six 90-second runs here (NP_SECONDS=90 cargo run), and noticeably more often when the machine is busy with unrelated work. The ingredients that all seem to be needed are: a multi-threaded runtime, repeated connect/exchange/drop, an idle gap between connections (the driver parks), and two processors that are genuinely contended.

What changes the outcome

Measured on the same machine, alternating configurations to avoid drift:

Change	Result
multi-threaded runtime (baseline, our application)	8 faulting runs / 11
#[tokio::main(flavor = "current_thread")], same workload	0 / 10
everything pinned to one logical processor (NP_AFFINITY=1)	0 / 4
no CPU contention, 24 client processes, ~9,600 connections	0 faults
cancelling in-flight operations (~1.5M cancelled connects/reads) with no contention	0 faults
dedicated pipe per client vs. two clients per pipe	both fault
interprocess 2.4.3 wrapper instead of bare NamedPipeClient	both fault

So it is not cancellation, not the ERROR_PIPE_BUSY retry path, not connection volume, and not a wrapper crate — it needs the driver and the dropping thread to run at the same time.

Where it dies

I instrumented the application's client with phase markers (credentials → connect → handshake → request → drop). In both crashes I captured, the last marker before the fault was at the connect ↔ drop boundary, which is consistent with a completion being written into memory that has already been released.

The only thing I can point at from reading src/sys/windows/named_pipe.rs is that Drop for NamedPipe issues CancelIoEx for a pending read and returns without waiting, relying on the leaked Arc reference being reclaimed when the completion is dequeued. I could not turn that into a proof — Application Verifier and page heap need elevation I do not have on this machine — so please treat it as a starting point rather than a diagnosis.

Impact

This was found in a Rust application that uses a named pipe for local IPC between a desktop service and its CLI: the CLI died silently on a loaded CI runner. Moving the CLI to a current_thread runtime worked around it, but servers that genuinely need a multi-threaded runtime cannot do that.

I am happy to run patches, try different configurations, or collect more data on this machine.