Fix async channel finishes all (#152)

* asyncChannel: enforce termination for all when finished

* asyncThrowingChannel: enforce termination for all when finished

* asyncChannel: improve unit tests readability

* asyncChannel: adapt the Guide to the new finish behaviour

* asyncChannel: adapt the Guide to the new finish behaviour

Author: twittemb

Sources/AsyncAlgorithms/AsyncAlgorithms.docc/Guides/Channel.md

@@ -14,7 +14,7 @@
 
 ## Proposed Solution
 
-To achieve a system that supports back pressure and allows for the communication of more than one value from one task to another we are introducing a new type, the _channel_. The channel will be a reference-type asynchronous sequence with an asynchronous sending capability that awaits the consumption of iteration. Each value sent by the channel, or finish transmitted, will await the consumption of that value or event by iteration. That awaiting behavior will allow for the affordance of back pressure applied from the consumption site to be transmitted to the production site. This means that the rate of production cannot exceed the rate of consumption, and that the rate of consumption cannot exceed the rate of production.
+To achieve a system that supports back pressure and allows for the communication of more than one value from one task to another we are introducing a new type, the _channel_. The channel will be a reference-type asynchronous sequence with an asynchronous sending capability that awaits the consumption of iteration. Each value sent by the channel will await the consumption of that value by iteration. That awaiting behavior will allow for the affordance of back pressure applied from the consumption site to be transmitted to the production site. This means that the rate of production cannot exceed the rate of consumption, and that the rate of consumption cannot exceed the rate of production. Sending a terminal event to the channel will instantly resume all pending operations for every producers and consumers.
 
 ## Detailed Design
 
@@ -31,7 +31,7 @@ public final class AsyncChannel<Element: Sendable>: AsyncSequence, Sendable {
   public init(element elementType: Element.Type = Element.self)
 
   public func send(_ element: Element) async
-  public func finish() async
+  public func finish()
 
   public func makeAsyncIterator() -> Iterator
 }
@@ -45,21 +45,21 @@ public final class AsyncThrowingChannel<Element: Sendable, Failure: Error>: Asyn
 
   public func send(_ element: Element) async
   public func fail(_ error: Error) async where Failure == Error
-  public func finish() async
+  public func finish()
 
   public func makeAsyncIterator() -> Iterator
 }
 ```
 
-Channels are intended to be used as communication types between tasks. Particularly when one task produces values and another task consumes said values. The back pressure applied by `send(_:)`, `fail(_:)` and `finish()` via the suspension/resume ensure that the production of values does not exceed the consumption of values from iteration. Each of these methods suspend after enqueuing the event and are resumed when the next call to `next()` on the `Iterator` is made. 
+Channels are intended to be used as communication types between tasks. Particularly when one task produces values and another task consumes said values. On the one hand, the back pressure applied by `send(_:)` and `fail(_:)` via the suspension/resume ensure that the production of values does not exceed the consumption of values from iteration. Each of these methods suspend after enqueuing the event and are resumed when the next call to `next()` on the `Iterator` is made. On the other hand, the call to `finish()` immediately resumes all the pending operations for every producers and consumers. Thus, every suspended `send(_:)` operations instantly resume, so as every suspended `next()` operations by producing a nil value, indicating the termination of the iterations. Further calls to `send(_:)` will immediately resume.
 
 ```swift
 let channel = AsyncChannel<String>()
 Task {
   while let resultOfLongCalculation = doLongCalculations() {
     await channel.send(resultOfLongCalculation)
   }
-  await channel.finish()
+  channel.finish()
 }
 
 for await calculationResult in channel {

Sources/AsyncAlgorithms/AsyncChannel.swift

@@ -131,7 +131,12 @@ public final class AsyncChannel<Element: Sendable>: AsyncSequence, Sendable {
   func next(_ generation: Int) async -> Element? {
     return await withUnsafeContinuation { continuation in
       var cancelled = false
+      var terminal = false
       state.withCriticalRegion { state -> UnsafeResumption<UnsafeContinuation<Element?, Never>?, Never>? in
+        if state.terminal {
+          terminal = true
+          return nil
+        }
         switch state.emission {
         case .idle:
           state.emission = .awaiting([Awaiting(generation: generation, continuation: continuation)])
@@ -157,13 +162,13 @@ public final class AsyncChannel<Element: Sendable>: AsyncSequence, Sendable {
           return nil
         }
       }?.resume()
-      if cancelled {
+      if cancelled || terminal {
         continuation.resume(returning: nil)
       }
     }
   }
 
-  func cancelSend() {
+  func finishAll() {
     let (sends, nexts) = state.withCriticalRegion { state -> ([UnsafeContinuation<UnsafeContinuation<Element?, Never>?, Never>], Set<Awaiting>) in
       if state.terminal {
         return ([], [])
@@ -188,23 +193,15 @@ public final class AsyncChannel<Element: Sendable>: AsyncSequence, Sendable {
     }
   }
 
-  func _send(_ result: Result<Element?, Never>) async {
+  func _send(_ element: Element) async {
     await withTaskCancellationHandler {
-      cancelSend()
+      finishAll()
     } operation: {
       let continuation: UnsafeContinuation<Element?, Never>? = await withUnsafeContinuation { continuation in
         state.withCriticalRegion { state -> UnsafeResumption<UnsafeContinuation<Element?, Never>?, Never>? in
           if state.terminal {
             return UnsafeResumption(continuation: continuation, success: nil)
           }
-          switch result {
-          case .success(let value):
-            if value == nil {
-              state.terminal = true
-            }
-          case .failure:
-            state.terminal = true
-          }
           switch state.emission {
           case .idle:
             state.emission = .pending([continuation])
@@ -224,20 +221,19 @@ public final class AsyncChannel<Element: Sendable>: AsyncSequence, Sendable {
           }
         }?.resume()
       }
-      continuation?.resume(with: result)
+      continuation?.resume(returning: element)
     }
   }
 
   /// Send an element to an awaiting iteration. This function will resume when the next call to `next()` is made.
   /// If the channel is already finished then this returns immediately
   public func send(_ element: Element) async {
-      await _send(.success(element))
+    await _send(element)
   }
 
-  /// Send a finish to an awaiting iteration. This function will resume when the next call to `next()` is made.
-  /// If the channel is already finished then this returns immediately
-  public func finish() async {
-      await _send(.success(nil))
+  /// Send a finish to all awaiting iterations.
+  public func finish() {
+    finishAll()
   }
 
   /// Create an `Iterator` for iteration of an `AsyncChannel`

Sources/AsyncAlgorithms/AsyncThrowingChannel.swift

@@ -129,7 +129,12 @@ public final class AsyncThrowingChannel<Element: Sendable, Failure: Error>: Asyn
   func next(_ generation: Int) async throws -> Element? {
     return try await withUnsafeThrowingContinuation { continuation in
       var cancelled = false
+      var terminal = false
       state.withCriticalRegion { state -> UnsafeResumption<UnsafeContinuation<Element?, Error>?, Never>? in
+        if state.terminal {
+          terminal = true
+          return nil
+        }
         switch state.emission {
         case .idle:
           state.emission = .awaiting([Awaiting(generation: generation, continuation: continuation)])
@@ -155,13 +160,13 @@ public final class AsyncThrowingChannel<Element: Sendable, Failure: Error>: Asyn
           return nil
         }
       }?.resume()
-      if cancelled {
+      if cancelled || terminal {
         continuation.resume(returning: nil)
       }
     }
   }
 
-  func cancelSend() {
+  func finishAll() {
     let (sends, nexts) = state.withCriticalRegion { state -> ([UnsafeContinuation<UnsafeContinuation<Element?, Error>?, Never>], Set<Awaiting>) in
       if state.terminal {
         return ([], [])
@@ -186,23 +191,20 @@ public final class AsyncThrowingChannel<Element: Sendable, Failure: Error>: Asyn
     }
   }
 
-  func _send(_ result: Result<Element?, Error>) async {
+  func _send(_ result: Result<Element, Error>) async {
     await withTaskCancellationHandler {
-      cancelSend()
+      finishAll()
     } operation: {
       let continuation: UnsafeContinuation<Element?, Error>? = await withUnsafeContinuation { continuation in
         state.withCriticalRegion { state -> UnsafeResumption<UnsafeContinuation<Element?, Error>?, Never>? in
           if state.terminal {
             return UnsafeResumption(continuation: continuation, success: nil)
           }
-          switch result {
-          case .success(let value):
-            if value == nil {
-              state.terminal = true
-            }
-          case .failure:
+
+          if case .failure = result {
             state.terminal = true
           }
+          
           switch state.emission {
           case .idle:
             state.emission = .pending([continuation])
@@ -222,7 +224,7 @@ public final class AsyncThrowingChannel<Element: Sendable, Failure: Error>: Asyn
           }
         }?.resume()
       }
-      continuation?.resume(with: result)
+      continuation?.resume(with: result.map { $0 as Element? })
     }
   }
 
@@ -238,10 +240,9 @@ public final class AsyncThrowingChannel<Element: Sendable, Failure: Error>: Asyn
     await _send(.failure(error))
   }
 
-  /// Send a finish to an awaiting iteration. This function will resume when the next call to `next()` is made.
-  /// If the channel is already finished then this returns immediately
-  public func finish() async {
-      await _send(.success(nil))
+  /// Send a finish to all awaiting iterations.
+  public func finish() {
+    finishAll()
   }
 
   public func makeAsyncIterator() -> Iterator {