Make Pipes and FileDescriptor ~Copyable

Author: iCharlesHu

Sources/Subprocess/API.swift

@@ -110,27 +110,32 @@ public func run<
         error: try error.createPipe()
     ) { execution, inputIO, outputIO, errorIO in
         // Write input, capture output and error in parallel
+        var inputIOBox: TrackedPlatformDiskIO? = consume inputIO
+        var outputIOBox: TrackedPlatformDiskIO? = consume outputIO
+        var errorIOBox: TrackedPlatformDiskIO? = consume errorIO
         return try await withThrowingTaskGroup(
             of: OutputCapturingState<Output.OutputType, Error.OutputType>.self,
             returning: RunResult.self
         ) { group in
+            var outputIOContainer: TrackedPlatformDiskIO? = outputIOBox.take()
+            var errorIOContainer: TrackedPlatformDiskIO? = errorIOBox.take()
             group.addTask {
                 let stdout = try await output.captureOutput(
-                    from: outputIO
+                    from: outputIOContainer.take()
                 )
                 return .standardOutputCaptured(stdout)
             }
             group.addTask {
                 let stderr = try await error.captureOutput(
-                    from: errorIO
+                    from: errorIOContainer.take()
                 )
                 return .standardErrorCaptured(stderr)
             }
 
             // Write span at the same isolation
-            if let writeFd = inputIO {
+            if let writeFd = inputIOBox.take() {
                 let writer = StandardInputWriter(diskIO: writeFd)
-                _ = try await writer.write(input.bytes)
+                _ = try await writer.write(input._bytes)
                 try await writer.finish()
             }
 
@@ -207,20 +212,23 @@ public func run<Result, Input: InputProtocol, Error: OutputProtocol>(
         output: try output.createPipe(),
         error: try error.createPipe()
     ) { execution, inputIO, outputIO, errorIO in
+        var inputIOBox: TrackedPlatformDiskIO? = consume inputIO
+        var outputIOBox: TrackedPlatformDiskIO? = consume outputIO
         return try await withThrowingTaskGroup(
             of: Void.self,
             returning: Result.self
         ) { group in
+            var inputIOContainer: TrackedPlatformDiskIO? = inputIOBox.take()
             group.addTask {
-                if let inputIO = inputIO {
+                if let inputIO = inputIOContainer.take() {
                     let writer = StandardInputWriter(diskIO: inputIO)
                     try await input.write(with: writer)
                     try await writer.finish()
                 }
             }
 
             // Body runs in the same isolation
-            let outputSequence = AsyncBufferSequence(diskIO: outputIO!)
+            let outputSequence = AsyncBufferSequence(diskIO: outputIOBox.take()!.consumeDiskIO())
             let result = try await body(execution, outputSequence)
             try await group.waitForAll()
             return result
@@ -254,20 +262,23 @@ public func run<Result, Input: InputProtocol, Output: OutputProtocol>(
         output: try output.createPipe(),
         error: try error.createPipe()
     ) { execution, inputIO, outputIO, errorIO in
+        var inputIOBox: TrackedPlatformDiskIO? = consume inputIO
+        var errorIOBox: TrackedPlatformDiskIO? = consume errorIO
         return try await withThrowingTaskGroup(
             of: Void.self,
             returning: Result.self
         ) { group in
+            var inputIOContainer: TrackedPlatformDiskIO? = inputIOBox.take()
             group.addTask {
-                if let inputIO = inputIO {
+                if let inputIO = inputIOContainer.take() {
                     let writer = StandardInputWriter(diskIO: inputIO)
                     try await input.write(with: writer)
                     try await writer.finish()
                 }
             }
 
             // Body runs in the same isolation
-            let errorSequence = AsyncBufferSequence(diskIO: errorIO!)
+            let errorSequence = AsyncBufferSequence(diskIO: errorIOBox.take()!.consumeDiskIO())
             let result = try await body(execution, errorSequence)
             try await group.waitForAll()
             return result
@@ -303,7 +314,7 @@ public func run<Result, Error: OutputProtocol>(
         error: try error.createPipe()
     ) { execution, inputIO, outputIO, errorIO in
         let writer = StandardInputWriter(diskIO: inputIO!)
-        let outputSequence = AsyncBufferSequence(diskIO: outputIO!)
+        let outputSequence = AsyncBufferSequence(diskIO: outputIO!.consumeDiskIO())
         return try await body(execution, writer, outputSequence)
     }
 }
@@ -336,7 +347,7 @@ public func run<Result, Output: OutputProtocol>(
         error: try error.createPipe()
     ) { execution, inputIO, outputIO, errorIO in
         let writer = StandardInputWriter(diskIO: inputIO!)
-        let errorSequence = AsyncBufferSequence(diskIO: errorIO!)
+        let errorSequence = AsyncBufferSequence(diskIO: errorIO!.consumeDiskIO())
         return try await body(execution, writer, errorSequence)
     }
 }
@@ -393,8 +404,8 @@ public func run<Result>(
         error: try error.createPipe()
     ) { execution, inputIO, outputIO, errorIO in
         let writer = StandardInputWriter(diskIO: inputIO!)
-        let outputSequence = AsyncBufferSequence(diskIO: outputIO!)
-        let errorSequence = AsyncBufferSequence(diskIO: errorIO!)
+        let outputSequence = AsyncBufferSequence(diskIO: outputIO!.consumeDiskIO())
+        let errorSequence = AsyncBufferSequence(diskIO: errorIO!.consumeDiskIO())
         return try await body(execution, writer, outputSequence, errorSequence)
     }
 }
@@ -433,12 +444,18 @@ public func run<
         error: try error.createPipe()
     ) { (execution, inputIO, outputIO, errorIO) -> RunResult in
         // Write input, capture output and error in parallel
+        var inputIOBox: TrackedPlatformDiskIO? = consume inputIO
+        var outputIOBox: TrackedPlatformDiskIO? = consume outputIO
+        var errorIOBox: TrackedPlatformDiskIO? = consume errorIO
         return try await withThrowingTaskGroup(
             of: OutputCapturingState<Output.OutputType, Error.OutputType>?.self,
             returning: RunResult.self
         ) { group in
+            var inputIOContainer: TrackedPlatformDiskIO? = inputIOBox.take()
+            var outputIOContainer: TrackedPlatformDiskIO? = outputIOBox.take()
+            var errorIOContainer: TrackedPlatformDiskIO? = errorIOBox.take()
             group.addTask {
-                if let writeFd = inputIO {
+                if let writeFd = inputIOContainer.take() {
                     let writer = StandardInputWriter(diskIO: writeFd)
                     try await input.write(with: writer)
                     try await writer.finish()
@@ -447,13 +464,13 @@ public func run<
             }
             group.addTask {
                 let stdout = try await output.captureOutput(
-                    from: outputIO
+                    from: outputIOContainer.take()
                 )
                 return .standardOutputCaptured(stdout)
             }
             group.addTask {
                 let stderr = try await error.captureOutput(
-                    from: errorIO
+                    from: errorIOContainer.take()
                 )
                 return .standardErrorCaptured(stderr)
             }
@@ -514,8 +531,8 @@ public func run<Result>(
         error: try error.createPipe()
     ) { execution, inputIO, outputIO, errorIO in
         let writer = StandardInputWriter(diskIO: inputIO!)
-        let outputSequence = AsyncBufferSequence(diskIO: outputIO!)
-        let errorSequence = AsyncBufferSequence(diskIO: errorIO!)
+        let outputSequence = AsyncBufferSequence(diskIO: outputIO!.consumeDiskIO())
+        let errorSequence = AsyncBufferSequence(diskIO: errorIO!.consumeDiskIO())
         return try await body(execution, writer, outputSequence, errorSequence)
     }
 }

Sources/Subprocess/AsyncBufferSequence.swift

@@ -15,23 +15,31 @@
 @preconcurrency import SystemPackage
 #endif
 
+internal import Dispatch
+
 #if SubprocessSpan
 @available(SubprocessSpan, *)
 #endif
 public struct AsyncBufferSequence: AsyncSequence, Sendable {
     public typealias Failure = any Swift.Error
     public typealias Element = Buffer
 
+    #if os(Windows)
+    internal typealias DiskIO = FileDescriptor
+    #else
+    internal typealias DiskIO = DispatchIO
+    #endif
+
     @_nonSendable
     public struct Iterator: AsyncIteratorProtocol {
         public typealias Element = Buffer
 
-        private let diskIO: TrackedPlatformDiskIO
+        private let diskIO: DiskIO
         private var buffer: [UInt8]
         private var currentPosition: Int
         private var finished: Bool
 
-        internal init(diskIO: TrackedPlatformDiskIO) {
+        internal init(diskIO: DiskIO) {
             self.diskIO = diskIO
             self.buffer = []
             self.currentPosition = 0
@@ -44,16 +52,20 @@ public struct AsyncBufferSequence: AsyncSequence, Sendable {
             )
             if data == nil {
                 // We finished reading. Close the file descriptor now
-                try self.diskIO.safelyClose()
+                #if os(Windows)
+                try self.diskIO.close()
+                #else
+                self.diskIO.close()
+                #endif
                 return nil
             }
             return data
         }
     }
 
-    private let diskIO: TrackedPlatformDiskIO
+    private let diskIO: DiskIO
 
-    internal init(diskIO: TrackedPlatformDiskIO) {
+    internal init(diskIO: DiskIO) {
         self.diskIO = diskIO
     }