nuclei 0.5.4

//! Byte streams that are not sockets, on Windows: a pipe opened for
//! synchronous IO, whose calls are jobs of the blocking pool, and one
//! opened for overlapped IO, whose operations complete on the port.

use std::os::windows::fs::OpenOptionsExt;
use std::os::windows::io::{AsRawHandle, FromRawHandle, OwnedHandle};

use super::*;
use crate::io::raw::Raw;
use crate::io::source::Kind;
use crate::syscore::windows::{
    CreateNamedPipeW, CreatePipe, FILE_FLAG_OVERLAPPED, HANDLE, INVALID_HANDLE_VALUE,
    PIPE_ACCESS_DUPLEX,
};

/// An anonymous pipe: its ends are opened for synchronous IO.
fn synchronous_pipe() -> (File, File) {
    let (mut r, mut w): (HANDLE, HANDLE) = (core::ptr::null_mut(), core::ptr::null_mut());
    // SAFETY: the call writes two new handles into our own.
    assert_ne!(
        unsafe { CreatePipe(&mut r, &mut w, core::ptr::null(), 0) },
        0
    );
    // SAFETY: both handles are new and ours alone.
    unsafe {
        (
            File::from(OwnedHandle::from_raw_handle(r.cast())),
            File::from(OwnedHandle::from_raw_handle(w.cast())),
        )
    }
}

