Auto merge of #98950 - ChrisDenton:getoverlapped-io, r=thomcc

Windows: Fallback for overlapped I/O

Fixes #98947

library/std/src/sys/windows/c.rs

@@ -326,7 +326,9 @@ union IO_STATUS_BLOCK_union {
 }
 impl Default for IO_STATUS_BLOCK_union {
     fn default() -> Self {
-        Self { Pointer: ptr::null_mut() }
+        let mut this = Self { Pointer: ptr::null_mut() };
+        this.Status = STATUS_PENDING;
+        this
     }
 }
 #[repr(C)]
@@ -335,6 +337,16 @@ pub struct IO_STATUS_BLOCK {
     u: IO_STATUS_BLOCK_union,
     pub Information: usize,
 }
+impl IO_STATUS_BLOCK {
+    pub fn status(&self) -> NTSTATUS {
+        // SAFETY: If `self.u.Status` was set then this is obviously safe.
+        // If `self.u.Pointer` was set then this is the equivalent to converting
+        // the pointer to an integer, which is also safe.
+        // Currently the only safe way to construct `IO_STATUS_BLOCK` outside of
+        // this module is to call the `default` method, which sets the `Status`.
+        unsafe { self.u.Status }
+    }
+}
 
 pub type LPOVERLAPPED_COMPLETION_ROUTINE = unsafe extern "system" fn(
     dwErrorCode: DWORD,

library/std/src/sys/windows/handle.rs

@@ -1,5 +1,8 @@
 #![unstable(issue = "none", feature = "windows_handle")]
 
+#[cfg(test)]
+mod tests;
+
 use crate::cmp;
 use crate::io::{self, ErrorKind, IoSlice, IoSliceMut, Read, ReadBuf};
 use crate::mem;
@@ -248,14 +251,18 @@ impl Handle {
             offset.map(|n| n as _).as_ref(),
             None,
         );
+
+        let status = if status == c::STATUS_PENDING {
+            c::WaitForSingleObject(self.as_raw_handle(), c::INFINITE);
+            io_status.status()
+        } else {
+            status
+        };
         match status {
             // If the operation has not completed then abort the process.
             // Doing otherwise means that the buffer and stack may be written to
             // after this function returns.
-            c::STATUS_PENDING => {
-                eprintln!("I/O error: operation failed to complete synchronously");
-                crate::process::abort();
-            }
+            c::STATUS_PENDING => rtabort!("I/O error: operation failed to complete synchronously"),
 
             // Return `Ok(0)` when there's nothing more to read.
             c::STATUS_END_OF_FILE => Ok(0),
@@ -294,13 +301,17 @@ impl Handle {
                 None,
             )
         };
+        let status = if status == c::STATUS_PENDING {
+            unsafe { c::WaitForSingleObject(self.as_raw_handle(), c::INFINITE) };
+            io_status.status()
+        } else {
+            status
+        };
         match status {
             // If the operation has not completed then abort the process.
             // Doing otherwise means that the buffer may be read and the stack
             // written to after this function returns.
-            c::STATUS_PENDING => {
-                rtabort!("I/O error: operation failed to complete synchronously");
-            }
+            c::STATUS_PENDING => rtabort!("I/O error: operation failed to complete synchronously"),
 
             // Success!
             status if c::nt_success(status) => Ok(io_status.Information),

library/std/src/sys/windows/handle/tests.rs

@@ -0,0 +1,22 @@
+use crate::sys::pipe::{anon_pipe, Pipes};
+use crate::{thread, time};
+
+/// Test the synchronous fallback for overlapped I/O.
+#[test]
+fn overlapped_handle_fallback() {
+    // Create some pipes. `ours` will be asynchronous.
+    let Pipes { ours, theirs } = anon_pipe(true, false).unwrap();
+
+    let async_readable = ours.into_handle();
+    let sync_writeable = theirs.into_handle();
+
+    thread::scope(|_| {
+        thread::sleep(time::Duration::from_millis(100));
+        sync_writeable.write(b"hello world!").unwrap();
+    });
+
+    // The pipe buffer starts empty so reading won't complete synchronously unless
+    // our fallback path works.
+    let mut buffer = [0u8; 1024];
+    async_readable.read(&mut buffer).unwrap();
+}

src/test/ui-fulldeps/issue-81357-unsound-file-methods.rs

@@ -0,0 +1,81 @@
+// run-fail
+// only-windows
+
+fn main() {
+    use std::fs;
+    use std::io::prelude::*;
+    use std::os::windows::prelude::*;
+    use std::ptr;
+    use std::sync::Arc;
+    use std::thread;
+    use std::time::Duration;
+
+    const FILE_FLAG_OVERLAPPED: u32 = 0x40000000;
+
+    fn create_pipe_server(path: &str) -> fs::File {
+        let mut path0 = path.as_bytes().to_owned();
+        path0.push(0);
+        extern "system" {
+            fn CreateNamedPipeA(
+                lpName: *const u8,
+                dwOpenMode: u32,
+                dwPipeMode: u32,
+                nMaxInstances: u32,
+                nOutBufferSize: u32,
+                nInBufferSize: u32,
+                nDefaultTimeOut: u32,
+                lpSecurityAttributes: *mut u8,
+            ) -> RawHandle;
+        }
+
+        unsafe {
+            let h = CreateNamedPipeA(path0.as_ptr(), 3, 0, 1, 0, 0, 0, ptr::null_mut());
+            assert_ne!(h as isize, -1);
+            fs::File::from_raw_handle(h)
+        }
+    }
+
+    let path = "\\\\.\\pipe\\repro";
+    let mut server = create_pipe_server(path);
+
+    let client = Arc::new(
+        fs::OpenOptions::new().custom_flags(FILE_FLAG_OVERLAPPED).read(true).open(path).unwrap(),
+    );
+
+    let spawn_read = |is_first: bool| {
+        thread::spawn({
+            let f = client.clone();
+            move || {
+                let mut buf = [0xcc; 1];
+                let mut f = f.as_ref();
+                f.read(&mut buf).unwrap();
+                if is_first {
+                    assert_ne!(buf[0], 0xcc);
+                } else {
+                    let b = buf[0]; // capture buf[0]
+                    thread::sleep(Duration::from_millis(200));
+
+                    // Check the buffer hasn't been written to after read.
+                    dbg!(buf[0], b);
+                    assert_eq!(buf[0], b);
+                }
+            }
+        })
+    };
+
+    let t1 = spawn_read(true);
+    thread::sleep(Duration::from_millis(20));
+    let t2 = spawn_read(false);
+    thread::sleep(Duration::from_millis(100));
+    let _ = server.write(b"x");
+    thread::sleep(Duration::from_millis(100));
+    let _ = server.write(b"y");
+
+    // This is run fail because we need to test for the `abort`.
+    // That failing to run is the success case.
+    if t1.join().is_err() || t2.join().is_err() {
+        return;
+    } else {
+        panic!("success");
+    }
+}