Add `MultiProducerSingleConsumerChannel` #305

FranzBusch · 2023-12-19T11:29:56Z

Motivation

The pitch to add external backpressure support to the standard libraries AsyncStream got returned for revision since there are larger open questions around AsyncSequence. However, having external backpressure in a source asynchronous sequence is becoming more and more important.

Modification

This PR adds a modified proposal and implementation that brings the Swift Evolution proposal over to Swift Async Algorithms.

FranzBusch · 2023-12-20T09:57:51Z

@swift-ci please test

glbrntt · 2024-06-14T14:46:21Z

Evolution/0016-mutli-producer-single-consumer-channel.md

+
+### Upstream producer termination
+
+The producer will get notified about termination through the `onTerminate`


nit: can you add calls to onTerminate to the snippets below?

glbrntt · 2024-06-14T14:48:13Z

Evolution/0016-mutli-producer-single-consumer-channel.md

+        })
+    }
+} catch {
+    // `send(contentsOf:)` throws if the asynchronous stream already terminated


Would this be clearer as another case on the sendResult?

glbrntt · 2024-06-14T14:50:59Z

Evolution/0016-mutli-producer-single-consumer-channel.md

+            /// - Parameters:
+            ///   - low: When the number of buffered elements drops below the low watermark, producers will be resumed.
+            ///   - high: When the number of buffered elements rises above the high watermark, producers will be suspended.
+            ///   - waterLevelForElement: A closure used to compute the contribution of each buffered element to the current water level.


Oh, neat. This works nicely for collections.

...orithms/MultiProducerSingleConsumerChannel/MultiProducerSingleConsumerChannel+Internal.swift

.../AsyncAlgorithms/MultiProducerSingleConsumerChannel/MultiProducerSingleConsumerChannel.swift

simonjbeaumont · 2024-07-09T14:53:05Z

.../AsyncAlgorithms/MultiProducerSingleConsumerChannel/MultiProducerSingleConsumerChannel.swift

+//
+//===----------------------------------------------------------------------===//
+
+#if compiler(>=6.0)


Please can we consider offering a pre-Swift 6 variant of this type. There are several places we'd like to adopt this that need to also support Swift 5.9+ for the foreseeable.

Hm not sure yet. Especially since it is part of the public API in the Source. Making the Source copyable makes the type behave way differently.

.../AsyncAlgorithms/MultiProducerSingleConsumerChannel/MultiProducerSingleConsumerChannel.swift

ser-0xff · 2024-10-23T14:34:25Z

Hi!
Thank you for the PR!
We are looking for the functionality you did in MultiProducerSingleConsumerChannel and tried to play with it.
Unfortunately we were not able to compile it because we need it in the framework which is compiled with evolution mode.
I did some changes to make it compile, but not sure these changes are good enough to be a part of public framework.
Do you have any plans to support evolution mode for the AsyncAlgorithms?
We prepared a small reproducing project for you for a case if you will have some time to look at that.
Just

git clone https://github.com/ordo-one/external-reproducers.git
cd external-reproducers/swift/async-algorithms-evolution
xcodebuild archive -quiet -scheme async-algorithms-evolution -configuration Debug -destination 'platform=macOS,arch=arm64' BUILD_LIBRARY_FOR_DISTRIBUTION=YES

FranzBusch · 2024-10-23T15:51:19Z

@ser-0xff Thanks for trying it out. We don't plan on support library evolution mode in any of our packages. What we normally advise is to not compile packages as frameworks but rather add this library as an internal import only dependency inside the frameworks that need it.

ser-0xff · 2024-10-24T07:34:19Z

Thank you for the reply...
Unfortunately internal import does not work if we want to use package type (like MultiProducerSingleConsumerChannel) in the framework public API...

hassila · 2024-10-24T10:34:32Z

Thank you for the reply... Unfortunately internal import does not work if we want to use package type (like MultiProducerSingleConsumerChannel) in the framework public API...

@ser-0xff I guess we need to wrap up the API surface we want to support then and just do internal import as suggested? But let's take that offline from this PR - thanks @FranzBusch for quick reply.

ser-0xff · 2024-10-24T11:45:32Z

Hi, @FranzBusch
We run into another issue when we played with the MultiProducerSingleConsumerChannel.
Our projects are already on Swift6 which has a stricter non-copyable types checks. The MultiProducerSingleConsumerChannel<>.Source type is non-copyable so we can't share it across many tasks, as a result we can use it only in single producer mode.
How could we solve that issue and produce data within multiple tasks?

Here is a minimized test showing the usage pattern we want if you will find some time to look at.

FranzBusch · 2024-10-28T10:10:30Z

We run into another issue when we played with the MultiProducerSingleConsumerChannel.
Our projects are already on Swift6 which has a stricter non-copyable types checks. The MultiProducerSingleConsumerChannel<>.Source type is non-copyable so we can't share it across many tasks, as a result we can use it only in single producer mode.
How could we solve that issue and produce data within multiple tasks?

The source has a copy method that allows you to get a second source. Now the problem is that closures cannot be marked as ~Copyable or rathe @calledOnce. To solution to this is to box the source into a class right now and move it into a task. This either requires a lock or an @unchecked Sendable annotation right now.

ser-0xff · 2024-10-28T14:19:05Z

Missed copy() method in source code.
Will look at that, thank you for the reply.

ser-0xff · 2024-11-05T14:34:34Z

HI, Franz!
We tried using MultiProducersSingleConsumerChannel instead of AsyncStream and encountered behavior unexpected for us. We expected the channel to remain functional as long as at least one associated source was active. However, in the current implementation, it appears that the channel transitions to a finished state as soon as any single source instance is deinitialized.
I created a PR with minimized test into your feature branch for a case if you could find some time to look at that and may be later will include that test to the test suite for MultiProducersSingleConsumerChannel.

FranzBusch · 2024-11-11T13:35:45Z

HI, Franz! We tried using MultiProducersSingleConsumerChannel instead of AsyncStream and encountered behavior unexpected for us. We expected the channel to remain functional as long as at least one associated source was active. However, in the current implementation, it appears that the channel transitions to a finished state as soon as any single source instance is deinitialized. I created a PR with minimized test into your feature branch for a case if you could find some time to look at that and may be later will include that test to the test suite for MultiProducersSingleConsumerChannel.

You are right. I haven't gotten around to implement the correct handling of multiple sources yet.

phausler

I think with some minor edits the pitch is good to go; id like to at the bare minimum see some of that testing reenabled to ensure it isn't leaking.

phausler · 2025-03-24T20:33:34Z

Sources/AsyncAlgorithms/Internal/_TinyArray.swift

+/// It supports arbitrary many elements but if only up to one ``Element`` is stored it does **not** allocate separate storage on the heap
+/// and instead stores the ``Element`` inline.
+@usableFromInline
+struct _TinyArray<Element> {


I wonder if this could perhaps be subsumed by InlineArray instead since the original intent was to have a faster storage (this is not a blocking concept)

InlineArray isn't capable of fully replacing the need here. The idea behind TinyArray is to have a fast path for a single element that doesn't allocate and one that allocates if there are more. Since we can't tell at compile time how many producers we have we need this runtime dynamism.

phausler · 2025-03-24T20:34:54Z

Sources/AsyncAlgorithms/Locking.swift

@@ -17,24 +17,35 @@ import Glibc
 import WinSDK
 #endif

-internal struct Lock {
+@usableFromInline
+internal class Lock {


umm why is this a non-final class? If we are moving to a class then we should just use Mutex from Synchronization

I removed all of the changes here since the latest code now uses Mutex.

.../AsyncAlgorithms/MultiProducerSingleConsumerChannel/MultiProducerSingleConsumerChannel.swift

phausler · 2025-03-24T20:40:47Z

...rithmsTests/MultiProducerSingleConsumerChannel/MultiProducerSingleConsumerChannelTests.swift

+final class MultiProducerSingleConsumerChannelTests: XCTestCase {
+    // MARK: - sequenceDeinitialized
+
+    // Following tests are disabled since the channel is not getting deinited due to a known bug


This might be a requirement to landing; id rather not introduce a type with a known leak

I added significantly more tests now and also re-enabled those.

phausler · 2025-03-24T20:42:15Z

Evolution/0016-mutli-producer-single-consumer-channel.md

+
+```swift
+// Writing new values and providing a callback when to produce more
+try source.send(contentsOf: sequence, onProduceMore: { result in


can this be written without ambiguity as a trailing closure?

Yes in a synchronous context both send methods can be called without ambiguity, e.g. this compiles:

try source.send(1) source.send(2) { result in print(result) }

The only thing that needs some special handling is if you want to call a sync send from an async context since the async methods are preferred; however, users can workaround this by creating an inline closure, e.g.:

try { try source.send(1) }()

# Motivation The pitch to add external backpressure support to the standard libraries `AsyncStream` got returned for revision since there are larger open questions around `AsyncSequence`. However, having external backpressure in a source asynchronous sequence is becoming more and more important. # Modification This PR adds a modified proposal and implementation that brings the Swift Evolution proposal over to Swift Async Algorithms.

jamieQ · 2025-03-30T18:32:16Z

.../AsyncAlgorithms/MultiProducerSingleConsumerChannel/MultiProducerSingleConsumerChannel.swift

+    /// The channel will only automatically be finished if all existing sources have been deinited.
+    ///
+    /// - Returns: A new source for sending elements to the channel.
+    public mutating func copy() -> sending Self {


here (and a in a few other places) we have sending Self where Self conforms to Sendable. in such cases is the sending doing anything meaningful, since the type can already be freely passed across isolations? i think adding a diagnostic to flag such cases as redundant has also been proposed/drafted. i suppose if the Sendable conformance were to be dropped in the future, maybe such a change would be 'less breaking' if things are spelled this way?

Thanks for calling this out. I think the right thing is to actually remove the Sendable conformance on Source since it is in fact not Sendable.

jamieQ · 2025-03-30T18:37:03Z

.../AsyncAlgorithms/MultiProducerSingleConsumerChannel/MultiProducerSingleConsumerChannel.swift

+      /// - Parameters:
+      ///   - low: When the number of buffered elements drops below the low watermark, producers will be resumed.
+      ///   - high: When the number of buffered elements rises above the high watermark, producers will be suspended.
+      public static func watermark(low: Int, high: Int) -> BackpressureStrategy {


apologies if i missed this having been discussed somewhere in the prior history of this work, but did you consider using a ClosedRange of an appropriate unsigned integer type to model the watermark state? naively it seems that might obviate some of the invariant checks.

I think it's interesting but I personally prefer having to Int's here since the watermark strategy is not really about the range but a low and high limit where the production is started and stopped respectively.

jamieQ · 2025-03-30T18:42:36Z

...orithms/MultiProducerSingleConsumerChannel/MultiProducerSingleConsumerChannel+Internal.swift

+      mutating func didSend(elements: Deque<Element>.SubSequence) -> Bool {
+        if let waterLevelForElement = self._waterLevelForElement {
+          for element in elements {
+            self._currentWatermark += waterLevelForElement(element)


it looked to me like this logic runs at various points while the state machine lock is held – any concerns about invoking client code while holding a lock?

I updated the public docs to call this out. I think the reality is that we need to call this code while being under a lock since the state management otherwise would become almost impossible with potentially any code interleaving.

phausler

Over all I think this is a lot of complicated machinery, it is definitely filling a gap of use; but we need to be careful about how widely this is expected to be used; the long name right now is definitely keeping that progressive disclosure correct, but I think we can soften some of the requirements a bit and find a better name; right now it is pretty wordy.

All that being said; id say that since the need (and conjunction with other efforts makes it reasonable) then it is worth running this to get some additional feedback by focusing the review on how we can make some of the uses a bit more straightforward and suggestions on what the appropriate name is.

phausler · 2025-08-06T20:25:01Z

Evolution/0016-mutli-producer-single-consumer-channel.md

+///
+/// This error is thrown when the channel is already finished when
+/// trying to send new elements to the source.
+public struct MultiProducerSingleConsumerAsyncChannelAlreadyFinishedError: Error { }


This seems like it should be a nested type; it is a keyboard full of naming here...

I agree it is a long but we can't nest it due to the MPSCCAsyncChannel being generic.

I guess folks aren't going to be typing this type out much eh, so perhaps not a huge deal; do we really even need this type to be public?

phausler · 2025-08-06T20:30:30Z

Evolution/0016-mutli-producer-single-consumer-channel.md

+system and compares them to the behaviors of `Async[Throwing]Stream` and
+`Async[Throwing]Channel`.
+
+This section proposes a new type called `MultiProducerSingleConsumerAsyncChannel`


Id really love to come up with a smaller name. We have been using water metaphors, so perhaps something in that family of names? From my understanding of the pitch it is like you have many small tributaries contributing to a larger stream of values. Id like to come up with some alternatives than this.

What if we call it MPSCAsyncChannel. The German in me does like long explicit wordings but I'm happy to go with abbreviations here. The other name I was thinking about was MultiProducerAsyncChannel or ExternalBackpressuredAsyncChannel.

MultiProducerAsyncChannel sounds good to me!

4 words is better than 6 I guess...

phausler · 2025-08-06T20:37:09Z

Evolution/0016-mutli-producer-single-consumer-channel.md

+            ///
+            /// - Important: This token must only be passed once to ``MultiProducerSingleConsumerAsyncChannel/Source/enqueueCallback(callbackToken:onProduceMore:)``
+            ///  and ``MultiProducerSingleConsumerAsyncChannel/Source/cancelCallback(callbackToken:)``.
+            public struct CallbackToken: Sendable, Hashable { }


Is it possible that these tokens are somehow user provided?

If you mean constructed by the user then no. There is no public init for those.

no, I was suggesting that the user provides a type to represent a token and this could be token that the framework provides. It would mean that things would need to be flipped around to be generic around that.

There are some APIs that have tokens associated with callbacks. Thinking particularly the CF APIs or the KVO APIs

I see what you mean I would like to avoid that though since we are using those tokens to understand if there is a potential API mis-use. We wouldn't be able to do this if the user were to create those tokens.

phausler · 2025-08-06T20:38:12Z

Evolution/0016-mutli-producer-single-consumer-channel.md

+        /// A callback to invoke when the channel finished.
+        ///
+        /// This is called after the last element has been consumed by the channel.
+        public func setOnTerminationCallback(_ callback: @escaping @Sendable () -> Void)


it might be useful to have a parameter in the closure to tell why the termination happened, either finished, failed, or cancelled.

FranzBusch force-pushed the fb-async-backpressured-stream branch from ddd7523 to 65e8957 Compare December 19, 2023 11:49

FranzBusch requested a review from phausler December 20, 2023 09:57

FranzBusch force-pushed the fb-async-backpressured-stream branch from 4904039 to 662693e Compare June 14, 2024 08:15

FranzBusch changed the title ~~Add AsyncBackpressuredStream proposal and implementation~~ Add MultiProducerSingleConsumerChannel Jun 14, 2024

glbrntt reviewed Jun 14, 2024

View reviewed changes

simonjbeaumont reviewed Jul 9, 2024

View reviewed changes

...orithms/MultiProducerSingleConsumerChannel/MultiProducerSingleConsumerChannel+Internal.swift Outdated Show resolved Hide resolved

.../AsyncAlgorithms/MultiProducerSingleConsumerChannel/MultiProducerSingleConsumerChannel.swift Outdated Show resolved Hide resolved

simonjbeaumont reviewed Jul 9, 2024

View reviewed changes

ser-0xff reviewed Oct 23, 2024

View reviewed changes

.../AsyncAlgorithms/MultiProducerSingleConsumerChannel/MultiProducerSingleConsumerChannel.swift Outdated Show resolved Hide resolved

phausler reviewed Mar 24, 2025

View reviewed changes

FranzBusch force-pushed the fb-async-backpressured-stream branch from 662693e to 944b727 Compare March 25, 2025 14:00

FranzBusch added 8 commits March 29, 2025 12:56

Update proposal and implementation

4d7b271

Update proposal

d0eef05

Add example project

7b7c359

Formatting

712bc12

Fix Swift 6.0 build

baf72ef

Future direction for ~Copyable elements

c9b9c81

Apply formatting

0ae6092

FranzBusch force-pushed the fb-async-backpressured-stream branch 2 times, most recently from f086eda to 216e38f Compare March 29, 2025 12:15

FranzBusch force-pushed the fb-async-backpressured-stream branch from 216e38f to ae3f677 Compare March 29, 2025 13:26

Fix CI

b30ec17

FranzBusch force-pushed the fb-async-backpressured-stream branch from ae3f677 to b30ec17 Compare March 29, 2025 13:30

FranzBusch added 2 commits March 29, 2025 15:00

Move to 6.1 and update proposal

9372179

Guard tests

8d2f4d6

FranzBusch force-pushed the fb-async-backpressured-stream branch from 6b24e93 to 8d2f4d6 Compare March 29, 2025 14:07

FranzBusch added 3 commits March 29, 2025 15:30

Minor edits to the proposal

357e9ef

Fix revision order

c3911f1

FIxup setOnTerminationCallback

02b6c86

jamieQ reviewed Mar 30, 2025

View reviewed changes

FranzBusch added 3 commits April 2, 2025 13:53

Address review feedback

3523cd3

Rename to MultiProducerSingleConsumerAsyncChannel

0b39d1e

Allow one termination callback per source.

417b947

FranzBusch mentioned this pull request May 23, 2025

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. swiftlang/swift-subprocess#40

Open

iCharlesHu mentioned this pull request May 30, 2025

Investigate AsyncStream alternative for output streaming swiftlang/swift-subprocess#51

Closed

phausler reviewed Aug 6, 2025

View reviewed changes


		### Upstream producer termination

		The producer will get notified about termination through the `onTerminate`

Add MultiProducerSingleConsumerChannel #305

Are you sure you want to change the base?

Add MultiProducerSingleConsumerChannel #305

Conversation

FranzBusch commented Dec 19, 2023

Motivation

Modification

Uh oh!

FranzBusch commented Dec 20, 2023

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

ser-0xff commented Oct 23, 2024

Uh oh!

FranzBusch commented Oct 23, 2024

Uh oh!

ser-0xff commented Oct 24, 2024

Uh oh!

hassila commented Oct 24, 2024

Uh oh!

ser-0xff commented Oct 24, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

FranzBusch commented Oct 28, 2024

Uh oh!

ser-0xff commented Oct 28, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ser-0xff commented Nov 5, 2024

Uh oh!

FranzBusch commented Nov 11, 2024

Uh oh!

phausler left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jamieQ Mar 30, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

phausler left a comment

Choose a reason for hiding this comment

Uh oh!

Add `MultiProducerSingleConsumerChannel` #305

Add `MultiProducerSingleConsumerChannel` #305

ser-0xff commented Oct 24, 2024 •

edited

Loading

ser-0xff commented Oct 28, 2024 •

edited

Loading

jamieQ Mar 30, 2025 •

edited

Loading

phausler Aug 7, 2025 •

edited

Loading