Skip to content

Introduce AsyncIO for Windows and Linux #117

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 10 commits into from
Jul 29, 2025
Merged

Introduce AsyncIO for Windows and Linux #117

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
b6181ea
Create platform specific AsyncIO
iCharlesHu May 17, 2025
e374669
Refactor captureOutput() to minimize force unwrap
iCharlesHu Jul 3, 2025
61508ad
Create platform specific AsyncIO
iCharlesHu May 17, 2025
a72d35f
Fix fd was not closed error on Windows
iCharlesHu Jun 23, 2025
74995e4
Fix Windows test errors
iCharlesHu Jun 24, 2025
44a9e97
Introduce Windows IOCP based AsyncIO implementation
iCharlesHu Jun 25, 2025
22e4b38
Move platform specific AsyncIO implementations to separate files
iCharlesHu Jul 23, 2025
ba8c6c8
Use _beginthreadex instead of CreatThread on Windows for AsyncIO
iCharlesHu Jul 24, 2025
9c6703f
Remove DispatchData._ContiguousBufferView since Linux no longer relie…
iCharlesHu Jul 25, 2025
6f030af
Linux: reap child process if fork succeed but exec fails
iCharlesHu Jul 28, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion Package.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,9 @@ import PackageDescription
var dep: [Package.Dependency] = [
.package(
url: "https://github.com/apple/swift-system",
from: "1.5.0"
// Temporarily pin to 1.5.0 because 1.6.0 has a breaking change for Ubuntu Focal
// https://github.com/apple/swift-system/issues/237
exact: "1.5.0"
)
]
#if !os(Windows)
Expand Down
50 changes: 25 additions & 25 deletions Sources/Subprocess/API.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,9 +105,9 @@ public func run<
output: try output.createPipe(),
error: try error.createPipe()
) { execution, inputIO, outputIO, errorIO in
var inputIOBox: TrackedPlatformDiskIO? = consume inputIO
var outputIOBox: TrackedPlatformDiskIO? = consume outputIO
var errorIOBox: TrackedPlatformDiskIO? = consume errorIO
var inputIOBox: IOChannel? = consume inputIO
var outputIOBox: IOChannel? = consume outputIO
var errorIOBox: IOChannel? = consume errorIO

// Write input, capture output and error in parallel
async let stdout = try output.captureOutput(from: outputIOBox.take())
Expand Down Expand Up @@ -177,12 +177,12 @@ public func run<Result, Input: InputProtocol, Output: OutputProtocol, Error: Out
output: try output.createPipe(),
error: try error.createPipe()
) { execution, inputIO, outputIO, errorIO in
var inputIOBox: TrackedPlatformDiskIO? = consume inputIO
var inputIOBox: IOChannel? = consume inputIO
return try await withThrowingTaskGroup(
of: Void.self,
returning: Result.self
) { group in
var inputIOContainer: TrackedPlatformDiskIO? = inputIOBox.take()
var inputIOContainer: IOChannel? = inputIOBox.take()
group.addTask {
if let inputIO = inputIOContainer.take() {
let writer = StandardInputWriter(diskIO: inputIO)
Expand Down Expand Up @@ -237,13 +237,13 @@ 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
var inputIOBox: IOChannel? = consume inputIO
var outputIOBox: IOChannel? = consume outputIO
return try await withThrowingTaskGroup(
of: Void.self,
returning: Result.self
) { group in
var inputIOContainer: TrackedPlatformDiskIO? = inputIOBox.take()
var inputIOContainer: IOChannel? = inputIOBox.take()
group.addTask {
if let inputIO = inputIOContainer.take() {
let writer = StandardInputWriter(diskIO: inputIO)
Expand All @@ -253,7 +253,7 @@ public func run<Result, Input: InputProtocol, Error: OutputProtocol>(
}

// Body runs in the same isolation
let outputSequence = AsyncBufferSequence(diskIO: outputIOBox.take()!.consumeDiskIO())
let outputSequence = AsyncBufferSequence(diskIO: outputIOBox.take()!.consumeIOChannel())
let result = try await body(execution, outputSequence)
try await group.waitForAll()
return result
Expand Down Expand Up @@ -299,13 +299,13 @@ 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
var inputIOBox: IOChannel? = consume inputIO
var errorIOBox: IOChannel? = consume errorIO
return try await withThrowingTaskGroup(
of: Void.self,
returning: Result.self
) { group in
var inputIOContainer: TrackedPlatformDiskIO? = inputIOBox.take()
var inputIOContainer: IOChannel? = inputIOBox.take()
group.addTask {
if let inputIO = inputIOContainer.take() {
let writer = StandardInputWriter(diskIO: inputIO)
Expand All @@ -315,7 +315,7 @@ public func run<Result, Input: InputProtocol, Output: OutputProtocol>(
}

// Body runs in the same isolation
let errorSequence = AsyncBufferSequence(diskIO: errorIOBox.take()!.consumeDiskIO())
let errorSequence = AsyncBufferSequence(diskIO: errorIOBox.take()!.consumeIOChannel())
let result = try await body(execution, errorSequence)
try await group.waitForAll()
return result
Expand Down Expand Up @@ -363,7 +363,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!.consumeDiskIO())
let outputSequence = AsyncBufferSequence(diskIO: outputIO!.consumeIOChannel())
return try await body(execution, writer, outputSequence)
}
}
Expand Down Expand Up @@ -408,7 +408,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!.consumeDiskIO())
let errorSequence = AsyncBufferSequence(diskIO: errorIO!.consumeIOChannel())
return try await body(execution, writer, errorSequence)
}
}
Expand Down Expand Up @@ -460,8 +460,8 @@ public func run<Result>(
error: try error.createPipe()
) { execution, inputIO, outputIO, errorIO in
let writer = StandardInputWriter(diskIO: inputIO!)
let outputSequence = AsyncBufferSequence(diskIO: outputIO!.consumeDiskIO())
let errorSequence = AsyncBufferSequence(diskIO: errorIO!.consumeDiskIO())
let outputSequence = AsyncBufferSequence(diskIO: outputIO!.consumeIOChannel())
let errorSequence = AsyncBufferSequence(diskIO: errorIO!.consumeIOChannel())
return try await body(execution, writer, outputSequence, errorSequence)
}
}
Expand Down Expand Up @@ -497,16 +497,16 @@ 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
var inputIOBox: IOChannel? = consume inputIO
var outputIOBox: IOChannel? = consume outputIO
var errorIOBox: IOChannel? = 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()
var inputIOContainer: IOChannel? = inputIOBox.take()
var outputIOContainer: IOChannel? = outputIOBox.take()
var errorIOContainer: IOChannel? = errorIOBox.take()
group.addTask {
if let writeFd = inputIOContainer.take() {
let writer = StandardInputWriter(diskIO: writeFd)
Expand Down Expand Up @@ -580,8 +580,8 @@ public func run<Result>(
error: try error.createPipe()
) { execution, inputIO, outputIO, errorIO in
let writer = StandardInputWriter(diskIO: inputIO!)
let outputSequence = AsyncBufferSequence(diskIO: outputIO!.consumeDiskIO())
let errorSequence = AsyncBufferSequence(diskIO: errorIO!.consumeDiskIO())
let outputSequence = AsyncBufferSequence(diskIO: outputIO!.consumeIOChannel())
let errorSequence = AsyncBufferSequence(diskIO: errorIO!.consumeIOChannel())
return try await body(execution, writer, outputSequence, errorSequence)
}
}
Expand Down
45 changes: 23 additions & 22 deletions Sources/Subprocess/AsyncBufferSequence.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,14 +19,16 @@
internal import Dispatch
#endif

public struct AsyncBufferSequence: AsyncSequence, Sendable {
public struct AsyncBufferSequence: AsyncSequence, @unchecked Sendable {
public typealias Failure = any Swift.Error
public typealias Element = Buffer

#if os(Windows)
internal typealias DiskIO = FileDescriptor
#else
#if canImport(Darwin)
internal typealias DiskIO = DispatchIO
#elseif canImport(WinSDK)
internal typealias DiskIO = HANDLE
#else
internal typealias DiskIO = FileDescriptor
#endif

@_nonSendable
Expand All @@ -47,15 +49,18 @@ public struct AsyncBufferSequence: AsyncSequence, Sendable {
return self.buffer.removeFirst()
}
// Read more data
let data = try await self.diskIO.read(
upToLength: readBufferSize
let data = try await AsyncIO.shared.read(
from: self.diskIO,
upTo: readBufferSize
)
guard let data else {
// We finished reading. Close the file descriptor now
#if os(Windows)
try self.diskIO.close()
#if canImport(Darwin)
try _safelyClose(.dispatchIO(self.diskIO))
#elseif canImport(WinSDK)
try _safelyClose(.handle(self.diskIO))
#else
self.diskIO.close()
try _safelyClose(.fileDescriptor(self.diskIO))
#endif
return nil
}
Expand Down Expand Up @@ -132,17 +137,7 @@ extension AsyncBufferSequence {
self.eofReached = true
return nil
}
#if os(Windows)
// Cast data to CodeUnit type
let result = buffer.withUnsafeBytes { ptr in
return Array(
UnsafeBufferPointer<Encoding.CodeUnit>(
start: ptr.bindMemory(to: Encoding.CodeUnit.self).baseAddress!,
count: ptr.count / MemoryLayout<Encoding.CodeUnit>.size
)
)
}
#else
#if canImport(Darwin)
// 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
Expand All @@ -157,7 +152,13 @@ extension AsyncBufferSequence {
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
}
Expand Down Expand Up @@ -340,7 +341,7 @@ private let _pageSize: Int = {
Int(_subprocess_vm_size())
}()
#elseif canImport(WinSDK)
import WinSDK
@preconcurrency import WinSDK
private let _pageSize: Int = {
var sysInfo: SYSTEM_INFO = SYSTEM_INFO()
GetSystemInfo(&sysInfo)
Expand Down
82 changes: 30 additions & 52 deletions Sources/Subprocess/Buffer.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,35 +17,35 @@
extension AsyncBufferSequence {
/// A immutable collection of bytes
public struct Buffer: Sendable {
#if os(Windows)
internal let data: [UInt8]

internal init(data: [UInt8]) {
self.data = data
}

internal static func createFrom(_ data: [UInt8]) -> [Buffer] {
return [.init(data: data)]
}
#else
// We need to keep the backingData alive while _ContiguousBufferView is alive
#if canImport(Darwin)
// We need to keep the backingData alive while Slice is alive
internal let backingData: DispatchData
internal let data: DispatchData._ContiguousBufferView
internal let data: DispatchData.Region

internal init(data: DispatchData._ContiguousBufferView, backingData: DispatchData) {
internal init(data: DispatchData.Region, backingData: DispatchData) {
self.data = data
self.backingData = backingData
}

internal static func createFrom(_ data: DispatchData) -> [Buffer] {
let slices = data.contiguousBufferViews
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) }
}
#endif
#else
internal let data: [UInt8]

internal init(data: [UInt8]) {
self.data = data
}

internal static func createFrom(_ data: [UInt8]) -> [Buffer] {
return [.init(data: data)]
}
#endif // canImport(Darwin)
}
}

Expand Down Expand Up @@ -92,55 +92,33 @@ extension AsyncBufferSequence.Buffer {

// MARK: - Hashable, Equatable
extension AsyncBufferSequence.Buffer: Equatable, Hashable {
#if os(Windows)
// Compiler generated conformances
#else
#if canImport(Darwin)
public static func == (lhs: AsyncBufferSequence.Buffer, rhs: AsyncBufferSequence.Buffer) -> Bool {
return lhs.data.elementsEqual(rhs.data)
return lhs.data == rhs.data
}

public func hash(into hasher: inout Hasher) {
self.data.withUnsafeBytes { ptr in
hasher.combine(bytes: ptr)
}
return self.data.hash(into: &hasher)
}
#endif
// else Compiler generated conformances
}

// MARK: - DispatchData.Block
#if canImport(Darwin) || canImport(Glibc) || canImport(Android) || canImport(Musl)
extension DispatchData {
/// Unfortunately `DispatchData.Region` is not available on Linux, hence our own wrapper
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]
#if canImport(Darwin)
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 var contiguousBufferViews: [_ContiguousBufferView] {
var slices = [_ContiguousBufferView]()
enumerateBytes { (bytes, index, stop) in
slices.append(_ContiguousBufferView(bytes: bytes))
internal func hash(into hasher: inout Hasher) {
return self.withUnsafeBytes { ptr in
return hasher.combine(bytes: ptr)
}
return slices
}
}

#endif

3 changes: 3 additions & 0 deletions Sources/Subprocess/CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,9 @@ add_library(Subprocess
Result.swift
IO/Output.swift
IO/Input.swift
IO/AsyncIO+Darwin.swift
IO/AsyncIO+Linux.swift
IO/AsyncIO+Windows.swift
Span+Subprocess.swift
AsyncBufferSequence.swift
API.swift
Expand Down
Loading