Make AsyncIO.read's maxLength a cap and rewrite Darwin backend (#259)

Previously AsyncIO.read(upTo: maxLength) treated maxLength as a target length. It would accumulate bytes in a loop until it had gathered that many before returning. This deadlocks when a subprocess writes fewer bytes than maxLength and then pauses or waits on stdin: the parent blocks forever on data the child never sends.

This commit change maxLength to a "cap", as its name suggests. Each read(upTo:) now returns as soon as the underlying nonblocking read(2) (or ReadFile) completes. Callers that need a specific byte count accumulate across calls themselves.

Follow-on cleanup enabled by the semantics fix:

- Replace Darwin's DispatchIO backend with kqueue + read(2).
    The new Darwin path is a thin event-source shim over the same nonblocking read/write loop that Linux and Android already use, sharing implementation in a new IO/AsyncIO+Unix.swift.
- Collapse IOChannel into IODescriptor.
    IOChannel only existed to wrap a DispatchIO; with DispatchIO gone, a single fd-owning type suffices.
- Drop the public `preferredBufferSize` parameter from every run()
overload.
    It was introduced to let callers work around the old hang; buffer size is now derived from the pipe's capacity via F_GETPIPE_SZ (Linux/Android), fstat().st_blksize (Darwin/OpenBSD), or a fixed 64 KB fallback (FreeBSD, Windows).

Author: iCharlesHu

Package.swift

