Update AsyncProcess output streaming on Windows

Author: owenv

Sources/Basics/Concurrency/AsyncProcess.swift

@@ -514,15 +514,16 @@ package final class AsyncProcess {
         if self.outputRedirection.redirectsOutput {
             let stdoutPipe = Pipe()
             let stderrPipe = Pipe()
+            let stdoutStream = DispatchFD(fileHandle: stdoutPipe.fileHandleForReading).dataStream()
+            let stderrStream = DispatchFD(fileHandle: stderrPipe.fileHandleForReading).dataStream()
 
             group.enter()
-            stdoutPipe.fileHandleForReading.readabilityHandler = { (fh: FileHandle) in
-                let data = (try? fh.read(upToCount: Int.max)) ?? Data()
-                if data.count == 0 {
-                    stdoutPipe.fileHandleForReading.readabilityHandler = nil
+            Task {
+                defer {
                     group.leave()
-                } else {
-                    let contents = data.withUnsafeBytes { [UInt8]($0) }
+                }
+                for try await data in stdoutStream {
+                    let contents = [UInt8](data)
                     self.outputRedirection.outputClosures?.stdoutClosure(contents)
                     stdoutLock.withLock {
                         stdout += contents
@@ -531,16 +532,15 @@ package final class AsyncProcess {
             }
 
             group.enter()
-            stderrPipe.fileHandleForReading.readabilityHandler = { (fh: FileHandle) in
-                let data = (try? fh.read(upToCount: Int.max)) ?? Data()
-                if data.count == 0 {
-                    stderrPipe.fileHandleForReading.readabilityHandler = nil
+            Task {
+                defer {
                     group.leave()
-                } else {
-                    let contents = data.withUnsafeBytes { [UInt8]($0) }
-                    self.outputRedirection.outputClosures?.stderrClosure(contents)
-                    stderrLock.withLock {
-                        stderr += contents
+                }
+                for try await data in stderrStream {
+                    let contents = [UInt8](data)
+                    self.outputRedirection.outputClosures?.stdoutClosure(contents)
+                    stdoutLock.withLock {
+                        stdout += contents
                     }
                 }
             }
@@ -1354,3 +1354,51 @@ extension FileHandle: WritableByteStream {
     }
 }
 #endif
+
+extension DispatchFD {
+    public func readChunk(upToLength maxLength: Int) async throws -> DispatchData {
+        return try await withCheckedThrowingContinuation { continuation in
+            DispatchIO.read(fromFileDescriptor: numericCast(self.rawValue), maxLength: maxLength, runningHandlerOn: DispatchQueue.global())
+                { data, error in
+                    if error != 0 {
+                        continuation.resume(throwing: StringError("POSIX error: \(error)"))
+                        return
+                    }
+                    continuation.resume(returning: data)
+                }
+            }
+
+        }
+
+    /// Returns an async stream which reads bytes from the specified file descriptor. Unlike `FileHandle.bytes`, it does not block the caller.
+    @available(macOS 15.0, iOS 18.0, tvOS 18.0, watchOS 11.0, visionOS 2.0, *)
+    public func dataStream() -> some AsyncSequence<DispatchData, any Error> {
+        AsyncThrowingStream<DispatchData, any Error> {
+            while !Task.isCancelled {
+                let chunk = try await readChunk(upToLength: 4096)
+                if chunk.isEmpty {
+                    return nil
+                }
+                return chunk
+            }
+            throw CancellationError()
+        }
+    }
+}
+
+public struct DispatchFD {
+    #if os(Windows)
+    fileprivate let rawValue: Int
+    #else
+    fileprivate let rawValue: Int32
+    #endif
+
+    init(fileHandle: FileHandle) {
+        #if os(Windows)
+        // This may look unsafe, but is how swift-corelibs-dispatch works. Basically, dispatch_fd_t directly represents either a POSIX file descriptor OR a Windows HANDLE pointer address, meaning that the fileDescriptor parameter of various Dispatch APIs is actually NOT a file descriptor on Windows but rather a HANDLE. This means that the handle should NOT be converted using _open_osfhandle, and the return value of this function should ONLY be passed to Dispatch functions where the fileDescriptor parameter is masquerading as a HANDLE in this manner. Use with extreme caution.
+        rawValue = .init(bitPattern: fileHandle._handle)
+        #else
+        rawValue = fileHandle.fileDescriptor
+        #endif
+    }
+}