channel: update documentation for cancellation

Author: twittemb

Package.swift

@@ -16,9 +16,12 @@ 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", .upToNextMajor(from: "1.0.3"))],
   targets: [
-    .target(name: "AsyncAlgorithms"),
+    .target(
+      name: "AsyncAlgorithms",
+      dependencies: [.product(name: "Collections", package: "swift-collections")]
+    ),
     .target(
       name: "AsyncSequenceValidation",
       dependencies: ["_CAsyncSequenceValidationSupport", "AsyncAlgorithms"]),

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

@@ -51,7 +51,7 @@ public final class AsyncThrowingChannel<Element: Sendable, Failure: Error>: Asyn
 }
 ```
 
-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(_:)` via the suspension/resume ensures that the production of values does not exceed the consumption of values from iteration. This method suspends after enqueuing the event and is resumed when the next call to `next()` on the `Iterator` is made. On the other hand, the call to `finish()` or `fail(_:)` 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, or by throwing an error, indicating the termination of the iterations. Further calls to `send(_:)` will immediately resume.
+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(_:)` via the suspension/resume ensures that the production of values does not exceed the consumption of values from iteration. This method suspends after enqueuing the event and is resumed when the next call to `next()` on the `Iterator` is made. On the other hand, the call to `finish()` or `fail(_:)` 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, or by throwing an error, indicating the termination of the iterations. Further calls to `send(_:)` will immediately resume. The calls to `send(:)` and `next()` will immediately resume when their supporting task is cancelled, other operations from other tasks will remain active.
 
 ```swift
 let channel = AsyncChannel<String>()

Sources/AsyncAlgorithms/AsyncChannel.swift

@@ -9,6 +9,8 @@
 //
 //===----------------------------------------------------------------------===//
 
+import OrderedCollections
+
 /// A channel for sending elements from one task to another with back pressure.
 ///
 /// The `AsyncChannel` class is intended to be used as a communication type between tasks,
@@ -86,8 +88,8 @@ public final class AsyncChannel<Element: Sendable>: AsyncSequence, Sendable {
 
   enum Emission {
     case idle
-    case pending(Set<Pending>)
-    case awaiting(Set<Awaiting>)
+    case pending(OrderedSet<Pending>)
+    case awaiting(OrderedSet<Awaiting>)
     case finished
   }
 
@@ -158,7 +160,7 @@ public final class AsyncChannel<Element: Sendable>: AsyncSequence, Sendable {
           }
           return UnsafeResumption(continuation: send.continuation, success: continuation)
         case .awaiting(var nexts):
-          nexts.update(with: Awaiting(generation: generation, continuation: continuation))
+          nexts.updateOrAppend(Awaiting(generation: generation, continuation: continuation))
           state.emission = .awaiting(nexts)
           return nil
         case .finished:
@@ -213,7 +215,7 @@ public final class AsyncChannel<Element: Sendable>: AsyncSequence, Sendable {
           state.emission = .pending([Pending(generation: generation, continuation: continuation)])
           return nil
         case .pending(var sends):
-          sends.update(with: Pending(generation: generation, continuation: continuation))
+          sends.updateOrAppend(Pending(generation: generation, continuation: continuation))
           state.emission = .pending(sends)
           return nil
         case .awaiting(var nexts):
@@ -235,6 +237,7 @@ public final class AsyncChannel<Element: Sendable>: AsyncSequence, Sendable {
   /// Send an element to an awaiting iteration. This function will resume when the next call to `next()` is made
   /// or when a call to `finish()` is made from another Task.
   /// If the channel is already finished then this returns immediately
+  /// If the task is cancelled, this function will resume. Other sending operations from other tasks will remain active.
   public func send(_ element: Element) async {
     let generation = establish()
     let sendTokenStatus = ManagedCriticalState<ChannelTokenStatus>(.new)
@@ -249,8 +252,8 @@ public final class AsyncChannel<Element: Sendable>: AsyncSequence, Sendable {
   /// Send a finish to all awaiting iterations.
   /// All subsequent calls to `next(_:)` will resume immediately.
   public func finish() {
-    let (sends, nexts) = state.withCriticalRegion { state -> (Set<Pending>, Set<Awaiting>) in
-      let result: (Set<Pending>, Set<Awaiting>)
+    let (sends, nexts) = state.withCriticalRegion { state -> (OrderedSet<Pending>, OrderedSet<Awaiting>) in
+      let result: (OrderedSet<Pending>, OrderedSet<Awaiting>)
 
       switch state.emission {
       case .idle:

Sources/AsyncAlgorithms/AsyncThrowingChannel.swift

@@ -9,6 +9,8 @@
 //
 //===----------------------------------------------------------------------===//
 
+import OrderedCollections
+
 /// An error-throwing channel for sending elements from on task to another with back pressure.
 ///
 /// The `AsyncThrowingChannel` class is intended to be used as a communication types between tasks,
@@ -95,8 +97,8 @@ public final class AsyncThrowingChannel<Element: Sendable, Failure: Error>: Asyn
 
   enum Emission {
     case idle
-    case pending(Set<Pending>)
-    case awaiting(Set<Awaiting>)
+    case pending(OrderedSet<Pending>)
+    case awaiting(OrderedSet<Awaiting>)
     case terminated(Termination)
   }
 
@@ -167,7 +169,7 @@ public final class AsyncThrowingChannel<Element: Sendable, Failure: Error>: Asyn
           }
           return UnsafeResumption(continuation: send.continuation, success: continuation)
         case .awaiting(var nexts):
-          nexts.update(with: Awaiting(generation: generation, continuation: continuation))
+          nexts.updateOrAppend(Awaiting(generation: generation, continuation: continuation))
           state.emission = .awaiting(nexts)
           return nil
         case .terminated(let termination):
@@ -235,7 +237,7 @@ public final class AsyncThrowingChannel<Element: Sendable, Failure: Error>: Asyn
           state.emission = .pending([Pending(generation: generation, continuation: continuation)])
           return nil
         case .pending(var sends):
-          sends.update(with: Pending(generation: generation, continuation: continuation))
+          sends.updateOrAppend(Pending(generation: generation, continuation: continuation))
           state.emission = .pending(sends)
           return nil
         case .awaiting(var nexts):
@@ -255,7 +257,7 @@ public final class AsyncThrowingChannel<Element: Sendable, Failure: Error>: Asyn
   }
 
   func terminateAll(error: Failure? = nil) {
-    let (sends, nexts) = state.withCriticalRegion { state -> (Set<Pending>, Set<Awaiting>) in
+    let (sends, nexts) = state.withCriticalRegion { state -> (OrderedSet<Pending>, OrderedSet<Awaiting>) in
 
       let nextState: Emission
       if let error = error {
@@ -297,6 +299,7 @@ public final class AsyncThrowingChannel<Element: Sendable, Failure: Error>: Asyn
   /// Send an element to an awaiting iteration. This function will resume when the next call to `next()` is made
   /// or when a call to `finish()`/`fail(_:)` is made from another Task.
   /// If the channel is already finished then this returns immediately
+  /// If the task is cancelled, this function will resume. Other sending operations from other tasks will remain active.
   public func send(_ element: Element) async {
     let generation = establish()
     let sendTokenStatus = ManagedCriticalState<ChannelTokenStatus>(.new)