Remove `AsyncIterator: Sendable` requirement from merge #185

FranzBusch · 2022-08-10T11:55:06Z

Motivation

Currently a lot of the operator implementations in here that consume other AsyncSequences require the AsyncIterator to be Sendable. This is mostly due to the fact that we are calling makeAsyncIterator on the upstream AsyncSequence and then pass that iterator around to various newly spawned Tasks. This has two downsides:

It only allows users to use operators like merge if their AsyncSequence.AsyncIterator is Sendable
In merge we are creating new Tasks for every new demand. Creating Tasks is not cheap.

My main goal of this PR was to remove the Sendable constraint from merge.

Modification

This PR overhauls the complete inner workings of the AsyncMerge2Sequence. It does a couple of things:

The main change is that instead of creating new Tasks for every demand, we are creating one Task when the AsyncIterator is created. This task has as child task for every upstream sequence.
When calling next we are signalling the child tasks to demand from the upstream
A new state machine that is synchronizing the various concurrent operations that can happen
Handling cancellation since we are creating a bunch of continuations.

Result

In the end, this PR swaps the implementation of AsyncMerge2Sequence and drops the Sendable constraint and passes all tests. Furthermore, on my local performance testing I saw up 50% speed increase in throughput.

Open points

I need to make this sequence re-throwing but before going down that rabbit whole I wanna get buy-in on the implementation.
We should discuss and document if merge and other operators are hot or cold, i.e. if they only request if they got downstream demand
I need to switch AsyncMerge3Sequence over to the same iplementation

FranzBusch · 2022-08-10T11:59:55Z

Package.swift

@@ -16,9 +16,15 @@ let package = Package(
    .library(name: "_CAsyncSequenceValidationSupport", type: .static, targets: ["AsyncSequenceValidation"]),
    .library(name: "AsyncAlgorithms_XCTest", targets: ["AsyncAlgorithms_XCTest"]),
  ],
-  dependencies: [],
+  dependencies: [
+    .package(url: "https://github.com/apple/swift-collections.git", from: "1.0.2"),


I was unsure if we should add this dependency. It makes the implementation a lot better and IMO is a lower level package. Open to discuss this :)

FranzBusch · 2022-08-10T12:19:59Z

Switching this from CheckedContinuation to UnsafeContinuation is improving the performance by a lot:

Current main: 1x baseline
This PR with Checked: 1.5x
This PR with Unsafe: 3.5x

phausler · 2022-08-10T13:57:12Z

I'm out on vacation this week, but a few key points - merge (like all other algorithms) should only request the next value upon need/demand, I had a similar but stale branch of collapsing the tasks to gain some perf but the problem became the collapse versus the rethrow (long/short of it was that the pseudo cancellation of a throw put a gnarly wrench in the works for the collapse part). All that being said; I think it can be done - just hadn't gotten around to it just yet, there are a number of related algorithms that are effectively a composition of merge that we could improve so this is definitely something we should dig into.

FranzBusch · 2022-08-10T15:26:31Z

@phausler Yeah no rush on this! Enjoy your vacation and we can talk afterwards. I just wanted to put this up for discussion. I agree that re throwing is kinda coming the way but I am seeing up to 3.5 times performance improvements with this. However, performance is no the main driver, it really is the Sendable annotation. Let's discuss this once you are back!

phausler

