[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.
There was a problem hiding this comment.
I'd instead add why it's not sendable. People sometimes ask about it, we should explain that so they don't assume it's just because it's "missing"
There was a problem hiding this comment.
excellent point, thank you.
we should explain that so they don't assume it's just because it's "missing"
this is, in fact, my current assumption. i presume it's currently not Sendable mainly due to path-dependence/history – the original implementations intentionally did not support concurrent consumption, but then that restriction was lifted. the implementation of the relevant types appear to be effectively Sendable (basically everything gets forwarded from the public API into the various internal 'storage' types which appear generally thread-safe).
| // 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!
There was a problem hiding this comment.
Basically assert is this is ok to not check in a released toolchain, and precondition if it should keep getting checked. Don't worry too much about the internal versions. I'm not sure the message explains the issue very well, maybe make it more descriptive why this is a crash
| 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
There was a problem hiding this comment.
It's fine to test for it I guess. I'd probably not surface it in docs yet though. Opinions @FranzBusch @phausler ?
|
ping @ktoso @FranzBusch (or lmk if i should |
|
I'll look at it but we've all just been at the swift on server conference so we've been offline a bit these days. I'll be able to review shortly once I'm back home on Monday, sorry for the delay :) |
|
@ktoso low-priority ping |
There was a problem hiding this comment.
This would be better reviewed by @phausler and @FranzBusch -- I'm a bit out of the loop what our plans with this type are.
| @@ -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.
I'd instead add why it's not sendable. People sometimes ask about it, we should explain that so they don't assume it's just because it's "missing"
| // 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.
Basically assert is this is ok to not check in a released toolchain, and precondition if it should keep getting checked. Don't worry too much about the internal versions. I'm not sure the message explains the issue very well, maybe make it more descriptive why this is a crash
| @@ -437,41 +437,257 @@ class NotSendable {} | |||
|
|
|||
| // MARK: - Multiple consumers | |||
|
|
|||
| tests.test("FIFO delivery order with multiple consumers") { | |||
There was a problem hiding this comment.
It's fine to test for it I guess. I'd probably not surface it in docs yet though. Opinions @FranzBusch @phausler ?
|
Please unmark this "draft" if it's expecting reviews -- a draft basically signals "not ready to review" @jamieQ :) Thanks for the ping and your work here! I pined some other folks who are more up to date about the plans about the stream type |
|
@FranzBusch @phausler do either of you have feedback to share on this? thanks in advance! |
|
@FranzBusch @phausler – wondering if either of you have bandwidth to provide feedback on this effort in the relatively-near term. if not, is there someone you'd suggest i try to involve instead? thanks! |
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