Skip to content

Add StreamingBehavior to AsyncBufferSequence for throughput/latency control #228

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

Closed
dannys42 wants to merge 3 commits into from
Closed

Add StreamingBehavior to AsyncBufferSequence for throughput/latency control #228

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e61c03b
FEATURE-202602: Add new feature to support different streaming behaviors
dannys42 Mar 20, 2026
c67c36f
FEATURE-202602: simplify PendingReadState
dannys42 Mar 20, 2026
faa4381
FEATURE-202602: Add unit test to ensure we get streaming data as it c…
dannys42 Mar 20, 2026
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
53 changes: 47 additions & 6 deletions Sources/Subprocess/API.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -166,6 +166,8 @@ public func run<
/// as the default buffer size. Larger buffer sizes may improve performance for
/// subprocesses that produce large amounts of output, while smaller buffer sizes
/// may reduce memory usage and improve responsiveness for interactive applications.
/// - streamingBehavior: Controls the trade-off between throughput and latency when
/// streaming output from the subprocess. Defaults to `.throughput`.
/// - isolation: the isolation context to run the body closure.
/// - body: The custom execution body to manually control the running process.
/// - Returns: an `ExecutableResult` type containing the return value of the closure.
Expand All @@ -178,6 +180,7 @@ public func run<Result, Input: InputProtocol, Error: ErrorOutputProtocol>(
input: Input = .none,
error: Error = .discarded,
preferredBufferSize: Int? = nil,
streamingBehavior: AsyncBufferSequence.StreamingBehavior = .throughput,
isolation: isolated (any Actor)? = #isolation,
body: ((Execution, AsyncBufferSequence) async throws -> Result)
) async throws -> ExecutionOutcome<Result> where Error.OutputType == Void {
Expand All @@ -193,6 +196,7 @@ public func run<Result, Input: InputProtocol, Error: ErrorOutputProtocol>(
input: input,
error: error,
preferredBufferSize: preferredBufferSize,
streamingBehavior: streamingBehavior,
isolation: isolation,
body: body
)
Expand All @@ -213,6 +217,8 @@ public func run<Result, Input: InputProtocol, Error: ErrorOutputProtocol>(
/// as the default buffer size. Larger buffer sizes may improve performance for
/// subprocesses that produce large amounts of output, while smaller buffer sizes
/// may reduce memory usage and improve responsiveness for interactive applications.
/// - streamingBehavior: Controls the trade-off between throughput and latency when
/// streaming output from the subprocess. Defaults to `.throughput`.
/// - isolation: the isolation context to run the body closure.
/// - body: The custom execution body to manually control the running process
/// - Returns: an `ExecutableResult` type containing the return value of the closure.
Expand All @@ -225,6 +231,7 @@ public func run<Result, Input: InputProtocol, Output: OutputProtocol>(
input: Input = .none,
output: Output,
preferredBufferSize: Int? = nil,
streamingBehavior: AsyncBufferSequence.StreamingBehavior = .throughput,
isolation: isolated (any Actor)? = #isolation,
body: ((Execution, AsyncBufferSequence) async throws -> Result)
) async throws -> ExecutionOutcome<Result> where Output.OutputType == Void {
Expand All @@ -240,6 +247,7 @@ public func run<Result, Input: InputProtocol, Output: OutputProtocol>(
input: input,
output: output,
preferredBufferSize: preferredBufferSize,
streamingBehavior: streamingBehavior,
isolation: isolation,
body: body
)
Expand All @@ -259,6 +267,8 @@ public func run<Result, Input: InputProtocol, Output: OutputProtocol>(
/// as the default buffer size. Larger buffer sizes may improve performance for
/// subprocesses that produce large amounts of output, while smaller buffer sizes
/// may reduce memory usage and improve responsiveness for interactive applications.
/// - streamingBehavior: Controls the trade-off between throughput and latency when
/// streaming output from the subprocess. Defaults to `.throughput`.
/// - isolation: the isolation context to run the body closure.
/// - body: The custom execution body to manually control the running process
/// - Returns: An `ExecutableResult` type containing the return value of the closure.
Expand All @@ -270,6 +280,7 @@ public func run<Result, Error: ErrorOutputProtocol>(
platformOptions: PlatformOptions = PlatformOptions(),
error: Error = .discarded,
preferredBufferSize: Int? = nil,
streamingBehavior: AsyncBufferSequence.StreamingBehavior = .throughput,
isolation: isolated (any Actor)? = #isolation,
body: ((Execution, StandardInputWriter, AsyncBufferSequence) async throws -> Result)
) async throws -> ExecutionOutcome<Result> where Error.OutputType == Void {
Expand All @@ -284,6 +295,7 @@ public func run<Result, Error: ErrorOutputProtocol>(
configuration,
error: error,
preferredBufferSize: preferredBufferSize,
streamingBehavior: streamingBehavior,
isolation: isolation,
body: body
)
Expand All @@ -303,6 +315,8 @@ public func run<Result, Error: ErrorOutputProtocol>(
/// as the default buffer size. Larger buffer sizes may improve performance for
/// subprocesses that produce large amounts of output, while smaller buffer sizes
/// may reduce memory usage and improve responsiveness for interactive applications.
/// - streamingBehavior: Controls the trade-off between throughput and latency when
/// streaming output from the subprocess. Defaults to `.throughput`.
/// - isolation: the isolation context to run the body closure.
/// - body: The custom execution body to manually control the running process
/// - Returns: An `ExecutableResult` type containing the return value of the closure.
Expand All @@ -314,6 +328,7 @@ public func run<Result, Output: OutputProtocol>(
platformOptions: PlatformOptions = PlatformOptions(),
output: Output,
preferredBufferSize: Int? = nil,
streamingBehavior: AsyncBufferSequence.StreamingBehavior = .throughput,
isolation: isolated (any Actor)? = #isolation,
body: ((Execution, StandardInputWriter, AsyncBufferSequence) async throws -> Result)
) async throws -> ExecutionOutcome<Result> where Output.OutputType == Void {
Expand All @@ -328,6 +343,7 @@ public func run<Result, Output: OutputProtocol>(
configuration,
output: output,
preferredBufferSize: preferredBufferSize,
streamingBehavior: streamingBehavior,
isolation: isolation,
body: body
)
Expand All @@ -347,6 +363,8 @@ public func run<Result, Output: OutputProtocol>(
/// as the default buffer size. Larger buffer sizes may improve performance for
/// subprocesses that produce large amounts of output, while smaller buffer sizes
/// may reduce memory usage and improve responsiveness for interactive applications.
/// - streamingBehavior: Controls the trade-off between throughput and latency when
/// streaming output from the subprocess. Defaults to `.throughput`.
/// - isolation: the isolation context to run the body closure.
/// - body: The custom execution body to manually control the running process
/// - Returns: an `ExecutableResult` type containing the return value of the closure.
Expand All @@ -357,6 +375,7 @@ public func run<Result>(
workingDirectory: FilePath? = nil,
platformOptions: PlatformOptions = PlatformOptions(),
preferredBufferSize: Int? = nil,
streamingBehavior: AsyncBufferSequence.StreamingBehavior = .throughput,
isolation: isolated (any Actor)? = #isolation,
body: (
(
Expand All @@ -377,6 +396,7 @@ public func run<Result>(
return try await run(
configuration,
preferredBufferSize: preferredBufferSize,
streamingBehavior: streamingBehavior,
isolation: isolation,
body: body
)
Expand Down Expand Up @@ -615,6 +635,8 @@ public func run<
/// as the default buffer size. Larger buffer sizes may improve performance for
/// subprocesses that produce large amounts of output, while smaller buffer sizes
/// may reduce memory usage and improve responsiveness for interactive applications.
/// - streamingBehavior: Controls the trade-off between throughput and latency when
/// streaming output from the subprocess. Defaults to `.throughput`.
/// - isolation: the isolation context to run the body closure.
/// - body: The custom execution body to manually control the running process
/// - Returns an executableResult type containing the return value
Expand All @@ -628,6 +650,7 @@ public func run<
input: Input = .none,
error: Error = .discarded,
preferredBufferSize: Int? = nil,
streamingBehavior: AsyncBufferSequence.StreamingBehavior = .throughput,
isolation: isolated (any Actor)? = #isolation,
body: ((Execution, AsyncBufferSequence) async throws -> Result)
) async throws -> ExecutionOutcome<Result> where Error.OutputType == Void {
Expand Down Expand Up @@ -660,7 +683,8 @@ public func run<
// Body runs in the same isolation
let outputSequence = AsyncBufferSequence(
diskIO: outputIOBox.take()!.consumeIOChannel(),
preferredBufferSize: preferredBufferSize
preferredBufferSize: preferredBufferSize,
streamingBehavior: streamingBehavior
)

let result = try await body(execution, outputSequence)
Expand All @@ -681,6 +705,8 @@ public func run<
/// as the default buffer size. Larger buffer sizes may improve performance for
/// subprocesses that produce large amounts of output, while smaller buffer sizes
/// may reduce memory usage and improve responsiveness for interactive applications.
/// - streamingBehavior: Controls the trade-off between throughput and latency when
/// streaming output from the subprocess. Defaults to `.throughput`.
/// - isolation: the isolation context to run the body closure.
/// - body: The custom execution body to manually control the running process
/// - Returns an executableResult type containing the return value
Expand All @@ -690,6 +716,7 @@ public func run<Result, Input: InputProtocol, Output: OutputProtocol>(
input: Input = .none,
output: Output,
preferredBufferSize: Int? = nil,
streamingBehavior: AsyncBufferSequence.StreamingBehavior = .throughput,
isolation: isolated (any Actor)? = #isolation,
body: ((Execution, AsyncBufferSequence) async throws -> Result)
) async throws -> ExecutionOutcome<Result> where Output.OutputType == Void {
Expand Down Expand Up @@ -719,7 +746,8 @@ public func run<Result, Input: InputProtocol, Output: OutputProtocol>(
// Body runs in the same isolation
let errorSequence = AsyncBufferSequence(
diskIO: errorIOBox.take()!.consumeIOChannel(),
preferredBufferSize: preferredBufferSize
preferredBufferSize: preferredBufferSize,
streamingBehavior: streamingBehavior
)

let result = try await body(execution, errorSequence)
Expand All @@ -740,6 +768,8 @@ public func run<Result, Input: InputProtocol, Output: OutputProtocol>(
/// as the default buffer size. Larger buffer sizes may improve performance for
/// subprocesses that produce large amounts of output, while smaller buffer sizes
/// may reduce memory usage and improve responsiveness for interactive applications.
/// - streamingBehavior: Controls the trade-off between throughput and latency when
/// streaming output from the subprocess. Defaults to `.throughput`.
/// - isolation: the isolation context to run the body closure.
/// - body: The custom execution body to manually control the running process
/// - Returns an executableResult type containing the return value
Expand All @@ -748,6 +778,7 @@ public func run<Result, Error: ErrorOutputProtocol>(
_ configuration: Configuration,
error: Error = .discarded,
preferredBufferSize: Int? = nil,
streamingBehavior: AsyncBufferSequence.StreamingBehavior = .throughput,
isolation: isolated (any Actor)? = #isolation,
body: ((Execution, StandardInputWriter, AsyncBufferSequence) async throws -> Result)
) async throws -> ExecutionOutcome<Result> where Error.OutputType == Void {
Expand All @@ -765,7 +796,8 @@ public func run<Result, Error: ErrorOutputProtocol>(
let writer = StandardInputWriter(diskIO: inputIO!)
let outputSequence = AsyncBufferSequence(
diskIO: outputIO!.consumeIOChannel(),
preferredBufferSize: preferredBufferSize
preferredBufferSize: preferredBufferSize,
streamingBehavior: streamingBehavior
)

return try await body(execution, writer, outputSequence)
Expand All @@ -783,6 +815,8 @@ public func run<Result, Error: ErrorOutputProtocol>(
/// as the default buffer size. Larger buffer sizes may improve performance for
/// subprocesses that produce large amounts of output, while smaller buffer sizes
/// may reduce memory usage and improve responsiveness for interactive applications.
/// - streamingBehavior: Controls the trade-off between throughput and latency when
/// streaming output from the subprocess. Defaults to `.throughput`.
/// - isolation: the isolation context to run the body closure.
/// - body: The custom execution body to manually control the running process
/// - Returns an executableResult type containing the return value
Expand All @@ -791,6 +825,7 @@ public func run<Result, Output: OutputProtocol>(
_ configuration: Configuration,
output: Output,
preferredBufferSize: Int? = nil,
streamingBehavior: AsyncBufferSequence.StreamingBehavior = .throughput,
isolation: isolated (any Actor)? = #isolation,
body: ((Execution, StandardInputWriter, AsyncBufferSequence) async throws -> Result)
) async throws -> ExecutionOutcome<Result> where Output.OutputType == Void {
Expand All @@ -805,7 +840,8 @@ public func run<Result, Output: OutputProtocol>(
let writer = StandardInputWriter(diskIO: inputIO!)
let errorSequence = AsyncBufferSequence(
diskIO: errorIO!.consumeIOChannel(),
preferredBufferSize: preferredBufferSize
preferredBufferSize: preferredBufferSize,
streamingBehavior: streamingBehavior
)
return try await body(execution, writer, errorSequence)
}
Expand All @@ -821,6 +857,8 @@ public func run<Result, Output: OutputProtocol>(
/// as the default buffer size. Larger buffer sizes may improve performance for
/// subprocesses that produce large amounts of output, while smaller buffer sizes
/// may reduce memory usage and improve responsiveness for interactive applications.
/// - streamingBehavior: Controls the trade-off between throughput and latency when
/// streaming output from the subprocess. Defaults to `.throughput`.
/// - isolation: the isolation context to run the body closure.
/// - body: The custom configuration body to manually control
/// the running process, write to its standard input, stream
Expand All @@ -829,6 +867,7 @@ public func run<Result, Output: OutputProtocol>(
public func run<Result>(
_ configuration: Configuration,
preferredBufferSize: Int? = nil,
streamingBehavior: AsyncBufferSequence.StreamingBehavior = .throughput,
isolation: isolated (any Actor)? = #isolation,
body: (
(
Expand All @@ -851,11 +890,13 @@ public func run<Result>(
let writer = StandardInputWriter(diskIO: inputIO!)
let outputSequence = AsyncBufferSequence(
diskIO: outputIO!.consumeIOChannel(),
preferredBufferSize: preferredBufferSize
preferredBufferSize: preferredBufferSize,
streamingBehavior: streamingBehavior
)
let errorSequence = AsyncBufferSequence(
diskIO: errorIO!.consumeIOChannel(),
preferredBufferSize: preferredBufferSize
preferredBufferSize: preferredBufferSize,
streamingBehavior: streamingBehavior
)
return try await body(execution, writer, outputSequence, errorSequence)
}
Expand Down
Loading