[stdlib]: better align Async{Throwing}Stream implementations #76571
Conversation
Removes some of the discrepancies between the throwing & non-throwing variants of AsyncStream. In particular, adds multi-consumer support to the throwing variant, and updates the docs. Additional minor refactoring of existing implementation to streamline the code. Fixes swiftlang#76547
There was a problem hiding this comment.
@FranzBusch @ktoso i'd again appreciate your insights on the proposed changes here, if & when you have some time. thanks in advance!
| @@ -356,11 +356,6 @@ public struct AsyncStream<Element> { | |||
| @available(SwiftStdlib 5.1, *) | |||
| extension AsyncStream: AsyncSequence { | |||
| /// The asynchronous iterator for iterating an asynchronous stream. | |||
| /// | |||
| /// This type doesn't conform to `Sendable`. Don't use it from multiple | |||
There was a problem hiding this comment.
the 'this type isn't Sendable' assertion is still accurate, but i'm not sure it's particularly relevant or helpful to call out. the other warnings and references to fatalError() are no longer correct.
| // Presence of continuations implies no pending elements. | ||
| // TODO: which assertion flavor should be used? | ||
| assert( | ||
| state.pending.isEmpty, | ||
| "Continuations should imply no pending elements." | ||
| ) |
There was a problem hiding this comment.
i would appreciate feedback on a reasonable way to enforce this invariant. per the docs regarding assertion flavors in the stdlib, i would be inclined to think that _internalInvariant() is maybe the most appropriate function to use here. however, that symbol isn't visible in this context since this code resides in the Concurrency module. any thoughts on this, or references to prior art would be appreciated!
| unlock() | ||
| continuation.resume(returning: toSend) | ||
| } else if state.terminal { | ||
| if state.terminal { |
There was a problem hiding this comment.
per @FranzBusch's prior observation, the branches in which we were handling cases that implied both pending elements and outstanding continuations should in fact be dead. the rationale is that elements should only be buffered if there are no awaiting consumers, and continuations are stored only if there are no pending elements. sanity checks appreciated regarding this analysis, however.
| @@ -287,7 +267,7 @@ extension AsyncThrowingStream { | |||
| } | |||
|
|
|||
| struct State { | |||
| var continuation: UnsafeContinuation<Element?, Error>? | |||
| var continuations = [UnsafeContinuation<Element?, Error>]() | |||
There was a problem hiding this comment.
primary behavioral changes to make AsyncThrowingStream better match AsyncStream's existing behavior start here.
| @@ -437,41 +437,257 @@ class NotSendable {} | |||
|
|
|||
| // MARK: - Multiple consumers | |||
|
|
|||
| tests.test("FIFO delivery order with multiple consumers") { | |||
There was a problem hiding this comment.
i'm not sure this delivery order is ever publicly stated as a guarantee of these types, but it is how they currently work
Removes some of the discrepancies between the throwing & non-throwing variants of AsyncStream. In particular, adds multi-consumer support to the throwing variant, and updates the docs. Additional minor refactoring of existing implementation to streamline the code.
Fixes #76547