Switch from NIOAtomic to ManagedAtomic.

Author: mrstegeman

Sources/RediStack/RedisConnection.swift

@@ -14,6 +14,7 @@
 
 import struct Foundation.UUID
 import struct Dispatch.DispatchTime
+import Atomics
 import Logging
 import Metrics
 import NIO
@@ -117,22 +118,22 @@ public final class RedisConnection: RedisClient, RedisClientWithUserContext {
     /// - Important: Even when set to `true`, the host machine may still choose to delay sending commands.
     /// - Note: Setting this to `true` will immediately drain the buffer.
     public var sendCommandsImmediately: Bool {
-        get { return autoflush.load() }
+        get { return autoflush.load(ordering: .sequentiallyConsistent) }
         set(newValue) {
             if newValue { self.channel.flush() }
-            autoflush.store(newValue)
+            autoflush.store(newValue, ordering: .sequentiallyConsistent)
         }
     }
     /// Controls the permission of the connection to be able to have PubSub subscriptions or not.
     ///
     /// When set to `true`, this connection is allowed to create subscriptions.
     /// When set to `false`, this connection is not allowed to create subscriptions. Any potentially existing subscriptions will be removed.
     public var allowSubscriptions: Bool {
-        get { self.allowPubSub.load() }
+        get { self.allowPubSub.load(ordering: .sequentiallyConsistent) }
         set(newValue) {
-            self.allowPubSub.store(newValue)
+            self.allowPubSub.store(newValue, ordering: .sequentiallyConsistent)
             // if we're subscribed, and we're not allowed to be in pubsub, end our subscriptions
-            guard self.isSubscribed && !self.allowPubSub.load() else { return }
+            guard self.isSubscribed && !self.allowPubSub.load(ordering: .sequentiallyConsistent) else { return }
             _ = EventLoopFuture<Void>.whenAllComplete([
                 self.unsubscribe(),
                 self.punsubscribe()
@@ -147,9 +148,9 @@ public final class RedisConnection: RedisClient, RedisClientWithUserContext {
     internal let channel: Channel
     private let systemContext: Context
     private var logger: Logger { self.systemContext }
-    
-    private let autoflush: NIOAtomic<Bool> = .makeAtomic(value: true)
-    private let allowPubSub: NIOAtomic<Bool> = .makeAtomic(value: true)
+
+    private let autoflush = ManagedAtomic<Bool>(true)
+    private let allowPubSub = ManagedAtomic<Bool>(true)
     private let _stateLock = NIOLock()
     private var _state = ConnectionState.open
     private var state: ConnectionState {

Sources/RediStack/RedisMetrics.swift

@@ -12,6 +12,7 @@
 //
 //===----------------------------------------------------------------------===//
 
+import Atomics
 import Metrics
 import NIOConcurrencyHelpers
 
@@ -69,10 +70,10 @@ extension RedisMetrics {
     /// A specialized wrapper class for working with `Metrics.Gauge` objects for the purpose of an incrementing or decrementing count of active objects.
     public class IncrementalGauge {
         private let gauge: Gauge
-        private let count = NIOAtomic<Int>.makeAtomic(value: 0)
+        private let count = ManagedAtomic<Int>(0)
 
         /// The number of the objects that are currently reported as active.
-        public var currentCount: Int { return count.load() }
+        public var currentCount: Int { return count.load(ordering: .sequentiallyConsistent) }
 
         internal init(_ label: Label) {
             self.gauge = .init(label: label)
@@ -81,21 +82,21 @@ extension RedisMetrics {
         /// Increments the current count by the amount specified.
         /// - Parameter amount: The number to increase the current count by. Default is `1`.
         public func increment(by amount: Int = 1) {
-            _ = self.count.add(amount)
-            self.gauge.record(self.count.load())
+            self.count.wrappingIncrement(by: amount, ordering: .sequentiallyConsistent)
+            self.gauge.record(self.count.load(ordering: .sequentiallyConsistent))
         }
 
         /// Decrements the current count by the amount specified.
         /// - Parameter amount: The number to decrease the current count by. Default is `1`.
         public func decrement(by amount: Int = 1) {
-            _ = self.count.sub(amount)
-            self.gauge.record(self.count.load())
+            self.count.wrappingDecrement(by: amount, ordering: .sequentiallyConsistent)
+            self.gauge.record(self.count.load(ordering: .sequentiallyConsistent))
         }
 
         /// Resets the current count to `0`.
         public func reset() {
-            _ = self.count.exchange(with: 0)
-            self.gauge.record(self.count.load())
+            _ = self.count.exchange(0, ordering: .sequentiallyConsistent)
+            self.gauge.record(self.count.load(ordering: .sequentiallyConsistent))
         }
     }
 }