Remove AsyncIterator: Sendable requirement from debounce (apple#198)

* 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 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.

* Fix apple#174

* Code review

* Remove lock methods

* Cleanup some unused code

* Setup task after first call to next

Author: FranzBusch

Sources/AsyncAlgorithms/AsyncDebounceSequence.swift

@@ -0,0 +1,99 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift Async Algorithms open source project
+//
+// Copyright (c) 2022 Apple Inc. and the Swift project authors
+// Licensed under Apache License v2.0 with Runtime Library Exception
+//
+// See https://swift.org/LICENSE.txt for license information
+//
+//===----------------------------------------------------------------------===//
+
+extension AsyncSequence {
+    /// Creates an asynchronous sequence that emits the latest element after a given quiescence period
+    /// has elapsed by using a specified Clock.
+    @available(macOS 13.0, iOS 16.0, watchOS 9.0, tvOS 16.0, *)
+    public func debounce<C: Clock>(for interval: C.Instant.Duration, tolerance: C.Instant.Duration? = nil, clock: C) -> AsyncDebounceSequence<Self, C> where Self: Sendable {
+        AsyncDebounceSequence(self, interval: interval, tolerance: tolerance, clock: clock)
+    }
+
+    /// Creates an asynchronous sequence that emits the latest element after a given quiescence period
+    /// has elapsed.
+    @available(macOS 13.0, iOS 16.0, watchOS 9.0, tvOS 16.0, *)
+    public func debounce(for interval: Duration, tolerance: Duration? = nil) -> AsyncDebounceSequence<Self, ContinuousClock> where Self: Sendable {
+        self.debounce(for: interval, tolerance: tolerance, clock: .continuous)
+    }
+}
+
+/// An `AsyncSequence` that emits the latest element after a given quiescence period
+/// has elapsed.
+@available(macOS 13.0, iOS 16.0, watchOS 9.0, tvOS 16.0, *)
+public struct AsyncDebounceSequence<Base: AsyncSequence, C: Clock>: Sendable where Base: Sendable {
+    private let base: Base
+    private let clock: C
+    private let interval: C.Instant.Duration
+    private let tolerance: C.Instant.Duration?
+
+    /// Initializes a new ``AsyncDebounceSequence``.
+    ///
+    /// - Parameters:
+    ///   - base: The base sequence.
+    ///   - interval: The interval to debounce.
+    ///   - tolerance: The tolerance of the clock.
+    ///   - clock: The clock.
+    public init(_ base: Base, interval: C.Instant.Duration, tolerance: C.Instant.Duration?, clock: C) {
+        self.base = base
+        self.interval = interval
+        self.tolerance = tolerance
+        self.clock = clock
+    }
+}
+
+@available(macOS 13.0, iOS 16.0, watchOS 9.0, tvOS 16.0, *)
+extension AsyncDebounceSequence: AsyncSequence {
+    public typealias Element = Base.Element
+
+    public func makeAsyncIterator() -> AsyncIterator {
+        let storage = DebounceStorage(
+            base: self.base,
+            interval: self.interval,
+            tolerance: self.tolerance,
+            clock: self.clock
+        )
+        return AsyncIterator(storage: storage)
+    }
+}
+
+@available(macOS 13.0, iOS 16.0, watchOS 9.0, tvOS 16.0, *)
+extension AsyncDebounceSequence {
+    public struct AsyncIterator: AsyncIteratorProtocol {
+        /// This class is needed to hook the deinit to observe once all references to the ``AsyncIterator`` are dropped.
+        ///
+        /// If we get move-only types we should be able to drop this class and use the `deinit` of the ``AsyncIterator`` struct itself.
+        final class InternalClass: Sendable {
+            private let storage: DebounceStorage<Base, C>
+
+            fileprivate init(storage: DebounceStorage<Base, C>) {
+                self.storage = storage
+            }
+
+            deinit {
+                self.storage.iteratorDeinitialized()
+            }
+
+            func next() async rethrows -> Element? {
+                try await self.storage.next()
+            }
+        }
+
+        let internalClass: InternalClass
+
+        fileprivate init(storage: DebounceStorage<Base, C>) {
+            self.internalClass = InternalClass(storage: storage)
+        }
+
+        public mutating func next() async rethrows -> Element? {
+            try await self.internalClass.next()
+        }
+    }
+}