use Atomics package instead of Locking where possible

Author: lhoward

Package.resolved

@@ -1,4 +1,4 @@
-// swift-tools-version:5.5
+// swift-tools-version:6.0
 // The swift-tools-version declares the minimum version of Swift required to build this package.
 
 import PackageDescription
@@ -20,19 +20,17 @@ let package = Package(
         .package(url: "https://github.com/apple/swift-collections.git", .upToNextMajor(from: "1.0.3")),
         .package(url: "https://github.com/OpenCombine/OpenCombine.git", from: "0.14.0"),
         .package(url: "https://github.com/apple/swift-async-algorithms.git", .upToNextMajor(from: "1.0.0")),
+        .package(url: "https://github.com/apple/swift-atomics.git", .upToNextMajor(from: "1.2.0")),
     ],
     targets: [
         .target(
             name: "AsyncExtensions",
-            dependencies: [.product(name: "Collections", package: "swift-collections")],
-            path: "Sources"
-//            ,
-//            swiftSettings: [
-//              .unsafeFlags([
-//                "-Xfrontend", "-warn-concurrency",
-//                "-Xfrontend", "-enable-actor-data-race-checks",
-//              ])
-//            ]
+            dependencies: [
+              .product(name: "Collections", package: "swift-collections"),
+              .product(name: "Atomics", package: "swift-atomics")
+            ],
+            path: "Sources",
+            swiftSettings: [.swiftLanguageMode(.v5)]
         ),
         .testTarget(
             name: "AsyncExtensionsTests",
@@ -41,6 +39,8 @@ let package = Package(
                 .product(name: "OpenCombine", package: "OpenCombine", condition: .when(platforms: [.linux])),
                 .product(name: "AsyncAlgorithms", package: "swift-async-algorithms")
             ],
-            path: "Tests"),
+            path: "Tests",
+            swiftSettings: [.swiftLanguageMode(.v5)]
+        ),
     ]
 )

Package.swift

@@ -5,6 +5,7 @@
 //  Created by Thibault Wittemberg on 07/01/2022.
 //
 
+import Atomics
 import DequeModule
 import OrderedCollections
 
@@ -77,19 +78,16 @@ public final class AsyncBufferedChannel<Element>: AsyncSequence, Sendable {
     }
   }
 
-  let ids: ManagedCriticalState<Int>
+  let ids: ManagedAtomic<Int>
   let state: ManagedCriticalState<State>
 
   public init() {
-    self.ids = ManagedCriticalState(0)
+    self.ids = ManagedAtomic(0)
     self.state = ManagedCriticalState(.initial)
   }
 
   func generateId() -> Int {
-    self.ids.withCriticalRegion { ids in
-      ids += 1
-      return ids
-    }
+    ids.wrappingIncrementThenLoad(by: 1, ordering: .relaxed)
   }
 
   var hasBufferedElements: Bool {
@@ -155,12 +153,12 @@ public final class AsyncBufferedChannel<Element>: AsyncSequence, Sendable {
 
   func next(onSuspend: (() -> Void)? = nil) async -> Element? {
     let awaitingId = self.generateId()
-    let cancellation = ManagedCriticalState<Bool>(false)
+    let cancellation = ManagedAtomic<Bool>(false)
 
     return await withTaskCancellationHandler {
       await withCheckedContinuation { [state] (continuation: CheckedContinuation<Element?, Never>) in
         let decision = state.withCriticalRegion { state -> AwaitingDecision in
-          let isCancelled = cancellation.withCriticalRegion { $0 }
+          let isCancelled = cancellation.load(ordering: .acquiring)
           guard !isCancelled else { return .resume(nil) }
 
           switch state {
@@ -200,9 +198,7 @@ public final class AsyncBufferedChannel<Element>: AsyncSequence, Sendable {
       }
     } onCancel: { [state] in
       let awaiting = state.withCriticalRegion { state -> Awaiting? in
-        cancellation.withCriticalRegion { cancellation in
-          cancellation = true
-        }
+        cancellation.store(true, ordering: .releasing)
         switch state {
           case .awaiting(var awaitings):
             let awaiting = awaitings.remove(.placeHolder(id: awaitingId))

Sources/AsyncChannels/AsyncBufferedChannel.swift

@@ -5,6 +5,7 @@
 //  Created by Thibault Wittemberg on 07/01/2022.
 //
 
+import Atomics
 import DequeModule
 import OrderedCollections
 
@@ -88,19 +89,16 @@ public final class AsyncThrowingBufferedChannel<Element, Failure: Error>: AsyncS
     }
   }
 
-  let ids: ManagedCriticalState<Int>
+  let ids: ManagedAtomic<Int>
   let state: ManagedCriticalState<State>
 
   public init() {
-    self.ids = ManagedCriticalState(0)
+    self.ids = ManagedAtomic(0)
     self.state = ManagedCriticalState(.initial)
   }
 
   func generateId() -> Int {
-    self.ids.withCriticalRegion { ids in
-      ids += 1
-      return ids
-    }
+    ids.wrappingIncrementThenLoad(by: 1, ordering: .relaxed)
   }
 
   var hasBufferedElements: Bool {
@@ -176,12 +174,12 @@ public final class AsyncThrowingBufferedChannel<Element, Failure: Error>: AsyncS
 
   func next(onSuspend: (() -> Void)? = nil) async throws -> Element? {
     let awaitingId = self.generateId()
-    let cancellation = ManagedCriticalState<Bool>(false)
+    let cancellation = ManagedAtomic<Bool>(false)
 
     return try await withTaskCancellationHandler {
       try await withUnsafeThrowingContinuation { [state] (continuation: UnsafeContinuation<Element?, Error>) in
         let decision = state.withCriticalRegion { state -> AwaitingDecision in
-          let isCancelled = cancellation.withCriticalRegion { $0 }
+          let isCancelled = cancellation.load(ordering: .acquiring)
           guard !isCancelled else { return .resume(nil) }
 
           switch state {
@@ -227,9 +225,7 @@ public final class AsyncThrowingBufferedChannel<Element, Failure: Error>: AsyncS
       }
     } onCancel: { [state] in
       let awaiting = state.withCriticalRegion { state -> Awaiting? in
-        cancellation.withCriticalRegion { cancellation in
-          cancellation = true
-        }
+        cancellation.store(true, ordering: .releasing)
         switch state {
           case .awaiting(var awaitings):
             let awaiting = awaitings.remove(.placeHolder(id: awaitingId))