`throttle` does not emit the latest element at interval occurrence when latest is `true`

[tarbaiev-smg]

Unlike similar operator in Combine the throttle's latest parameter is confusing, as it does not emit any latest, but a next element instead.

When latest parameter is set to true

Actual behavior:

The throttle emits the next element from the base sequence, if any, after the interval occurs.
Meaning if there's no new elements emitted after the interval occurrence, the latest element is just skipped, leading to only the first element being emitted.

Expected behavior:

The throttle emits the latest cached element instantly when the interval occurs.

[notcome]

I think the semantics is very unclear on this one. Let's say we publish 0...99 every 100ms with a throttling window of 1 sec.

• When latest = false we have 0, 1, 11, 21, ..., 81.
• When latest = true we have 0, 10, 20, 30, ..., 90.

That feels very strange. I agree that at least there should be an option to produce the last element, if this element was initially skipped in the throttling window. But is there a clear semantics for this concept?

[tarbaiev-smg]

@notcome given the Combine and Swift Concurrency interoperability and switching from Publisher to AsyncSequence, I think AsyncAlgorithms's behavior should be consistent with Combine as much as possible.
I had to switch to the Combine's throttle because of this flaw. If I had an AsyncSequence instead of a Publisher, I would need to convert it to a Publisher, apply throttle and convert back to AsyncSequence, which is annoying and inefficient.

[notcome]

@tarbaiev-smg So I did a little search on RxSwift, and it seems that something as simple as throttle actually has lots of subtle variations.

Anyway, it seems that adding a clock task, which is required to publish the latest element after signal stabilizing, is very tricky. The current throttle implementation has less than a hundred lines of code, whereas bounce spans near 2k LOC. Stunning…

I can't use Combine here because I am moving to concurrency infras for testing. So I end up focusing on my particular case — infinite demand, no error handling — and get away with a very simple actor. It's quite tricky though, Task.init captures your actor without warning (you don't need to write explicit self) and gets cancellation right is hard.

[Kolos65]

One would indeed expect to receive the last element when using throttle. A very common use case for example is to throttle a sequence of progress values where receiving the last element (i.e. 100%) is crucial.

In this example:

var numbers = Array((0...100).reversed())
let stream = AsyncStream {
    try! await Task.sleep(nanoseconds: 50_000_000)
    return numbers.popLast()
}

Task {
    let sequence = stream
        .throttle(for: .seconds(1))
    for try await num in sequence {
        print(num)
    }
}

The output is:

0
19
38
58
77
96

Which is unexpected as the latest element is 100.

The problem is that in the AsyncThrottleSequence implementation we can wait for the next element longer then interval which can result in the latest element being 'stuck'. Wouldn't it make sense to add a timeout when awaiting the base sequence's next() so that the latest element is always emitted after interval amount of time passed?

public mutating func next() async rethrows -> Reduced? {
    var reduced: Reduced?
    let start = last ?? clock.now
    repeat {
        // Return reduced if we wait longer then interval (and reduced was not returned previously)
        guard let element = try await base.next() else {
            return nil
        }
        let reduction = await reducing(reduced, element)
        let now = clock.now
        if start.duration(to: now) >= interval || last == nil {
            last = now
            return reduction
        } else {
            reduced = reduction
        }
    } while true
}

[kdubb]

I also find throttle's implementation confusing. In addition to the behavior of 'latest', given infinite demand, the loop consumes available processing. For example an AsyncSequence that is simply a counter returning immediately the next iteration's value.

Maybe I am expecting something I shouldn't but throttling by rapidly consuming a stream's elements, only to discard them, seems incorrect. Throttling should handle the case of over and under producing streams without a severe performance penalty in either case.

[kdubb]

Just as a use case, I attempted to use throttle to sample the audio level from an AVAudioRecorder; which immediately returns the current value. During a simple test I realized it was consuming values at an unchecked pace.

[FranzBusch]

@kdubb I think that is the expected behaviour from throttle though. It will apply infinite demand as long as it has demand on itself. If you have a root asynchronous sequence that produces infinite values instantly then throttle will produce exactly one value at every interval.

I think you want another algorithm for what you describe here.

[kdubb]

The issue for me is that I was switching from a slow producing stream to an infinite producing stream and you need something that can deal with both cases. I attempted to use timer sequence and withLatest (from AsyncExtensnions) but that has the same issue with infinite streams (it loops in a similar way to throttle).

Finally, I flatMapd the timer sequence to the infinite stream to get a reasonable implementation but this all required knowing exactly the production timing of each stream.

To me the name throttle is expressly saying that it is reducing the production of the source stream; but currently it's not. If that just not how it's ever going to work then you are correct we need another algorithm.

Maybe we create one called 'polling' that uses a Clock to sleep until it acquires the next element from the source stream; and if calling next takes too long it immediately pulls. But if we have polling that works as I described, what use would there be for throttle?

[kdubb]

Also, adding a doc - Note: to this, and any other algorithms that will runaway with an infinitely supplying stream, would help users with their expectation and avoid unexpected performance issues when using these algorithms.

[FranzBusch]

@phausler just put up a PR #292 to change the behaviour. We would appreciate if you all could check that it is doing what you expected it to do now.

[tarbaiev-smg]

@FranzBusch Thanks for the update! However the PR does not seem to be addressing this issue. The AsyncThrottleSequence.next() is still awaiting the next element of its base sequence even after the interval ends. This means it won't emit an actual last element of the last interval, but will emit the first element of the subsequent interval instead.

[Kolos65]

Yes, while the PR addresses issues in cases where the sequence finishes (by returning nil), but as @tarbaiev-smg said, we can still await for the next element longer than interval which means that the last element wont be emitted.

Here is an updated example of the issue that uses a sequence that waits for the next element longer and is not finishing:

var numbers = Array((0...200).reversed())
let stream = AsyncStream<Int> {
   guard let number = numbers.popLast() else { return nil }
   if number == 101 {
        try! await Task.sleep(nanoseconds: 1_000_000_000_000)
    } else {
        try! await Task.sleep(nanoseconds: 50_000_000)
    }
    return number
}

Task {
    let sequence = stream
        .throttle(for: .seconds(1))
    for try await num in sequence {
        print(num)
    }
}

prints:

0
20
39
58
78
97
<long break>
<remaining numbers>

instead of:

0
20
39
58
78
97
100
<long break>
<remaining numbers>

[FranzBusch]

@Kolos65 Thanks for the example. Could you do me a favour an try to write a validation test that represents what currently is not working like you expect it to work. Validation tests look like this

    validate {
      "abcdefghijk|"
      $0.inputs[0].throttle(for: .steps(3), clock: $0.clock, latest: false)
      "a--b--e--h--[k|]"
    }

I am just trying to nail down if we have a problem and what it exactly is and that we are not mixing semantics of debounce with throttle here.