Auto merge of rust-lang#95841 - ChrisDenton:pipe-server, r=m-ou-se

Windows: Use a pipe relay for chaining pipes

Fixes rust-lang#95759

This fixes the issue by chaining pipes synchronously and manually pumping messages between them. It's not ideal but it has the advantage of not costing anything if pipes are not chained ("don't pay for what you don't use") and it also avoids breaking existing code that rely on our end of the pipe being asynchronous (which includes rustc's own testing framework).

Libraries can avoid needing this by using their own pipes to chain commands.

Author: bors

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

@@ -165,6 +165,46 @@ pub fn anon_pipe(ours_readable: bool, their_handle_inheritable: bool) -> io::Res
     }
 }
 
+/// Takes an asynchronous source pipe and returns a synchronous pipe suitable
+/// for sending to a child process.
+///
+/// This is achieved by creating a new set of pipes and spawning a thread that
+/// relays messages between the source and the synchronous pipe.
+pub fn spawn_pipe_relay(
+    source: &AnonPipe,
+    ours_readable: bool,
+    their_handle_inheritable: bool,
+) -> io::Result<AnonPipe> {
+    // We need this handle to live for the lifetime of the thread spawned below.
+    let source = source.duplicate()?;
+
+    // create a new pair of anon pipes.
+    let Pipes { theirs, ours } = anon_pipe(ours_readable, their_handle_inheritable)?;
+
+    // Spawn a thread that passes messages from one pipe to the other.
+    // Any errors will simply cause the thread to exit.
+    let (reader, writer) = if ours_readable { (ours, source) } else { (source, ours) };
+    crate::thread::spawn(move || {
+        let mut buf = [0_u8; 4096];
+        'reader: while let Ok(len) = reader.read(&mut buf) {
+            if len == 0 {
+                break;
+            }
+            let mut start = 0;
+            while let Ok(written) = writer.write(&buf[start..len]) {
+                start += written;
+                if start == len {
+                    continue 'reader;
+                }
+            }
+            break;
+        }
+    });
+
+    // Return the pipe that should be sent to the child process.
+    Ok(theirs)
+}
+
 fn random_number() -> usize {
     static N: AtomicUsize = AtomicUsize::new(0);
     loop {
@@ -192,6 +232,9 @@ impl AnonPipe {
     pub fn into_handle(self) -> Handle {
         self.inner
     }
+    fn duplicate(&self) -> io::Result<Self> {
+        self.inner.duplicate(0, false, c::DUPLICATE_SAME_ACCESS).map(|inner| AnonPipe { inner })
+    }
 
     pub fn read(&self, buf: &mut [u8]) -> io::Result<usize> {
         let result = unsafe {

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

@@ -172,6 +172,7 @@ pub enum Stdio {
     Inherit,
     Null,
     MakePipe,
+    Pipe(AnonPipe),
     Handle(Handle),
 }
 
@@ -528,6 +529,11 @@ impl Stdio {
                 Ok(pipes.theirs.into_handle())
             }
 
+            Stdio::Pipe(ref source) => {
+                let ours_readable = stdio_id != c::STD_INPUT_HANDLE;
+                pipe::spawn_pipe_relay(source, ours_readable, true).map(AnonPipe::into_handle)
+            }
+
             Stdio::Handle(ref handle) => handle.duplicate(0, true, c::DUPLICATE_SAME_ACCESS),
 
             // Open up a reference to NUL with appropriate read/write
@@ -552,7 +558,7 @@ impl Stdio {
 
 impl From<AnonPipe> for Stdio {
     fn from(pipe: AnonPipe) -> Stdio {
-        Stdio::Handle(pipe.into_handle())
+        Stdio::Pipe(pipe)
     }
 }