Introduce preferredBufferSize parameter to allow custom buffer size when streaming subprocess output

Author: iCharlesHu

Sources/Subprocess/API.swift

@@ -106,7 +106,7 @@ public func run<
 // MARK: - Custom Execution Body
 
 /// Run an executable with given parameters and a custom closure
-/// to manage the running subprocess' lifetime and stream its standard output.
+/// to manage the running subprocess' lifetime.
 /// - Parameters:
 ///   - executable: The executable to run.
 ///   - arguments: The arguments to pass to the executable.
@@ -160,6 +160,11 @@ public func run<Result, Input: InputProtocol, Output: OutputProtocol, Error: Out
 ///     when running the executable.
 ///   - input: The input to send to the executable.
 ///   - error: How to manager executable standard error.
+///   - preferredBufferSize: The preferred size in bytes for the buffer used when reading
+///     from the subprocess's standard error stream. If `nil`, uses the system page size
+///     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.
 ///   - 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
@@ -172,6 +177,7 @@ public func run<Result, Input: InputProtocol, Error: OutputProtocol>(
     platformOptions: PlatformOptions = PlatformOptions(),
     input: Input = .none,
     error: Error = .discarded,
+    preferredOutputSize: Int? = nil,
     isolation: isolated (any Actor)? = #isolation,
     body: ((Execution, AsyncBufferSequence) async throws -> Result)
 ) async throws -> ExecutionResult<Result> where Error.OutputType == Void {
@@ -186,6 +192,7 @@ public func run<Result, Input: InputProtocol, Error: OutputProtocol>(
         configuration,
         input: input,
         error: error,
+        preferredBufferSize: preferredOutputSize,
         isolation: isolation,
         body: body
     )
@@ -202,6 +209,11 @@ public func run<Result, Input: InputProtocol, Error: OutputProtocol>(
 ///     when running the executable.
 ///   - input: The input to send to the executable.
 ///   - output: How to manager executable standard output.
+///   - preferredBufferSize: The preferred size in bytes for the buffer used when reading
+///     from the subprocess's standard error stream. If `nil`, uses the system page size
+///     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.
 ///   - 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
@@ -214,6 +226,7 @@ public func run<Result, Input: InputProtocol, Output: OutputProtocol>(
     platformOptions: PlatformOptions = PlatformOptions(),
     input: Input = .none,
     output: Output,
+    preferredBufferSize: Int? = nil,
     isolation: isolated (any Actor)? = #isolation,
     body: ((Execution, AsyncBufferSequence) async throws -> Result)
 ) async throws -> ExecutionResult<Result> where Output.OutputType == Void {
@@ -228,6 +241,7 @@ public func run<Result, Input: InputProtocol, Output: OutputProtocol>(
         configuration,
         input: input,
         output: output,
+        preferredBufferSize: preferredBufferSize,
         isolation: isolation,
         body: body
     )
@@ -244,6 +258,11 @@ public func run<Result, Input: InputProtocol, Output: OutputProtocol>(
 ///   - platformOptions: The platform specific options to use
 ///     when running the executable.
 ///   - error: How to manager executable standard error.
+///   - preferredBufferSize: The preferred size in bytes for the buffer used when reading
+///     from the subprocess's standard output stream. If `nil`, uses the system page size
+///     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.
 ///   - 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
@@ -255,6 +274,7 @@ public func run<Result, Error: OutputProtocol>(
     workingDirectory: FilePath? = nil,
     platformOptions: PlatformOptions = PlatformOptions(),
     error: Error = .discarded,
+    preferredBufferSize: Int? = nil,
     isolation: isolated (any Actor)? = #isolation,
     body: ((Execution, StandardInputWriter, AsyncBufferSequence) async throws -> Result)
 ) async throws -> ExecutionResult<Result> where Error.OutputType == Void {
@@ -268,6 +288,7 @@ public func run<Result, Error: OutputProtocol>(
     return try await run(
         configuration,
         error: error,
+        preferredBufferSize: preferredBufferSize,
         isolation: isolation,
         body: body
     )
@@ -284,6 +305,11 @@ public func run<Result, Error: OutputProtocol>(
 ///   - platformOptions: The platform specific options to use
 ///     when running the executable.
 ///   - output: How to manager executable standard output.
+///   - preferredBufferSize: The preferred size in bytes for the buffer used when reading
+///     from the subprocess's standard error stream. If `nil`, uses the system page size
+///     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.
 ///   - 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
@@ -295,6 +321,7 @@ public func run<Result, Output: OutputProtocol>(
     workingDirectory: FilePath? = nil,
     platformOptions: PlatformOptions = PlatformOptions(),
     output: Output,
+    preferredBufferSize: Int? = nil,
     isolation: isolated (any Actor)? = #isolation,
     body: ((Execution, StandardInputWriter, AsyncBufferSequence) async throws -> Result)
 ) async throws -> ExecutionResult<Result> where Output.OutputType == Void {
@@ -308,6 +335,7 @@ public func run<Result, Output: OutputProtocol>(
     return try await run(
         configuration,
         output: output,
+        preferredBufferSize: preferredBufferSize,
         isolation: isolation,
         body: body
     )
@@ -323,6 +351,11 @@ public func run<Result, Output: OutputProtocol>(
 ///   - workingDirectory: The working directory in which to run the executable.
 ///   - platformOptions: The platform specific options to use
 ///     when running the executable.
+///   - preferredBufferSize: The preferred size in bytes for the buffer used when reading
+///     from the subprocess's standard output and error stream. If `nil`, uses the system page size
+///     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.
 ///   - 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
@@ -333,6 +366,7 @@ public func run<Result>(
     environment: Environment = .inherit,
     workingDirectory: FilePath? = nil,
     platformOptions: PlatformOptions = PlatformOptions(),
+    preferredBufferSize: Int? = nil,
     isolation: isolated (any Actor)? = #isolation,
     body: (
         (
@@ -352,6 +386,7 @@ public func run<Result>(
     )
     return try await run(
         configuration,
+        preferredBufferSize: preferredBufferSize,
         isolation: isolation,
         body: body
     )
@@ -561,6 +596,11 @@ public func run<Result, Input: InputProtocol, Output: OutputProtocol, Error: Out
 ///   - configuration: The configuration to run.
 ///   - input: The input to send to the executable.
 ///   - error: How to manager executable standard error.
+///   - preferredBufferSize: The preferred size in bytes for the buffer used when reading
+///     from the subprocess's standard output stream. If `nil`, uses the system page size
+///     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.
 ///   - 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
@@ -569,6 +609,7 @@ public func run<Result, Input: InputProtocol, Error: OutputProtocol>(
     _ configuration: Configuration,
     input: Input = .none,
     error: Error = .discarded,
+    preferredBufferSize: Int? = nil,
     isolation: isolated (any Actor)? = #isolation,
     body: ((Execution, AsyncBufferSequence) async throws -> Result)
 ) async throws -> ExecutionResult<Result> where Error.OutputType == Void {
@@ -594,7 +635,10 @@ public func run<Result, Input: InputProtocol, Error: OutputProtocol>(
             }
 
             // Body runs in the same isolation
-            let outputSequence = AsyncBufferSequence(diskIO: outputIOBox.take()!.consumeIOChannel())
+            let outputSequence = AsyncBufferSequence(
+                diskIO: outputIOBox.take()!.consumeIOChannel(),
+                preferredBufferSize: preferredBufferSize
+            )
             let result = try await body(execution, outputSequence)
             try await group.waitForAll()
             return result
@@ -608,6 +652,11 @@ public func run<Result, Input: InputProtocol, Error: OutputProtocol>(
 ///   - configuration: The configuration to run.
 ///   - input: The input to send to the executable.
 ///   - output: How to manager executable standard output.
+///   - preferredBufferSize: The preferred size in bytes for the buffer used when reading
+///     from the subprocess's standard error stream. If `nil`, uses the system page size
+///     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.
 ///   - 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
@@ -616,6 +665,7 @@ public func run<Result, Input: InputProtocol, Output: OutputProtocol>(
     _ configuration: Configuration,
     input: Input = .none,
     output: Output,
+    preferredBufferSize: Int? = nil,
     isolation: isolated (any Actor)? = #isolation,
     body: ((Execution, AsyncBufferSequence) async throws -> Result)
 ) async throws -> ExecutionResult<Result> where Output.OutputType == Void {
@@ -641,7 +691,10 @@ public func run<Result, Input: InputProtocol, Output: OutputProtocol>(
             }
 
             // Body runs in the same isolation
-            let errorSequence = AsyncBufferSequence(diskIO: errorIOBox.take()!.consumeIOChannel())
+            let errorSequence = AsyncBufferSequence(
+                diskIO: errorIOBox.take()!.consumeIOChannel(),
+                preferredBufferSize: preferredBufferSize
+            )
             let result = try await body(execution, errorSequence)
             try await group.waitForAll()
             return result
@@ -655,13 +708,19 @@ public func run<Result, Input: InputProtocol, Output: OutputProtocol>(
 /// - Parameters:
 ///   - configuration: The `Configuration` to run.
 ///   - error: How to manager executable standard error.
+///   - preferredBufferSize: The preferred size in bytes for the buffer used when reading
+///     from the subprocess's standard output stream. If `nil`, uses the system page size
+///     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.
 ///   - 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.
 public func run<Result, Error: OutputProtocol>(
     _ configuration: Configuration,
     error: Error = .discarded,
+    preferredBufferSize: Int? = nil,
     isolation: isolated (any Actor)? = #isolation,
     body: ((Execution, StandardInputWriter, AsyncBufferSequence) async throws -> Result)
 ) async throws -> ExecutionResult<Result> where Error.OutputType == Void {
@@ -673,7 +732,10 @@ 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!.consumeIOChannel())
+        let outputSequence = AsyncBufferSequence(
+            diskIO: outputIO!.consumeIOChannel(),
+            preferredBufferSize: preferredBufferSize
+        )
         return try await body(execution, writer, outputSequence)
     }
 }
@@ -684,13 +746,19 @@ public func run<Result, Error: OutputProtocol>(
 /// - Parameters:
 ///   - configuration: The `Configuration` to run.
 ///   - output: How to manager executable standard output.
+///   - preferredBufferSize: The preferred size in bytes for the buffer used when reading
+///     from the subprocess's standard error stream. If `nil`, uses the system page size
+///     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.
 ///   - 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.
 public func run<Result, Output: OutputProtocol>(
     _ configuration: Configuration,
     output: Output,
+    preferredBufferSize: Int? = nil,
     isolation: isolated (any Actor)? = #isolation,
     body: ((Execution, StandardInputWriter, AsyncBufferSequence) async throws -> Result)
 ) async throws -> ExecutionResult<Result> where Output.OutputType == Void {
@@ -702,14 +770,24 @@ 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!.consumeIOChannel())
+        let errorSequence = AsyncBufferSequence(
+            diskIO: errorIO!.consumeIOChannel(),
+            preferredBufferSize: preferredBufferSize
+        )
         return try await body(execution, writer, errorSequence)
     }
 }
 
 /// Run an executable with given parameters specified by a `Configuration`
+/// and a custom closure to manage the running subprocess' lifetime, write to its
+/// standard input, and stream its standard output and standard error.
 /// - Parameters:
 ///   - configuration: The `Subprocess` configuration to run.
+///   - preferredBufferSize: The preferred size in bytes for the buffer used when reading
+///     from the subprocess's standard output and error stream. If `nil`, uses the system page size
+///     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.
 ///   - 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
@@ -718,6 +796,7 @@ public func run<Result, Output: OutputProtocol>(
 ///     of the closure.
 public func run<Result>(
     _ configuration: Configuration,
+    preferredBufferSize: Int? = nil,
     isolation: isolated (any Actor)? = #isolation,
     body: ((Execution, StandardInputWriter, AsyncBufferSequence, AsyncBufferSequence) async throws -> Result)
 ) async throws -> ExecutionResult<Result> {
@@ -730,8 +809,14 @@ public func run<Result>(
         error: try error.createPipe()
     ) { execution, inputIO, outputIO, errorIO in
         let writer = StandardInputWriter(diskIO: inputIO!)
-        let outputSequence = AsyncBufferSequence(diskIO: outputIO!.consumeIOChannel())
-        let errorSequence = AsyncBufferSequence(diskIO: errorIO!.consumeIOChannel())
+        let outputSequence = AsyncBufferSequence(
+            diskIO: outputIO!.consumeIOChannel(),
+            preferredBufferSize: preferredBufferSize
+        )
+        let errorSequence = AsyncBufferSequence(
+            diskIO: errorIO!.consumeIOChannel(),
+            preferredBufferSize: preferredBufferSize
+        )
         return try await body(execution, writer, outputSequence, errorSequence)
     }
 }

Sources/Subprocess/AsyncBufferSequence.swift

@@ -36,11 +36,13 @@ public struct AsyncBufferSequence: AsyncSequence, @unchecked Sendable {
         public typealias Element = Buffer
 
         private let diskIO: DiskIO
+        private let preferredBufferSize: Int
         private var buffer: [Buffer]
 
-        internal init(diskIO: DiskIO) {
+        internal init(diskIO: DiskIO, preferredBufferSize: Int?) {
             self.diskIO = diskIO
             self.buffer = []
+            self.preferredBufferSize = preferredBufferSize ?? readBufferSize
         }
 
         public mutating func next() async throws -> Buffer? {
@@ -51,7 +53,7 @@ public struct AsyncBufferSequence: AsyncSequence, @unchecked Sendable {
             // Read more data
             let data = try await AsyncIO.shared.read(
                 from: self.diskIO,
-                upTo: readBufferSize
+                upTo: self.preferredBufferSize
             )
             guard let data else {
                 // We finished reading. Close the file descriptor now
@@ -77,13 +79,18 @@ public struct AsyncBufferSequence: AsyncSequence, @unchecked Sendable {
     }
 
     private let diskIO: DiskIO
+    private let preferredBufferSize: Int?
 
-    internal init(diskIO: DiskIO) {
+    internal init(diskIO: DiskIO, preferredBufferSize: Int?) {
         self.diskIO = diskIO
+        self.preferredBufferSize = preferredBufferSize
     }
 
     public func makeAsyncIterator() -> Iterator {
-        return Iterator(diskIO: self.diskIO)
+        return Iterator(
+            diskIO: self.diskIO,
+            preferredBufferSize: self.preferredBufferSize
+        )
     }
 
     // [New API: 0.0.1]