/// A named pipe and a client of it, both ends opened for overlapped IO.
fn overlapped_pipe() -> (File, File) {
    let name = format!(
        r"\\.\pipe\nuclei-{}-{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    );
    let wide: Vec<u16> = name.encode_utf16().chain([0]).collect();
    // SAFETY: a new pipe of our own name; the name lives for the call.
    let server = unsafe {
        CreateNamedPipeW(
            wide.as_ptr(),
            PIPE_ACCESS_DUPLEX | FILE_FLAG_OVERLAPPED,
            0,
            1,
            64 * 1024,
            64 * 1024,
            0,
            core::ptr::null(),
        )
    };
    assert_ne!(server, INVALID_HANDLE_VALUE);
    let client = OpenOptions::new()
        .read(true)
        .write(true)
        .custom_flags(FILE_FLAG_OVERLAPPED)
        .open(&name)
        .unwrap();
    // SAFETY: the handle is new and ours alone. The client is connected:
    // opening it connected the server end.
    (
        File::from(unsafe { OwnedHandle::from_raw_handle(server.cast()) }),
        client,
    )
}

fn raw(file: &File) -> Raw {
    file.as_raw_handle() as usize as Raw
}

/// Bytes go through `writer` and come out of `reader` whole and in order,
/// and the end of the writer is the reader's end of stream.
fn carries_bytes_then_the_end(reader: File, writer: File) {
    let rt = runtime();
    rt.block_on(async {
        let mut reader = Handle::new(reader).unwrap();
        assert_eq!(reader.source().kind(), Kind::Bytes);
        let data = pattern(300 * 1024);
        let expected = data.clone();
        let writer = spawn(async move {
            let mut writer = Handle::new(writer).unwrap();
            within(write_all(&mut writer, &data)).await.unwrap();
            within(flush(&mut writer)).await.unwrap();
        });
        let mut got = vec![0u8; expected.len()];
        let mut at = 0;
        while at < got.len() {
            match within(read_some(&mut reader, &mut got[at..]))
                .await
                .unwrap()
            {
                0 => panic!("the stream ended after {at} bytes"),
                n => at += n,
            }
        }
        within(writer).await;
        assert!(got == expected, "the bytes arrived changed or out of order");
        let mut more = [0u8; 8];
        assert_eq!(within(read_some(&mut reader, &mut more)).await.unwrap(), 0);
    });
}

#[test]
fn a_synchronous_pipe_carries_bytes_through_the_pools_jobs() {
    let (r, w) = synchronous_pipe();
    assert!(crate::io::sys::synchronous(raw(&r)).unwrap());
    carries_bytes_then_the_end(r, w);
}

#[test]
fn an_overlapped_pipe_carries_bytes_through_the_port() {
    let (server, client) = overlapped_pipe();
    assert!(!crate::io::sys::synchronous(raw(&server)).unwrap());
    carries_bytes_then_the_end(server, client);
}

/// The overlapped pipe's ends are duplex: each reads what the other wrote,
/// with lent buffers.
#[test]
fn an_overlapped_pipe_is_duplex_with_lent_buffers() {
    let rt = runtime();
    let (server, client) = overlapped_pipe();
    rt.block_on(async {
        let server = Handle::new(server).unwrap();
        let client = Handle::new(client).unwrap();
        let (n, _) = within(client.write_owned(b"ping".to_vec())).await;
        assert_eq!(n.unwrap(), 4);
        let (n, got) = within(server.read_owned(Vec::with_capacity(16))).await;
        assert_eq!((n.unwrap(), &got[..]), (4, &b"ping"[..]));
        let (n, _) = within(server.write_owned(b"pong".to_vec())).await;
        assert_eq!(n.unwrap(), 4);
        let (n, got) = within(client.read_owned(Vec::with_capacity(16))).await;
        assert_eq!((n.unwrap(), &got[..]), (4, &b"pong"[..]));
    });
}

/// A read that waits on a synchronous pipe holds a thread of the pool in
/// its call: handing the file back cancels that call, so it returns, and
/// the read loses no byte that came after.
#[test]
fn a_synchronous_pipes_waiting_read_is_cancelled_when_the_file_is_handed_back() {
    let rt = runtime();
    let (r, w) = synchronous_pipe();
    let reader = rt.block_on(async {
        let reader = Handle::new(r).unwrap();
        let mut buf = [0u8; 8];
        let mut waiting = Box::pin(read_some_ref(&reader, &mut buf));
        let polled = poll_fn(|cx| core::task::Poll::Ready(waiting.as_mut().poll(cx))).await;
        assert!(polled.is_pending(), "nothing was written yet");
        drop(waiting);
        // Let the job reach its call.
        crate::time::sleep(Duration::from_millis(20)).await;
        reader
    });
    let (tell, handed) = std::sync::mpsc::channel();
    std::thread::spawn(move || tell.send(reader.into_inner()).unwrap());
    let mut file = handed
        .recv_timeout(Duration::from_secs(10))
        .expect("the waiting read was never cancelled");
    (&w).write_all_now(b"after");
    let mut got = [0u8; 5];
    file.read_exact(&mut got).unwrap();
    assert_eq!(&got, b"after");
}

/// A lent read dropped while its job waits in its call: the next read gets
/// every byte written after, none twice and none lost.
#[test]
fn a_dropped_read_on_a_synchronous_pipe_loses_nothing() {
    let rt = runtime();
    let (r, w) = synchronous_pipe();
    rt.block_on(async {
        let reader = Handle::new(r).unwrap();
        let mut lent = Box::pin(reader.read_owned(Vec::with_capacity(4)));
        let polled = poll_fn(|cx| core::task::Poll::Ready(lent.as_mut().poll(cx))).await;
        assert!(polled.is_pending(), "nothing was written yet");
        drop(lent);
        (&w).write_all_now(b"abcdefgh");
        let mut got = Vec::new();
        while got.len() < 8 {
            let (n, buf) = within(reader.read_owned(Vec::with_capacity(8))).await;
            assert!(n.unwrap() > 0);
            got.extend_from_slice(&buf);
        }
        assert_eq!(got, b"abcdefgh");
    });
}

/// Writes through `writer` until a write does not complete (the pipe's
/// buffer is full, and nothing reads it), drops that write, and checks
/// that none of it ever comes out of `reader`: the write was withdrawn.
fn withdraws_a_write_the_pipe_never_took(reader: File, writer: File) {
    let rt = runtime();
    let (taken, reader) = rt.block_on(async {
        let writer = Handle::new(writer).unwrap();
        let mut taken = 0;
        loop {
            let mut write = Box::pin(writer.write_owned(vec![7u8; 4096]));
            match crate::time::timeout(Duration::from_millis(200), &mut write).await {
                Ok((n, _)) => {
                    taken += n.unwrap();
                    assert!(taken < 64 << 20, "the pipe never filled");
                }
                Err(_) => {
                    drop(write);
                    break;
                }
            }
        }
        // The withdrawal is done once the cancelled write came back.
        let deadline = Instant::now() + Duration::from_secs(10);
        while writer.source().has_unsent() {
            assert!(
                Instant::now() < deadline,
                "the withdrawn write is still out"
            );
            crate::time::sleep(Duration::from_millis(5)).await;
        }
        drop(writer);
        let mut reader = Handle::new(reader).unwrap();
        let mut got = 0;
        let mut buf = vec![0u8; 64 * 1024];
        loop {
            match within(read_some(&mut reader, &mut buf)).await.unwrap() {
                0 => break,
                n => got += n,
            }
        }
        (taken, got)
    });
    assert_eq!(reader, taken, "bytes of the withdrawn write came out");
}

#[test]
fn a_dropped_write_a_synchronous_pipe_never_took_is_withdrawn() {
    let (r, w) = synchronous_pipe();
    withdraws_a_write_the_pipe_never_took(r, w);
}

#[test]
fn a_dropped_write_an_overlapped_pipe_never_took_is_withdrawn() {
    let (server, client) = overlapped_pipe();
    withdraws_a_write_the_pipe_never_took(server, client);
}

async fn read_some_ref(r: &Handle<File>, buf: &mut [u8]) -> io::Result<usize> {
    let mut r = r;
    poll_fn(|cx| Pin::new(&mut r).poll_read(cx, buf)).await
}

/// Writes with std, from the test.
trait WriteNow {
    fn write_all_now(self, bytes: &[u8]);
}

impl WriteNow for &File {
    fn write_all_now(mut self, bytes: &[u8]) {
        std::io::Write::write_all(&mut self, bytes).unwrap();
    }
}

/// A disk file opened for overlapped IO, with no buffering and its writes through, as a
/// database opens its log: its calls run on the pool's threads, and each one the handle answers
/// pending (ERROR_IO_PENDING) waits for its own end, so none answers a false error and the
/// kernel writes into no frame or buffer the call left. Cursor writes of whole pages and
/// positional writes of other pages at once, a sync, and every byte read back.
#[test]
fn an_overlapped_unbuffered_file_writes_and_reads_every_byte() {
    use crate::io::AlignedBuf;
    /// Opening flags of a file the system neither caches nor holds its writes for.
    const FILE_FLAG_NO_BUFFERING: u32 = 0x2000_0000;
    const FILE_FLAG_WRITE_THROUGH: u32 = 0x8000_0000;
    const PAGE: usize = 4096;
    const PAGES: usize = 16;
    let page = |i: usize| -> AlignedBuf {
        let mut buf = AlignedBuf::new(PAGE, PAGE).unwrap();
        let bytes: Vec<u8> = (0..PAGE)
            .map(|j| (i * 131 + j * 7 + j / 251) as u8)
            .collect();
        assert_eq!(buf.extend_from_slice(&bytes), PAGE);
        buf
    };
    let dir = Temp::new("overlapped");
    let path = dir.file("log");
    let opened = std::fs::OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .custom_flags(FILE_FLAG_OVERLAPPED | FILE_FLAG_NO_BUFFERING | FILE_FLAG_WRITE_THROUGH)
        .open(&path)
        .unwrap();
    let rt = runtime();
    rt.block_on(async {
        let file = Handle::new(opened).unwrap();
        let cursor = async {
            for i in 0..PAGES {
                let (wrote, _) = within(file.write_owned(page(i))).await;
                assert_eq!(wrote.unwrap(), PAGE, "cursor page {i}");
            }
        };
        let positional = async {
            for i in PAGES..2 * PAGES {
                let (wrote, _) = within(file.write_at_owned(page(i), (i * PAGE) as u64)).await;
                assert_eq!(wrote.unwrap(), PAGE, "positional page {i}");
            }
        };
        crate::join!(cursor, positional);
        within(file.sync_all()).await.unwrap();
        for i in 0..2 * PAGES {
            let (read, got) =
                within(file.read_at_owned(AlignedBuf::new(PAGE, PAGE).unwrap(), (i * PAGE) as u64))
                    .await;
            assert_eq!(read.unwrap(), PAGE, "page {i}");
            assert!(got[..] == page(i)[..], "page {i} differs");
        }
    });
}