Overall this looks good, it could use some sprucing up with regards to the locks before merging (id rather not introduce a new locking scheme and keep to the managed critical state stuff if at all possible.

It took me a bit to follow all of the logic but it seems relatively sound (excluding the whole fatal error cases marked in the review).

With changing to the unchecked variants I think we can eek out some more perf!

Sources/AsyncAlgorithms/Locking.swift

phausler · 2022-08-17T20:13:56Z

Sources/AsyncAlgorithms/Merge2StateMachine.swift

+
+  /// Advances to the next element and returns it or `nil` if no next element exists.
+  mutating func next() async rethrows -> Either? {
+    if Task.isCancelled {


I wonder if this his sufficient; shouldn't we have a reaction to the cancellation while it is awaiting?

Consider a merge of two very long async ops: if a cancel happens we should immediately be able to discern that the resultant is nil.

This is the old code. I just copied this since other algos were using this and I did not want to change them.

phausler · 2022-08-17T20:15:37Z

Sources/AsyncAlgorithms/AsyncMerge2Sequence.swift

+private extension AsyncMerge2Sequence {
+    final class Storage: @unchecked Sendable {
+        /// The lock that protects our state.
+        private let lock = Lock.allocate()


this should probably be done with the managed critical state type instead

I tried to adopt this but ManagedCriticalState does only support using a lock with a closure. For the next method I would like to manually lock and unlock to be able to hold the lock across the withUnsafeContinuation method.

Holding a lock across a suspension is almost always a really bad idea; it can cause some pretty heinous fall out.

Holding a lock across with*Continuation is safe to do since the closure is executed right away. We have been doing this in various places of the NIO ecosystem now.

To further clarify this is not holding a lock across a suspension point but just across the with*Continuation call.

so there is only 1 way to do that safely (and it is a very edge case scenario).

lock() await withUnsafeContinuation { continuation in self.continuation = continuation unlock() }

It is very easy to modify that pattern into some really gnarly failure modes. From a maintainability standpoint id rather make sure there is no other way of doing that.

Right I am using exactly that pattern just with calls to the state inbetween. I agree with you that this needs to be done with care.
However, if don’t hold the lock across that call, the state handling becomes trickier since it introduces a potential interleaving between next and closure of withUnsafeContinuation where everything else can happen

phausler · 2022-08-30T17:20:04Z

Sources/AsyncAlgorithms/Merge2StateMachine.swift

+  }
+
+  mutating func apply(_ task1: Task<Merge2StateMachine<Base1, Base2>.Partial, Never>?, _ task2: Task<Merge2StateMachine<Base1, Base2>.Partial, Never>?) async rethrows -> Either? {
+    switch await Task.select([task1, task2].compactMap ({ $0 })).value {


do you think there is a way we can approach this without Task.select? That is the other perf bottle neck (which could be either solved by not needing it, or solved by improving it at a language level)

Same here, not my code just copied over.

Sources/AsyncAlgorithms/AsyncMerge2Sequence.swift

phausler · 2022-08-30T17:28:01Z

Sources/AsyncAlgorithms/AsyncMerge2Sequence.swift

+        /// Actions returned by `childTaskSuspended()`.
+        enum ChildTaskSuspendedAction {
+            /// Indicates that the continuation should be resumed which will lead to calling `next` on the upstream.
+            case resumeContinuation(


stylistic nit; most folks have wide enough monitors to see the comment, so no need to limit code to 40 col... (cause if you are editing swift on a non emulated VT100... well time to upgrade :) )

I am using newlines here mostly to make reading it easier not because of screen width. I would prefer to keep that to maintain readability.

phausler · 2022-08-30T17:28:52Z

Sources/AsyncAlgorithms/AsyncMerge2Sequence.swift

+            case let .merging(task, buffer, upstreamContinuations, upstreamsFinished, .some(downstreamContinuation)):
+                // We produced an element and have an outstanding downstream continuation
+                // this means we can go right ahead and resume the continuation with that element
+                precondition(buffer.isEmpty, "We are holding a continuation so the buffer must be empty")


this feels like it somehow could not be fully true... do we have tests that validate this?

Various tests are validating that indirectly. For this precondition to fail we would need to get into a state where we have a suspended call to next() and things in the buffer. However, when we call next and have things in the buffer we are returning them right away. We are only suspending if the buffer is empty. Furthermore, since we are holding the lock across the withUnsafeContinuation call in next nothing can get in-between us checking if the buffer is empty and us suspending.

When any of the upstreams produces a new element, it will acquire the lock and then we come into this case. If we have a continuation we are guaranteed to have an empty buffer since we would have never suspended in the first place if we had something in the buffer.

Sources/AsyncAlgorithms/AsyncMerge2Sequence.swift

phausler · 2022-08-30T17:30:27Z

Sources/AsyncAlgorithms/AsyncMerge2Sequence.swift

+            )
+            /// Indicates that the downstream continuation should be resumed with `nil` and
+            /// the task and the upstream continuations should be cancelled.
+            case resumeContinuationWithNilAndCancelTaskAndUpstreamContinuations(


some of these long names and numerous associated types kinda almost infer structness of the associations. Would it be better to change them to structures instead of effectively tuples?

In general I am with you; however, this would introduce just a struct for one case of an action. My personal opinion, is that this is overkill for something internal. If you this is important to you, I am happy to change it though. Your call :)

Sources/AsyncAlgorithms/AsyncMerge2Sequence.swift

phausler · 2022-08-30T17:42:35Z

One required item for merging this: it needs to definitely handle rethrowing cases (and we should validate that is true)

FranzBusch · 2022-09-09T14:32:30Z

@phausler I just updated the PR again. To summarise what changed from your initial review:

My implementation supports rethrowing now
I split up the implementation into separate files to aid with readability
I swapped merge3 to also make use of the new state machine
I renamed the internal upstreamThrew case to upstreamFailure
I added more tests validating various edge cases
Aligned precondition messages

I replied to some of your feedback inline but I haven't changed the following:

Use ManagedCriticalState: I can't adopt this since I need to be able to manually lock and unlock in the case when I need to suspend on next()
Remove preconditions: I went through the code again and checked these preconditions and I wasn't able to come up with a test to break them. There might still be an edge case I haven't thought about, but I would rather let this crash then to transition into an unknown state.
Line breaks: I find them easier to read but if you want me to use a single line I am happy to change (maybe it would be good to get swift format setup in this repo?)

# Motivation The current implementation of `AsyncDebounceSequence` requires the base `AsyncIterator` to be `Sendable`. This is causing two problems: 1. It only allows users to use `debounce` if their `AsyncSequence.AsyncIterator` is `Sendable` 2. In `debounce` we are creating a lot of new `Task`s and reating `Task`s is not cheap. My main goal of this PR was to remove the `Sendable` constraint from `debounce`. # Modification This PR overhauls the implementation of `debounce` and aligns it with the implementation of the open `merge` PR apple#185 . The most important changes are this: - I removed the `Sendable` requirement from the base sequences `AsyncIterator`. - Instead of creating new Tasks for the sleep and for the upstream consumption. I am now creating one Task and manipulate it by signalling continuations - I am not cancelling the sleep. Instead I am recalculating the time left to sleep when a sleep finishes. # Result In the end, this PR swaps the implementation of `AsyncDebounceSequence` and drops the `Sendable` constraint and passes all tests. Furthermore, on my local performance testing I saw up 150% speed increase in throughput.

phausler · 2022-09-13T20:09:35Z

@swift-ci please test

* Remove `AsyncIterator: Sendable` requirement from debounce # Motivation The current implementation of `AsyncDebounceSequence` requires the base `AsyncIterator` to be `Sendable`. This is causing two problems: 1. It only allows users to use `debounce` if their `AsyncSequence.AsyncIterator` is `Sendable` 2. In `debounce` we are creating a lot of new `Task`s and reating `Task`s is not cheap. My main goal of this PR was to remove the `Sendable` constraint from `debounce`. # Modification This PR overhauls the implementation of `debounce` and aligns it with the implementation of the open `merge` PR #185 . The most important changes are this: - I removed the `Sendable` requirement from the base sequences `AsyncIterator`. - Instead of creating new Tasks for the sleep and for the upstream consumption. I am now creating one Task and manipulate it by signalling continuations - I am not cancelling the sleep. Instead I am recalculating the time left to sleep when a sleep finishes. # Result In the end, this PR swaps the implementation of `AsyncDebounceSequence` and drops the `Sendable` constraint and passes all tests. Furthermore, on my local performance testing I saw up 150% speed increase in throughput. * Fix #174 * Code review * Remove lock methods * Cleanup some unused code * Setup task after first call to next

# Motivation Currently a lot of the operator implementations in here that consume other `AsyncSequence`s require the `AsyncIterator` to be `Sendable`. This is mostly due to the fact that we are calling `makeAsyncIterator` on the upstream `AsyncSequence` and then pass that iterator around to various newly spawned `Task`s. This has two downsides: 1. It only allows users to use operators like `merge` if their `AsyncSequence.AsyncIterator` is `Sendable` 2. In merge we are creating new `Task`s for every new demand. Creating `Task`s is not cheap. My main goal of this PR was to remove the `Sendable` constraint from `merge`. # Modification This PR overhauls the complete inner workings of the `AsyncMerge2Sequence`. It does a couple of things: 1. The main change is that instead of creating new `Task`s for every demand, we are creating one `Task` when the `AsyncIterator` is created. This task has as child task for every upstream sequence. 2. When calling `next` we are signalling the child tasks to demand from the upstream 3. A new state machine that is synchronizing the various concurrent operations that can happen 4. Handling cancellation since we are creating a bunch of continuations. # Result In the end, this PR swaps the implementation of `AsyncMerge2Sequence` and drops the `Sendable` constraint and passes all tests. Furthermore, on my local performance testing I saw up 50% speed increase in throughput. # Open points 1. I need to make this sequence re-throwing but before going down that rabbit whole I wanna get buy-in on the implementation. 2. We should discuss and document if `merge` and other operators are hot or cold, i.e. if they only request if they got downstream demand 3. I need to switch `AsyncMerge3Sequence` over to the same iplementation

…back

FranzBusch · 2022-10-05T13:04:24Z

@phausler Just rebased this PR. Feel free to merge it if you are happy with it!

FranzBusch requested a review from phausler August 10, 2022 11:56

FranzBusch commented Aug 10, 2022

View reviewed changes

FranzBusch force-pushed the fb-merge branch from 9a3f304 to 1706fb3 Compare August 10, 2022 12:11

FranzBusch force-pushed the fb-merge branch from 1706fb3 to 0060af5 Compare August 18, 2022 17:01

phausler approved these changes Aug 30, 2022

View reviewed changes

FranzBusch force-pushed the fb-merge branch from 0060af5 to 535d822 Compare August 31, 2022 13:13

This was referenced Aug 31, 2022

[Channel] improve send cancellation #184

Merged

[Merge] Improve conception #192

Closed

FranzBusch force-pushed the fb-merge branch from 535d822 to 3a1a255 Compare September 9, 2022 14:27

FranzBusch requested a review from phausler September 9, 2022 14:33

This was referenced Sep 11, 2022

Remove AsyncIterator: Sendable requirement from debounce #197

Closed

Remove AsyncIterator: Sendable requirement from debounce #198

Merged

FranzBusch force-pushed the fb-merge branch from d09298c to 747c766 Compare September 29, 2022 13:07

FranzBusch force-pushed the fb-merge branch from 747c766 to 8122e54 Compare October 4, 2022 13:08

FranzBusch added 6 commits October 4, 2022 15:29

Split logic into multiple files, adapt merge3 and incorporate PR feed…

4f992dd

…back

Add more tests and fix upstream throw behaviour

0aa1dd3

Remove internal class from the sequence

4e31506

Setup task after first next

91de406

Remove unnecessary tests

121734a

FranzBusch force-pushed the fb-merge branch from 74289e4 to 121734a Compare October 5, 2022 13:03

phausler merged commit 314b10c into apple:main Oct 10, 2022

FranzBusch deleted the fb-merge branch October 26, 2022 20:29

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Remove `AsyncIterator: Sendable` requirement from merge #185

Remove `AsyncIterator: Sendable` requirement from merge #185

FranzBusch commented Aug 10, 2022

FranzBusch Aug 10, 2022

FranzBusch commented Aug 10, 2022 •

edited

Loading

phausler commented Aug 10, 2022

FranzBusch commented Aug 10, 2022

phausler left a comment

phausler Aug 17, 2022

FranzBusch Aug 31, 2022

phausler Aug 17, 2022

FranzBusch Aug 31, 2022

phausler Sep 9, 2022

FranzBusch Sep 9, 2022

FranzBusch Sep 9, 2022

phausler Sep 9, 2022

FranzBusch Sep 9, 2022

phausler Aug 30, 2022

FranzBusch Aug 31, 2022

phausler Aug 30, 2022

FranzBusch Aug 31, 2022

phausler Aug 30, 2022

FranzBusch Aug 31, 2022

phausler Aug 30, 2022

FranzBusch Aug 31, 2022

phausler commented Aug 30, 2022

FranzBusch commented Sep 9, 2022

phausler commented Sep 13, 2022

FranzBusch commented Oct 5, 2022

Remove AsyncIterator: Sendable requirement from merge #185

Remove AsyncIterator: Sendable requirement from merge #185

Conversation

FranzBusch commented Aug 10, 2022

Motivation

Modification

Result

Open points

Choose a reason for hiding this comment

FranzBusch commented Aug 10, 2022 • edited Loading

phausler commented Aug 10, 2022

FranzBusch commented Aug 10, 2022

phausler left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

phausler commented Aug 30, 2022

FranzBusch commented Sep 9, 2022

phausler commented Sep 13, 2022

FranzBusch commented Oct 5, 2022

Remove `AsyncIterator: Sendable` requirement from merge #185

Remove `AsyncIterator: Sendable` requirement from merge #185

FranzBusch commented Aug 10, 2022 •

edited

Loading