Allow callers to run a subprocess and provide low and high water marks when using SequenceOutput to emit standard output and standard error as soon as it arrives. #40
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -35,44 +35,68 @@ public struct AsyncBufferSequence: AsyncSequence, Sendable { | |
| @_nonSendable | ||
| public struct Iterator: AsyncIteratorProtocol { | ||
| public typealias Element = Buffer | ||
| internal typealias Stream = AsyncThrowingStream<Buffer, Swift.Error> | ||
|
|
||
| private let diskIO: DiskIO | ||
| private var buffer: [UInt8] | ||
| private var currentPosition: Int | ||
| private var finished: Bool | ||
| private var streamIterator: Stream.AsyncIterator | ||
| private let continuation: Stream.Continuation | ||
| private var bytesRemaining: Int | ||
|
|
||
| internal init(diskIO: DiskIO, streamOptions: PlatformOptions.StreamOptions) { | ||
| self.diskIO = diskIO | ||
| self.buffer = [] | ||
| self.currentPosition = 0 | ||
| self.finished = false | ||
| let (stream, continuation) = AsyncThrowingStream<Buffer, Swift.Error>.makeStream() | ||
| self.streamIterator = stream.makeAsyncIterator() | ||
| self.continuation = continuation | ||
| self.bytesRemaining = 0 | ||
|
|
||
| #if !os(Windows) | ||
| if let minimumBufferSize = streamOptions.minimumBufferSize { | ||
| diskIO.setLimit(lowWater: minimumBufferSize) | ||
| } | ||
|
|
||
| if let maximumBufferSize = streamOptions.maximumBufferSize { | ||
| diskIO.setLimit(highWater: maximumBufferSize) | ||
| } | ||
| #endif | ||
| } | ||
|
|
||
| public mutating func next() async throws -> Buffer? { | ||
|
|
||
| if bytesRemaining <= 0 { | ||
| bytesRemaining = readBufferSize | ||
| diskIO.stream(upToLength: readBufferSize, continuation: continuation) | ||
|
There was a problem hiding this comment. I don't think you need to manually stream only up to There was a problem hiding this comment. Are we sure we want that? That exacerbates the concerns of @weissi and @FranzBusch. With the current approach, the easing of back pressure is limited to buffering up to about There was a problem hiding this comment. I was trying to unblock this PR so I created #51 to address the However, after some time to think over and studying @FranzBusch’s suggestion (apple/swift-async-algorithms#305.), I’ve come to the conclusion that we can’t make the The current method (AsyncStream + buffer size limit) is better because it prevents the unlimited buffer problem, but it loses the advantage of letting Therefore, for this PR, could you please:
I believe the above approach should still allow callers to set how frequent they wish to be updated without needing to introduce |
||
| } | ||
|
|
||
| if let buffer = try await streamIterator.next() { | ||
| bytesRemaining -= buffer.count | ||
| return buffer | ||
| } else { | ||
| #if os(Windows) | ||
| try self.diskIO.close() | ||
| #else | ||
| self.diskIO.close() | ||
| #endif | ||
| return nil | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private let diskIO: DiskIO | ||
| private let streamOptions: PlatformOptions.StreamOptions | ||
|
|
||
| internal init(diskIO: DiskIO, streamOptions: PlatformOptions.StreamOptions) { | ||
| self.diskIO = diskIO | ||
| self.streamOptions = streamOptions | ||
| } | ||
|
|
||
| public func makeAsyncIterator() -> Iterator { | ||
| return Iterator(diskIO: self.diskIO, streamOptions: streamOptions) | ||
| } | ||
| } | ||
|
|
||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -662,6 +662,50 @@ extension SubprocessUnixTests { | |
| #expect(catResult.terminationStatus.isSuccess) | ||
| #expect(catResult.standardError == expected) | ||
| } | ||
|
|
||
| @Test func testSlowDripRedirectedOutputRedirectToSequence() async throws { | ||
| guard #available(SubprocessSpan , *) else { | ||
| return | ||
| } | ||
| let threshold: Double = 0.5 | ||
|
There was a problem hiding this comment. Unfortunately in tests you'll have to write In the beginning to work around the same availability issue. See other tests for examples. There was a problem hiding this comment. @iCharlesHu When I first started working on this, I was very confused as to why some of the tests weren't running my new code and it was because of this check. Wouldn't it be better to have them skipped and noted as such in the test output rather than falsely succeeding? I'm thinking something like this: Of course, we can have a helper function to make this less verbose. Thoughts? There was a problem hiding this comment. @iCharlesHu I went ahead and conditionalized this one test this way as an example. Let me know if you don't like that and would like me to revert to a There was a problem hiding this comment. @rdingman unfortunately this won't work because on Swift 6.2 and above the compiler will complain the code inside of Unfortunately so far this is the only way I found that works. Very ugly... but it'll have to do for now. There was a problem hiding this comment. @iCharlesHu Hmm. This builds just fine for me with the latest Swift 6.2 toolchain, has the intended results, and works with the currently shipping Swift 6.1 toolchain. However, I'll go ahead and revert this and match the existing tests. |
||
|
|
||
| let script = """ | ||
| echo "DONE" | ||
| sleep \(threshold) | ||
| """ | ||
|
|
||
| var platformOptions = PlatformOptions() | ||
| platformOptions.streamOptions = .init(minimumBufferSize: 0) | ||
|
|
||
| let start = ContinuousClock().now | ||
|
|
||
| let catResult = try await Subprocess.run( | ||
| .path("/bin/bash"), | ||
| arguments: ["-c", script], | ||
| platformOptions: platformOptions, | ||
| error: .discarded, | ||
| body: { (execution, standardOutput) in | ||
| for try await chunk in standardOutput { | ||
| let string = chunk._withUnsafeBytes { String(decoding: $0, as: UTF8.self) } | ||
|
|
||
| if string.hasPrefix("DONE") { | ||
| let end = ContinuousClock().now | ||
|
|
||
| if (end - start) > .seconds(threshold) { | ||
| return "Failure" | ||
|
|
||
| } else { | ||
| return "Success" | ||
| } | ||
| } | ||
| } | ||
|
|
||
| return "Failure" | ||
| } | ||
| ) | ||
| #expect(catResult.terminationStatus.isSuccess) | ||
| #expect(catResult.value == "Success") | ||
| } | ||
| } | ||
|
|
||
| // MARK: - PlatformOption Tests | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I'm a bit concerned with the usage of the
AsyncStreamhere for two reasons. The first is thatAsyncStreamhas no way to propagate the internal async sequence back pressure to the external system that is producing the elements.We had the same issue in
swift-nioand created the NIOAsyncSequenceProducer that allowed us to bridge NIO's back pressure into the back pressure of an async sequence. The learned a lot from the APIs and implementation in NIO and have an open PR inswift-async-algorithmsthat generalizes this concept as a MultiProducerSingleConsumerAsyncChannel.Now having said that it might be fine here back pressure wise since the
DispatchIOis going to call theioHandlermultiple times but only up toreadBufferSize. So we do have some maximum limit and our buffer can't balloon indefinitely.However, this brings me to my second point which is the performance of this.
AsyncStreamis not super fast so I expect this PR to have performance impact when streaming a lot of data to/from a subprocess. It would be good to understand that impact and if theMultiProducerSingleConsumerAsyncChannelcan improve this.Lastly, aren't we missing to close the
diskIOwhen the iterator is dropped. We probably want to setup aonTerminationCallbackon the stream or the channel to close the diskIO right?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I'll look into this in a separate PR. I think the
AsyncStreambased implementation is good enough for this PR.Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Will investigate under #51