@@ -22,7 +22,7 @@ dep.append(
 let defaultTraits: Set<String> = ["SubprocessFoundation"]
 
 let packageSwiftSettings: [SwiftSetting] = [
-    .define("SUBPROCESS_ASYNCIO_DISPATCH", .when(platforms: [.macOS, .custom("freebsd"), .openbsd])),
+    .define("SUBPROCESS_ASYNCIO_KQUEUE", .when(platforms: [.macOS, .custom("freebsd"), .openbsd])),
     .enableUpcomingFeature("ExistentialAny"),
     .enableUpcomingFeature("MemberImportVisibility"),
     .enableUpcomingFeature("InternalImportsByDefault"),

Sources/Subprocess/API.swift

@@ -38,9 +38,7 @@ public struct AsyncBufferSequence: AsyncSequence, @unchecked Sendable {
     /// The element type for the asynchronous sequence.
     public typealias Element = Buffer
 
-    #if SUBPROCESS_ASYNCIO_DISPATCH
-    internal typealias DiskIO = DispatchIO
-    #elseif canImport(WinSDK)
+    #if canImport(WinSDK)
     internal typealias DiskIO = HANDLE
     #else
     internal typealias DiskIO = FileDescriptor
@@ -55,10 +53,11 @@ public struct AsyncBufferSequence: AsyncSequence, @unchecked Sendable {
         private let preferredBufferSize: Int
         private var buffer: [Buffer]
 
-        internal init(diskIO: DiskIO, preferredBufferSize: Int?) {
+        internal init(diskIO: DiskIO) {
             self.diskIO = diskIO
             self.buffer = []
-            self.preferredBufferSize = preferredBufferSize ?? readBufferSize
+            // Only need to query it once at beginning of stream
+            self.preferredBufferSize = AsyncIO.queryPipeBufferSize(for: diskIO)
         }
 
         /// Retrieves the next buffer in the sequence, or `nil` if the sequence ended.
@@ -74,24 +73,14 @@ public struct AsyncBufferSequence: AsyncSequence, @unchecked Sendable {
             )
             guard let data else {
                 // We finished reading. Close the file descriptor now
-                #if SUBPROCESS_ASYNCIO_DISPATCH
-                try _safelyClose(.dispatchIO(self.diskIO))
-                #elseif canImport(WinSDK)
+                #if canImport(WinSDK)
                 try _safelyClose(.handle(self.diskIO))
                 #else
                 try _safelyClose(.fileDescriptor(self.diskIO))
                 #endif
                 return nil
             }
-            let createdBuffers = Buffer.createFrom(data)
-            // Most (all?) cases there should be only one buffer
-            // because DispatchData are mostly contiguous
-            if _fastPath(createdBuffers.count == 1) {
-                // No need to push to the stack
-                return createdBuffers[0]
-            }
-            self.buffer = createdBuffers
-            return self.buffer.removeFirst()
+            return Buffer(data: data)
         }
 
         /// Retrieves the next buffer in the sequence, or `nil` if the sequence ended.
@@ -101,19 +90,14 @@ public struct AsyncBufferSequence: AsyncSequence, @unchecked Sendable {
     }
 
     private let diskIO: DiskIO
-    private let preferredBufferSize: Int?
 
-    internal init(diskIO: DiskIO, preferredBufferSize: Int?) {
+    internal init(diskIO: DiskIO) {
         self.diskIO = diskIO
-        self.preferredBufferSize = preferredBufferSize
     }
 
     /// Creates an iterator for this asynchronous sequence.
     public func makeAsyncIterator() -> Iterator {
-        return Iterator(
-            diskIO: self.diskIO,
-            preferredBufferSize: self.preferredBufferSize
-        )
+        return Iterator(diskIO: self.diskIO)
     }
 
     /// Splits the buffer into strings using the specified separator.
@@ -268,29 +252,12 @@ extension AsyncBufferSequence {
                         self.eofReached = true
                         return nil
                     }
-                    #if SUBPROCESS_ASYNCIO_DISPATCH
-                    // Unfortunately here we _have to_ copy the bytes out because
-                    // DispatchIO (rightfully) reuses buffer, which means `buffer.data`
-                    // has the same address on all iterations, therefore we can't directly
-                    // create the result array from buffer.data
-
-                    // Calculate how many CodePoint elements we have
-                    let elementCount = buffer.data.count / MemoryLayout<Encoding.CodeUnit>.stride
-
-                    // Create array by copying from the buffer reinterpreted as CodePoint
-                    let result: Array<Encoding.CodeUnit> = buffer.data.withUnsafeBytes { ptr -> Array<Encoding.CodeUnit> in
-                        return Array(
-                            UnsafeBufferPointer(start: ptr.baseAddress?.assumingMemoryBound(to: Encoding.CodeUnit.self), count: elementCount)
-                        )
-                    }
-                    #else
                     // Cast data to CodeUnit type
                     let result = buffer.withUnsafeBytes { ptr in
                         return ptr.withMemoryRebound(to: Encoding.CodeUnit.self) { codeUnitPtr in
                             return Array(codeUnitPtr)
                         }
                     }
-                    #endif
                     return result.isEmpty ? nil : result
                 }
 
@@ -604,6 +571,6 @@ private let _pageSize: Int = Int(getpagesize())
 #endif // canImport(Darwin)
 
 @inline(__always)
-internal var readBufferSize: Int {
+internal var systemPageSize: Int {
     return _pageSize
 }

Sources/Subprocess/AsyncBufferSequence.swift

@@ -9,11 +9,6 @@
 //
 //===----------------------------------------------------------------------===//
 
-// swift-format-ignore-file
-
-#if canImport(Darwin) || canImport(Glibc) || canImport(Android) || canImport(Musl)
-@preconcurrency internal import Dispatch
-
 #if SubprocessFoundation
 
 #if canImport(Darwin)
@@ -24,41 +19,16 @@ internal import Foundation
 internal import FoundationEssentials
 #endif
 
-#endif
 #endif
 
 extension AsyncBufferSequence {
     /// An immutable collection of bytes.
     public struct Buffer: Sendable {
-        #if SUBPROCESS_ASYNCIO_DISPATCH
-        // We need to keep the backingData alive while Slice is alive
-        internal let backingData: DispatchData
-        internal let data: DispatchData.Region
-
-        internal init(data: DispatchData.Region, backingData: DispatchData) {
-            self.data = data
-            self.backingData = backingData
-        }
-
-        internal static func createFrom(_ data: DispatchData) -> [Buffer] {
-            let slices = data.regions
-            // In most (all?) cases data should only have one slice
-            if _fastPath(slices.count == 1) {
-                return [.init(data: slices[0], backingData: data)]
-            }
-            return slices.map { .init(data: $0, backingData: data) }
-        }
-        #else
         internal let data: [UInt8]
 
         internal init(data: [UInt8]) {
             self.data = data
         }
-
-        internal static func createFrom(_ data: [UInt8]) -> [Buffer] {
-            return [.init(data: data)]
-        }
-        #endif // SUBPROCESS_ASYNCIO_DISPATCH
     }
 }
 
@@ -104,78 +74,3 @@ extension AsyncBufferSequence.Buffer {
         }
     }
 }
-
-// MARK: - Hashable, Equatable
-extension AsyncBufferSequence.Buffer: Equatable, Hashable {
-    #if SUBPROCESS_ASYNCIO_DISPATCH
-    /// Returns a Boolean value that indicates whether two buffers are equal.
-    public static func == (lhs: AsyncBufferSequence.Buffer, rhs: AsyncBufferSequence.Buffer) -> Bool {
-        return lhs.data == rhs.data
-    }
-
-    /// Hashes the essential components of this value by feeding them into the given hasher.
-    public func hash(into hasher: inout Hasher) {
-        return self.data.hash(into: &hasher)
-    }
-    #endif
-    // else Compiler generated conformances
-}
-
-#if SUBPROCESS_ASYNCIO_DISPATCH
-extension DispatchData.Region {
-    static func == (lhs: DispatchData.Region, rhs: DispatchData.Region) -> Bool {
-        return lhs.withUnsafeBytes { lhsBytes in
-            return rhs.withUnsafeBytes { rhsBytes in
-                return lhsBytes.elementsEqual(rhsBytes)
-            }
-        }
-    }
-
-    internal func hash(into hasher: inout Hasher) {
-        return self.withUnsafeBytes { ptr in
-            return hasher.combine(bytes: ptr)
-        }
-    }
-}
-#if !canImport(Darwin) || !SubprocessFoundation
-/// `DispatchData.Region` is defined in Foundation, but we can't depend on Foundation when the SubprocessFoundation trait is disabled.
-extension DispatchData {
-    typealias Region = _ContiguousBufferView
-
-    var regions: [Region] {
-        contiguousBufferViews
-    }
-
-    internal struct _ContiguousBufferView: @unchecked Sendable, RandomAccessCollection {
-        typealias Element = UInt8
-
-        internal let bytes: UnsafeBufferPointer<UInt8>
-
-        internal var startIndex: Int { self.bytes.startIndex }
-        internal var endIndex: Int { self.bytes.endIndex }
-
-        internal init(bytes: UnsafeBufferPointer<UInt8>) {
-            self.bytes = bytes
-        }
-
-        internal func withUnsafeBytes<ResultType>(_ body: (UnsafeRawBufferPointer) throws -> ResultType) rethrows -> ResultType {
-            return try body(UnsafeRawBufferPointer(self.bytes))
-        }
-
-        subscript(position: Int) -> UInt8 {
-            _read {
-                yield self.bytes[position]
-            }
-        }
-    }
-
-    internal var contiguousBufferViews: [_ContiguousBufferView] {
-        var slices = [_ContiguousBufferView]()
-        enumerateBytes { (bytes, index, stop) in
-            slices.append(_ContiguousBufferView(bytes: bytes))
-        }
-        return slices
-    }
-}
-#endif
-#endif

Sources/Subprocess/Buffer.swift

@@ -18,9 +18,10 @@ add_library(Subprocess
   Result.swift
   IO/Output.swift
   IO/Input.swift
-  IO/AsyncIO+Dispatch.swift
+  IO/AsyncIO+KQueue.swift
   IO/AsyncIO+Linux.swift
   IO/AsyncIO+Windows.swift
+  IO/AsyncIO+Unix.swift
   Span+Subprocess.swift
   AsyncBufferSequence.swift
   API.swift
@@ -41,13 +42,13 @@ elseif(APPLE)
     Platforms/Subprocess+Darwin.swift
     Platforms/Subprocess+Unix.swift)
   target_compile_options(Subprocess PRIVATE
-    "$<$<COMPILE_LANGUAGE:Swift>:-DSUBPROCESS_ASYNCIO_DISPATCH>")
+    "$<$<COMPILE_LANGUAGE:Swift>:-DSUBPROCESS_ASYNCIO_KQUEUE>")
 elseif(FREEBSD OR OPENBSD)
   target_sources(Subprocess PRIVATE
     Platforms/Subprocess+BSD.swift
     Platforms/Subprocess+Unix.swift)
   target_compile_options(Subprocess PRIVATE
-    "$<$<COMPILE_LANGUAGE:Swift>:-DSUBPROCESS_ASYNCIO_DISPATCH>")
+    "$<$<COMPILE_LANGUAGE:Swift>:-DSUBPROCESS_ASYNCIO_KQUEUE>")
 endif()
 
 target_compile_options(Subprocess PRIVATE