Adopt `Mutex` where possible (#2026)

Motivation:

Swift 6 has a `Mutex` which we can use in place of `LockedValueBox` or
`NIOLockedValueBox` in a bunch of places.

Modifications:

- Use `Mutex` where possible
- This causes a number of other changes too as `Mutex` doesn't allocate:
the allocations move to the type holding it instead. This makes sense as
the objects which were previously structs didn't have value semantics.

Result:

Fewer uses of our own lock, no uses of NIOs lock

Sources/GRPCCore/Call/Client/Internal/ClientRPCExecutor+HedgingExecutor.swift

@@ -14,6 +14,8 @@
  * limitations under the License.
  */
 
+public import Synchronization  // would be internal but for usableFromInline
+
 @available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 extension ClientRPCExecutor {
   @usableFromInline
@@ -176,10 +178,10 @@ extension ClientRPCExecutor.HedgingExecutor {
       // of this. To avoid this each attempt goes via a state check before yielding to the sequence
       // ensuring that only one response is used. (If this wasn't the case the response handler
       // could be invoked more than once.)
-      let state = LockedValueBox(State(policy: self.policy))
+      let state = SharedState(policy: self.policy)
 
       // There's always a first attempt, safe to '!'.
-      let (attempt, scheduleNext) = state.withLockedValue({ $0.nextAttemptNumber() })!
+      let (attempt, scheduleNext) = state.withState({ $0.nextAttemptNumber() })!
 
       group.addTask {
         let result = await self._startAttempt(
@@ -210,7 +212,7 @@ extension ClientRPCExecutor.HedgingExecutor {
           switch outcome {
           case .ran:
             // Start a new attempt and possibly schedule the next.
-            if let (attempt, scheduleNext) = state.withLockedValue({ $0.nextAttemptNumber() }) {
+            if let (attempt, scheduleNext) = state.withState({ $0.nextAttemptNumber() }) {
               group.addTask {
                 let result = await self._startAttempt(
                   request: request,
@@ -263,7 +265,7 @@ extension ClientRPCExecutor.HedgingExecutor {
 
               nextScheduledAttempt.cancel()
 
-              if let (attempt, scheduleNext) = state.withLockedValue({ $0.nextAttemptNumber() }) {
+              if let (attempt, scheduleNext) = state.withState({ $0.nextAttemptNumber() }) {
                 group.addTask {
                   let result = await self._startAttempt(
                     request: request,
@@ -317,7 +319,7 @@ extension ClientRPCExecutor.HedgingExecutor {
     method: MethodDescriptor,
     options: CallOptions,
     attempt: Int,
-    state: LockedValueBox<State>,
+    state: SharedState,
     picker: (stream: BroadcastAsyncSequence<Int>, continuation: BroadcastAsyncSequence<Int>.Source),
     responseHandler: @Sendable @escaping (ClientResponse.Stream<Output>) async throws -> R
   ) async -> _HedgingAttemptTaskResult<R, Output>.AttemptResult {
@@ -364,7 +366,7 @@ extension ClientRPCExecutor.HedgingExecutor {
               case .success:
                 self.transport.retryThrottle?.recordSuccess()
 
-                if state.withLockedValue({ $0.receivedUsableResponse() }) {
+                if state.withState({ $0.receivedUsableResponse() }) {
                   try? await picker.continuation.write(attempt)
                   picker.continuation.finish()
                   let result = await Result { try await responseHandler(response) }
@@ -385,7 +387,7 @@ extension ClientRPCExecutor.HedgingExecutor {
                   // A fatal error code counts as a success to the throttle.
                   self.transport.retryThrottle?.recordSuccess()
 
-                  if state.withLockedValue({ $0.receivedUsableResponse() }) {
+                  if state.withState({ $0.receivedUsableResponse() }) {
                     try! await picker.continuation.write(attempt)
                     picker.continuation.finish()
                     let result = await Result { try await responseHandler(response) }
@@ -428,6 +430,24 @@ extension ClientRPCExecutor.HedgingExecutor {
     }
   }
 
+  @usableFromInline
+  final class SharedState {
+    @usableFromInline
+    let state: Mutex<State>
+
+    @inlinable
+    init(policy: HedgingPolicy) {
+      self.state = Mutex(State(policy: policy))
+    }
+
+    @inlinable
+    func withState<ReturnType>(_ body: @Sendable (inout State) -> ReturnType) -> ReturnType {
+      self.state.withLock {
+        body(&$0)
+      }
+    }
+  }
+
   @usableFromInline
   struct State {
     @usableFromInline

Sources/GRPCCore/Internal/Concurrency Primitives/Lock.swift

@@ -237,17 +237,17 @@ extension UnsafeMutablePointer {
 }
 
 @usableFromInline
-package struct LockedValueBox<Value> {
+struct LockedValueBox<Value> {
   @usableFromInline
   let storage: LockStorage<Value>
 
   @inlinable
-  package init(_ value: Value) {
+  init(_ value: Value) {
     self.storage = .create(value: value)
   }
 
   @inlinable
-  package func withLockedValue<T>(_ mutate: (inout Value) throws -> T) rethrows -> T {
+  func withLockedValue<T>(_ mutate: (inout Value) throws -> T) rethrows -> T {
     return try self.storage.withLockedValue(mutate)
   }
 
@@ -255,30 +255,30 @@ package struct LockedValueBox<Value> {
   ///
   /// Prefer ``withLockedValue(_:)`` where possible.
   @usableFromInline
-  package var unsafe: Unsafe {
+  var unsafe: Unsafe {
     Unsafe(storage: self.storage)
   }
 
   @usableFromInline
-  package struct Unsafe {
+  struct Unsafe {
     @usableFromInline
     let storage: LockStorage<Value>
 
     /// Manually acquire the lock.
     @inlinable
-    package func lock() {
+    func lock() {
       self.storage.lock()
     }
 
     /// Manually release the lock.
     @inlinable
-    package func unlock() {
+    func unlock() {
       self.storage.unlock()
     }
 
     /// Mutate the value, assuming the lock has been acquired manually.
     @inlinable
-    package func withValueAssumingLockIsAcquired<T>(
+    func withValueAssumingLockIsAcquired<T>(
       _ mutate: (inout Value) throws -> T
     ) rethrows -> T {
       return try self.storage.withUnsafeMutablePointerToHeader { value in

Sources/GRPCCore/Transport/RetryThrottle.swift

@@ -14,6 +14,8 @@
  * limitations under the License.
  */
 
+private import Synchronization
+
 /// A throttle used to rate-limit retries and hedging attempts.
 ///
 /// gRPC prevents servers from being overloaded by retries and hedging by using a token-based
@@ -28,13 +30,14 @@
 /// the server.
 ///
 /// See also [gRFC A6: client retries](https://github.com/grpc/proposal/blob/master/A6-client-retries.md).
-public struct RetryThrottle: Sendable {
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
+public final class RetryThrottle: Sendable {
   // Note: only three figures after the decimal point from the original token ratio are used so
   //   all computation is done a scaled number of tokens (tokens * 1000). This allows us to do all
   //   computation in integer space.
 
   /// The number of tokens available, multiplied by 1000.
-  private let scaledTokensAvailable: LockedValueBox<Int>
+  private let scaledTokensAvailable: Mutex<Int>
   /// The number of tokens, multiplied by 1000.
   private let scaledTokenRatio: Int
   /// The maximum number of tokens, multiplied by 1000.
@@ -66,14 +69,14 @@ public struct RetryThrottle: Sendable {
   /// If this value is less than or equal to the retry threshold (defined as `maximumTokens / 2`)
   /// then RPCs will not be retried and hedging will be disabled.
   public var tokens: Double {
-    self.scaledTokensAvailable.withLockedValue {
+    self.scaledTokensAvailable.withLock {
       Double($0) / 1000
     }
   }
 
   /// Returns whether retries and hedging are permitted at this time.
   public var isRetryPermitted: Bool {
-    self.scaledTokensAvailable.withLockedValue {
+    self.scaledTokensAvailable.withLock {
       $0 > self.scaledRetryThreshold
     }
   }
@@ -100,21 +103,20 @@ public struct RetryThrottle: Sendable {
     self.scaledMaximumTokens = scaledTokens
     self.scaledRetryThreshold = scaledTokens / 2
     self.scaledTokenRatio = scaledTokenRatio
-    self.scaledTokensAvailable = LockedValueBox(scaledTokens)
+    self.scaledTokensAvailable = Mutex(scaledTokens)
   }
 
   /// Create a new throttle.
   ///
   /// - Parameter policy: The policy to use to configure the throttle.
-  @available(macOS 13.0, iOS 16.0, watchOS 9.0, tvOS 16.0, *)
-  public init(policy: ServiceConfig.RetryThrottling) {
+  public convenience init(policy: ServiceConfig.RetryThrottling) {
     self.init(maximumTokens: policy.maxTokens, tokenRatio: policy.tokenRatio)
   }
 
   /// Records a success, adding a token to the throttle.
   @usableFromInline
   func recordSuccess() {
-    self.scaledTokensAvailable.withLockedValue { value in
+    self.scaledTokensAvailable.withLock { value in
       value = min(self.scaledMaximumTokens, value &+ self.scaledTokenRatio)
     }
   }
@@ -124,7 +126,7 @@ public struct RetryThrottle: Sendable {
   @usableFromInline
   @discardableResult
   func recordFailure() -> Bool {
-    self.scaledTokensAvailable.withLockedValue { value in
+    self.scaledTokensAvailable.withLock { value in
       value = max(0, value &- 1000)
       return value <= self.scaledRetryThreshold
     }

Sources/GRPCHTTP2Core/Client/Connection/Connection.swift

@@ -15,9 +15,9 @@
  */
 
 package import GRPCCore
-internal import NIOConcurrencyHelpers
 package import NIOCore
 package import NIOHTTP2
+private import Synchronization
 
 /// A `Connection` provides communication to a single remote peer.
 ///
@@ -43,8 +43,8 @@ package import NIOHTTP2
 ///   }
 /// }
 /// ```
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
-package struct Connection: Sendable {
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
+package final class Connection: Sendable {
   /// Events which can happen over the lifetime of the connection.
   package enum Event: Sendable {
     /// The connect attempt succeeded and the connection is ready to use.
@@ -96,7 +96,7 @@ package struct Connection: Sendable {
   private let http2Connector: any HTTP2Connector
 
   /// The state of the connection.
-  private let state: NIOLockedValueBox<State>
+  private let state: Mutex<State>
 
   /// The default max request message size in bytes, 4 MiB.
   private static var defaultMaxRequestMessageSizeBytes: Int {
@@ -120,7 +120,7 @@ package struct Connection: Sendable {
     self.http2Connector = http2Connector
     self.event = AsyncStream.makeStream(of: Event.self)
     self.input = AsyncStream.makeStream(of: Input.self)
-    self.state = NIOLockedValueBox(.notConnected)
+    self.state = Mutex(.notConnected)
   }
 
   /// Connect and run the connection.
@@ -135,7 +135,7 @@ package struct Connection: Sendable {
     switch connectResult {
     case .success(let connected):
       // Connected successfully, update state and report the event.
-      self.state.withLockedValue { state in
+      self.state.withLock { state in
         state.connected(connected)
       }
 
@@ -151,7 +151,7 @@ package struct Connection: Sendable {
         for await input in self.input.stream {
           switch input {
           case .close:
-            let asyncChannel = self.state.withLockedValue { $0.beginClosing() }
+            let asyncChannel = self.state.withLock { $0.beginClosing() }
             if let channel = asyncChannel?.channel {
               let event = ClientConnectionHandler.OutboundEvent.closeGracefully
               channel.triggerUserOutboundEvent(event, promise: nil)
@@ -162,7 +162,7 @@ package struct Connection: Sendable {
 
     case .failure(let error):
       // Connect failed, this connection is no longer useful.
-      self.state.withLockedValue { $0.closed() }
+      self.state.withLock { $0.closed() }
       self.finishStreams(withEvent: .connectFailed(error))
     }
   }
@@ -180,7 +180,7 @@ package struct Connection: Sendable {
     descriptor: MethodDescriptor,
     options: CallOptions
   ) async throws -> Stream {
-    let (multiplexer, scheme) = try self.state.withLockedValue { state in
+    let (multiplexer, scheme) = try self.state.withLock { state in
       switch state {
       case .connected(let connected):
         return (connected.multiplexer, connected.scheme)
@@ -259,7 +259,7 @@ package struct Connection: Sendable {
           self.event.continuation.yield(.connectSucceeded)
 
         case .closing(let reason):
-          self.state.withLockedValue { $0.closing() }
+          self.state.withLock { $0.closing() }
 
           switch reason {
           case .goAway(let errorCode, let reason):
@@ -282,7 +282,7 @@ package struct Connection: Sendable {
 
       let finalEvent: Event
       if isReady {
-        let connectionCloseReason: Self.CloseReason
+        let connectionCloseReason: CloseReason
         switch channelCloseReason {
         case .keepaliveExpired:
           connectionCloseReason = .keepaliveTimeout
@@ -323,7 +323,7 @@ package struct Connection: Sendable {
       }
 
       // The connection events sequence has finished: the connection is now closed.
-      self.state.withLockedValue { $0.closed() }
+      self.state.withLock { $0.closed() }
       self.finishStreams(withEvent: finalEvent)
     } catch {
       let finalEvent: Event
@@ -338,7 +338,7 @@ package struct Connection: Sendable {
         finalEvent = .connectFailed(makeNeverReadyError(cause: error))
       }
 
-      self.state.withLockedValue { $0.closed() }
+      self.state.withLock { $0.closed() }
       self.finishStreams(withEvent: finalEvent)
     }
   }
@@ -350,7 +350,7 @@ package struct Connection: Sendable {
   }
 }
 
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 extension Connection {
   package struct Stream {
     package typealias Inbound = NIOAsyncChannelInboundStream<RPCResponsePart>
@@ -412,7 +412,7 @@ extension Connection {
   }
 }
 
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 extension Connection {
   private enum State: Sendable {
     /// The connection is idle or connecting.

Sources/GRPCHTTP2Core/Client/Connection/GRPCChannel.swift

@@ -17,9 +17,10 @@
 internal import Atomics
 internal import DequeModule
 package import GRPCCore
+private import Synchronization
 
 @available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
-package struct GRPCChannel: ClientTransport {
+package final class GRPCChannel: ClientTransport {
   private enum Input: Sendable {
     /// Close the channel, if possible.
     case close
@@ -43,7 +44,7 @@ package struct GRPCChannel: ClientTransport {
   private let resolver: NameResolver
 
   /// The state of the channel.
-  private let state: LockedValueBox<StateMachine>
+  private let state: Mutex<StateMachine>
 
   /// The maximum number of times to attempt to create a stream per RPC.
   ///
@@ -68,8 +69,8 @@ package struct GRPCChannel: ClientTransport {
   private let defaultServiceConfig: ServiceConfig
 
   // These are both read frequently and updated infrequently so may be a bottleneck.
-  private let _methodConfig: LockedValueBox<MethodConfigs>
-  private let _retryThrottle: LockedValueBox<RetryThrottle?>
+  private let _methodConfig: Mutex<MethodConfigs>
+  private let _retryThrottle: Mutex<RetryThrottle?>
 
   package init(
     resolver: NameResolver,
@@ -78,7 +79,7 @@ package struct GRPCChannel: ClientTransport {
     defaultServiceConfig: ServiceConfig
   ) {
     self.resolver = resolver
-    self.state = LockedValueBox(StateMachine())
+    self.state = Mutex(StateMachine())
     self._connectivityState = AsyncStream.makeStream()
     self.input = AsyncStream.makeStream()
     self.connector = connector
@@ -94,10 +95,10 @@ package struct GRPCChannel: ClientTransport {
     self.defaultServiceConfig = defaultServiceConfig
 
     let throttle = defaultServiceConfig.retryThrottling.map { RetryThrottle(policy: $0) }
-    self._retryThrottle = LockedValueBox(throttle)
+    self._retryThrottle = Mutex(throttle)
 
     let methodConfig = MethodConfigs(serviceConfig: defaultServiceConfig)
-    self._methodConfig = LockedValueBox(methodConfig)
+    self._methodConfig = Mutex(methodConfig)
   }
 
   /// The connectivity state of the channel.
@@ -107,7 +108,7 @@ package struct GRPCChannel: ClientTransport {
 
   /// Returns a throttle which gRPC uses to determine whether retries can be executed.
   package var retryThrottle: RetryThrottle? {
-    self._retryThrottle.withLockedValue { $0 }
+    self._retryThrottle.withLock { $0 }
   }
 
   /// Returns the configuration for a given method.
@@ -115,12 +116,12 @@ package struct GRPCChannel: ClientTransport {
   /// - Parameter descriptor: The method to lookup configuration for.
   /// - Returns: Configuration for the method, if it exists.
   package func configuration(forMethod descriptor: MethodDescriptor) -> MethodConfig? {
-    self._methodConfig.withLockedValue { $0[descriptor] }
+    self._methodConfig.withLock { $0[descriptor] }
   }
 
   /// Establishes and maintains a connection to the remote destination.
   package func connect() async {
-    self.state.withLockedValue { $0.start() }
+    self.state.withLock { $0.start() }
     self._connectivityState.continuation.yield(.idle)
 
     await withDiscardingTaskGroup { group in
@@ -146,7 +147,7 @@ package struct GRPCChannel: ClientTransport {
         // When the channel is closed gracefully, the task group running the load balancer mustn't
         // be cancelled (otherwise in-flight RPCs would fail), but the push based resolver will
         // continue indefinitely. Store its handle and cancel it on close when closing the channel.
-        self.state.withLockedValue { state in
+        self.state.withLock { state in
           state.setNameResolverTaskHandle(handle)
         }
 
@@ -270,7 +271,7 @@ extension GRPCChannel {
     options: CallOptions
   ) async -> MakeStreamResult {
     let waitForReady = options.waitForReady ?? true
-    switch self.state.withLockedValue({ $0.makeStream(waitForReady: waitForReady) }) {
+    switch self.state.withLock({ $0.makeStream(waitForReady: waitForReady) }) {
     case .useLoadBalancer(let loadBalancer):
       return await self.makeStream(
         descriptor: descriptor,
@@ -327,7 +328,7 @@ extension GRPCChannel {
           return
         }
 
-        let enqueued = self.state.withLockedValue { state in
+        let enqueued = self.state.withLock { state in
           state.enqueue(continuation: continuation, waitForReady: waitForReady, id: id)
         }
 
@@ -338,7 +339,7 @@ extension GRPCChannel {
         }
       }
     } onCancel: {
-      let continuation = self.state.withLockedValue { state in
+      let continuation = self.state.withLock { state in
         state.dequeueContinuation(id: id)
       }
 
@@ -350,7 +351,7 @@ extension GRPCChannel {
 @available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 extension GRPCChannel {
   private func handleClose(in group: inout DiscardingTaskGroup) {
-    switch self.state.withLockedValue({ $0.close() }) {
+    switch self.state.withLock({ $0.close() }) {
     case .close(let current, let next, let resolver, let continuations):
       resolver?.cancel()
       current.close()
@@ -383,10 +384,10 @@ extension GRPCChannel {
     case .success(let config):
       // Update per RPC configuration.
       let methodConfig = MethodConfigs(serviceConfig: config)
-      self._methodConfig.withLockedValue { $0 = methodConfig }
+      self._methodConfig.withLock { $0 = methodConfig }
 
       let retryThrottle = config.retryThrottling.map { RetryThrottle(policy: $0) }
-      self._retryThrottle.withLockedValue { $0 = retryThrottle }
+      self._retryThrottle.withLock { $0 = retryThrottle }
 
       // Update the load balancer.
       self.updateLoadBalancer(serviceConfig: config, endpoints: result.endpoints, in: &group)
@@ -446,7 +447,7 @@ extension GRPCChannel {
     let loadBalancerConfig = configFromServiceConfig ?? .pickFirst(.init(shuffleAddressList: false))
     switch loadBalancerConfig {
     case .roundRobin:
-      onUpdatePolicy = self.state.withLockedValue { state in
+      onUpdatePolicy = self.state.withLock { state in
         state.changeLoadBalancerKind(to: loadBalancerConfig) {
           let loadBalancer = RoundRobinLoadBalancer(
             connector: self.connector,
@@ -463,7 +464,7 @@ extension GRPCChannel {
         endpoints[0].addresses.shuffle()
       }
 
-      onUpdatePolicy = self.state.withLockedValue { state in
+      onUpdatePolicy = self.state.withLock { state in
         state.changeLoadBalancerKind(to: loadBalancerConfig) {
           let loadBalancer = PickFirstLoadBalancer(
             connector: self.connector,
@@ -527,7 +528,7 @@ extension GRPCChannel {
   ) async {
     switch event {
     case .connectivityStateChanged(let connectivityState):
-      let actions = self.state.withLockedValue { state in
+      let actions = self.state.withLock { state in
         state.loadBalancerStateChanged(to: connectivityState, id: loadBalancerID)
       }
 

Sources/GRPCHTTP2Core/Client/Connection/LoadBalancers/LoadBalancer.swift

@@ -14,13 +14,13 @@
  * limitations under the License.
  */
 
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 package enum LoadBalancer: Sendable {
   case roundRobin(RoundRobinLoadBalancer)
   case pickFirst(PickFirstLoadBalancer)
 }
 
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 extension LoadBalancer {
   package init(_ loadBalancer: RoundRobinLoadBalancer) {
     self = .roundRobin(loadBalancer)

Sources/GRPCHTTP2Core/Client/Connection/LoadBalancers/PickFirstLoadBalancer.swift

@@ -15,6 +15,7 @@
  */
 
 package import GRPCCore
+private import Synchronization
 
 /// A load-balancer which has a single subchannel.
 ///
@@ -55,8 +56,8 @@ package import GRPCCore
 ///   }
 /// }
 /// ```
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
-package struct PickFirstLoadBalancer {
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
+package final class PickFirstLoadBalancer: Sendable {
   enum Input: Sendable, Hashable {
     /// Update the addresses used by the load balancer to the following endpoints.
     case updateEndpoint(Endpoint)
@@ -87,7 +88,7 @@ package struct PickFirstLoadBalancer {
   private let enabledCompression: CompressionAlgorithmSet
 
   /// The state of the load-balancer.
-  private let state: LockedValueBox<State>
+  private let state: Mutex<State>
 
   /// The ID of this load balancer.
   internal let id: LoadBalancerID
@@ -103,7 +104,7 @@ package struct PickFirstLoadBalancer {
     self.defaultCompression = defaultCompression
     self.enabledCompression = enabledCompression
     self.id = LoadBalancerID()
-    self.state = LockedValueBox(State())
+    self.state = Mutex(State())
 
     self.event = AsyncStream.makeStream(of: LoadBalancerEvent.self)
     self.input = AsyncStream.makeStream(of: Input.self)
@@ -153,7 +154,7 @@ package struct PickFirstLoadBalancer {
   ///
   /// - Returns: A subchannel, or `nil` if there aren't any ready subchannels.
   package func pickSubchannel() -> Subchannel? {
-    let onPickSubchannel = self.state.withLockedValue { $0.pickSubchannel() }
+    let onPickSubchannel = self.state.withLock { $0.pickSubchannel() }
     switch onPickSubchannel {
     case .picked(let subchannel):
       return subchannel
@@ -164,12 +165,12 @@ package struct PickFirstLoadBalancer {
   }
 }
 
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 extension PickFirstLoadBalancer {
   private func handleUpdateEndpoint(_ endpoint: Endpoint, in group: inout DiscardingTaskGroup) {
     if endpoint.addresses.isEmpty { return }
 
-    let onUpdate = self.state.withLockedValue { state in
+    let onUpdate = self.state.withLock { state in
       state.updateEndpoint(endpoint) { endpoint, id in
         Subchannel(
           endpoint: endpoint,
@@ -220,7 +221,7 @@ extension PickFirstLoadBalancer {
     _ connectivityState: ConnectivityState,
     id: SubchannelID
   ) {
-    let onUpdateState = self.state.withLockedValue {
+    let onUpdateState = self.state.withLock {
       $0.updateSubchannelConnectivityState(connectivityState, id: id)
     }
 
@@ -241,13 +242,13 @@ extension PickFirstLoadBalancer {
   }
 
   private func handleGoAway(id: SubchannelID) {
-    self.state.withLockedValue { state in
+    self.state.withLock { state in
       state.receivedGoAway(id: id)
     }
   }
 
   private func handleCloseInput() {
-    let onClose = self.state.withLockedValue { $0.close() }
+    let onClose = self.state.withLock { $0.close() }
     switch onClose {
     case .closeSubchannels(let subchannel1, let subchannel2):
       self.event.continuation.yield(.connectivityStateChanged(.shutdown))
@@ -265,7 +266,7 @@ extension PickFirstLoadBalancer {
   }
 }
 
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 extension PickFirstLoadBalancer {
   enum State: Sendable {
     case active(Active)
@@ -278,7 +279,7 @@ extension PickFirstLoadBalancer {
   }
 }
 
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 extension PickFirstLoadBalancer.State {
   struct Active: Sendable {
     var endpoint: Endpoint?
@@ -307,7 +308,7 @@ extension PickFirstLoadBalancer.State {
   }
 }
 
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 extension PickFirstLoadBalancer.State.Active {
   mutating func updateEndpoint(
     _ endpoint: Endpoint,
@@ -470,7 +471,7 @@ extension PickFirstLoadBalancer.State.Active {
   }
 }
 
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 extension PickFirstLoadBalancer.State.Closing {
   mutating func updateSubchannelConnectivityState(
     _ connectivityState: ConnectivityState,
@@ -511,7 +512,7 @@ extension PickFirstLoadBalancer.State.Closing {
   }
 }
 
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 extension PickFirstLoadBalancer.State {
   enum OnUpdateEndpoint {
     case connect(Subchannel, close: Subchannel?)

Sources/GRPCHTTP2Core/Client/Connection/LoadBalancers/RoundRobinLoadBalancer.swift

@@ -15,6 +15,7 @@
  */
 
 package import GRPCCore
+private import NIOConcurrencyHelpers
 
 /// A load-balancer which maintains to a set of subchannels and uses round-robin to pick a
 /// subchannel when picking a subchannel to use.
@@ -57,8 +58,8 @@ package import GRPCCore
 ///   }
 /// }
 /// ```
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
-package struct RoundRobinLoadBalancer {
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
+package final class RoundRobinLoadBalancer: Sendable {
   enum Input: Sendable, Hashable {
     /// Update the addresses used by the load balancer to the following endpoints.
     case updateAddresses([Endpoint])
@@ -102,8 +103,9 @@ package struct RoundRobinLoadBalancer {
   /// Inputs which this load balancer should react to.
   private let input: (stream: AsyncStream<Input>, continuation: AsyncStream<Input>.Continuation)
 
+  // Uses NIOLockedValueBox to workaround: https://github.com/swiftlang/swift/issues/76007
   /// The state of the load balancer.
-  private let state: LockedValueBox<State>
+  private let state: NIOLockedValueBox<State>
 
   /// A connector, capable of creating connections.
   private let connector: any HTTP2Connector
@@ -134,7 +136,7 @@ package struct RoundRobinLoadBalancer {
 
     self.event = AsyncStream.makeStream(of: LoadBalancerEvent.self)
     self.input = AsyncStream.makeStream(of: Input.self)
-    self.state = LockedValueBox(.active(State.Active()))
+    self.state = NIOLockedValueBox(.active(State.Active()))
 
     // The load balancer starts in the idle state.
     self.event.continuation.yield(.connectivityStateChanged(.idle))
@@ -196,7 +198,7 @@ package struct RoundRobinLoadBalancer {
   }
 }
 
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 extension RoundRobinLoadBalancer {
   /// Handles an update in endpoints.
   ///
@@ -338,7 +340,7 @@ extension RoundRobinLoadBalancer {
   }
 }
 
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 extension RoundRobinLoadBalancer {
   private enum State {
     case active(Active)

Sources/GRPCHTTP2Core/Client/Connection/LoadBalancers/Subchannel.swift

@@ -15,7 +15,7 @@
  */
 
 package import GRPCCore
-internal import NIOConcurrencyHelpers
+private import Synchronization
 
 /// A ``Subchannel`` provides communication to a single ``Endpoint``.
 ///
@@ -43,8 +43,8 @@ internal import NIOConcurrencyHelpers
 ///   }
 /// }
 /// ```
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
-package struct Subchannel {
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
+package final class Subchannel: Sendable {
   package enum Event: Sendable, Hashable {
     /// The connection received a GOAWAY and will close soon. No new streams
     /// should be opened on this connection.
@@ -73,7 +73,7 @@ package struct Subchannel {
   private let input: (stream: AsyncStream<Input>, continuation: AsyncStream<Input>.Continuation)
 
   /// The state of the subchannel.
-  private let state: NIOLockedValueBox<State>
+  private let state: Mutex<State>
 
   /// The endpoint this subchannel is targeting.
   let endpoint: Endpoint
@@ -103,7 +103,7 @@ package struct Subchannel {
   ) {
     assert(!endpoint.addresses.isEmpty, "endpoint.addresses mustn't be empty")
 
-    self.state = NIOLockedValueBox(.notConnected(.initial))
+    self.state = Mutex(.notConnected(.initial))
     self.endpoint = endpoint
     self.id = id
     self.connector = connector
@@ -117,7 +117,7 @@ package struct Subchannel {
   }
 }
 
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 extension Subchannel {
   /// A stream of events which can happen to the subchannel.
   package var events: AsyncStream<Event> {
@@ -173,7 +173,7 @@ extension Subchannel {
     descriptor: MethodDescriptor,
     options: CallOptions
   ) async throws -> Connection.Stream {
-    let connection: Connection? = self.state.withLockedValue { state in
+    let connection: Connection? = self.state.withLock { state in
       switch state {
       case .notConnected, .connecting, .goingAway, .shuttingDown, .shutDown:
         return nil
@@ -190,10 +190,10 @@ extension Subchannel {
   }
 }
 
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 extension Subchannel {
   private func handleConnectInput(in group: inout DiscardingTaskGroup) {
-    let connection = self.state.withLockedValue { state in
+    let connection = self.state.withLock { state in
       state.makeConnection(
         to: self.endpoint.addresses,
         using: self.connector,
@@ -214,7 +214,7 @@ extension Subchannel {
   }
 
   private func handleBackedOffInput(in group: inout DiscardingTaskGroup) {
-    switch self.state.withLockedValue({ $0.backedOff() }) {
+    switch self.state.withLock({ $0.backedOff() }) {
     case .none:
       ()
 
@@ -230,7 +230,7 @@ extension Subchannel {
   }
 
   private func handleShutDownInput(in group: inout DiscardingTaskGroup) {
-    switch self.state.withLockedValue({ $0.shutDown() }) {
+    switch self.state.withLock({ $0.shutDown() }) {
     case .none:
       ()
 
@@ -269,7 +269,7 @@ extension Subchannel {
   }
 
   private func handleConnectSucceededEvent() {
-    switch self.state.withLockedValue({ $0.connectSucceeded() }) {
+    switch self.state.withLock({ $0.connectSucceeded() }) {
     case .updateStateToReady:
       // Emit a connectivity state change: the load balancer can now use this subchannel.
       self.event.continuation.yield(.connectivityStateChanged(.ready))
@@ -286,7 +286,7 @@ extension Subchannel {
   }
 
   private func handleConnectFailedEvent(in group: inout DiscardingTaskGroup) {
-    let onConnectFailed = self.state.withLockedValue { $0.connectFailed(connector: self.connector) }
+    let onConnectFailed = self.state.withLock { $0.connectFailed(connector: self.connector) }
     switch onConnectFailed {
     case .connect(let connection):
       // Try the next address.
@@ -316,7 +316,7 @@ extension Subchannel {
   }
 
   private func handleGoingAwayEvent() {
-    let isGoingAway = self.state.withLockedValue { $0.goingAway() }
+    let isGoingAway = self.state.withLock { $0.goingAway() }
     guard isGoingAway else { return }
 
     // Notify the load balancer that the subchannel is going away to stop it from being used.
@@ -330,7 +330,7 @@ extension Subchannel {
     _ reason: Connection.CloseReason,
     in group: inout DiscardingTaskGroup
   ) {
-    switch self.state.withLockedValue({ $0.closed(reason: reason) }) {
+    switch self.state.withLock({ $0.closed(reason: reason) }) {
     case .nothing:
       ()
 
@@ -368,7 +368,7 @@ extension Subchannel {
   }
 }
 
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 extension Subchannel {
   ///            ┌───────────────┐
   ///   ┌───────▶│ NOT CONNECTED │───────────shutDown─────────────┐

Sources/GRPCHTTP2Core/Client/Connection/RequestQueue.swift

@@ -16,7 +16,7 @@
 
 internal import DequeModule
 
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 struct RequestQueue {
   typealias Continuation = CheckedContinuation<LoadBalancer, any Error>
 

Sources/GRPCHTTP2TransportNIOPosix/HTTP2ServerTransport+Posix.swift

@@ -20,6 +20,7 @@ internal import NIOCore
 internal import NIOExtras
 internal import NIOHTTP2
 public import NIOPosix  // has to be public because of default argument value in init
+private import Synchronization
 
 #if canImport(NIOSSL)
 import NIOSSL
@@ -28,7 +29,7 @@ import NIOSSL
 extension HTTP2ServerTransport {
   /// A NIOPosix-backed implementation of a server transport.
   @available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
-  public struct Posix: ServerTransport, ListeningServerTransport {
+  public final class Posix: ServerTransport, ListeningServerTransport {
     private let address: GRPCHTTP2Core.SocketAddress
     private let config: Config
     private let eventLoopGroup: MultiThreadedEventLoopGroup
@@ -129,7 +130,7 @@ extension HTTP2ServerTransport {
       }
     }
 
-    private let listeningAddressState: LockedValueBox<State>
+    private let listeningAddressState: Mutex<State>
 
     /// The listening address for this server transport.
     ///
@@ -141,7 +142,7 @@ extension HTTP2ServerTransport {
     public var listeningAddress: GRPCHTTP2Core.SocketAddress {
       get async throws {
         try await self.listeningAddressState
-          .withLockedValue { try $0.listeningAddressFuture }
+          .withLock { try $0.listeningAddressFuture }
           .get()
       }
     }
@@ -163,14 +164,14 @@ extension HTTP2ServerTransport {
       self.serverQuiescingHelper = ServerQuiescingHelper(group: self.eventLoopGroup)
 
       let eventLoop = eventLoopGroup.any()
-      self.listeningAddressState = LockedValueBox(.idle(eventLoop.makePromise()))
+      self.listeningAddressState = Mutex(.idle(eventLoop.makePromise()))
     }
 
     public func listen(
       _ streamHandler: @escaping @Sendable (RPCStream<Inbound, Outbound>) async -> Void
     ) async throws {
       defer {
-        switch self.listeningAddressState.withLockedValue({ $0.close() }) {
+        switch self.listeningAddressState.withLock({ $0.close() }) {
         case .failPromise(let promise, let error):
           promise.fail(error)
         case .doNothing:
@@ -240,7 +241,7 @@ extension HTTP2ServerTransport {
           }
         }
 
-      let action = self.listeningAddressState.withLockedValue {
+      let action = self.listeningAddressState.withLock {
         $0.addressBound(
           serverChannel.channel.localAddress,
           userProvidedAddress: self.address

Sources/GRPCHTTP2TransportNIOTransportServices/HTTP2ServerTransport+TransportServices.swift

@@ -23,10 +23,12 @@ internal import NIOCore
 internal import NIOExtras
 internal import NIOHTTP2
 
+private import Synchronization
+
 extension HTTP2ServerTransport {
   /// A NIO Transport Services-backed implementation of a server transport.
   @available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
-  public struct TransportServices: ServerTransport, ListeningServerTransport {
+  public final class TransportServices: ServerTransport, ListeningServerTransport {
     private let address: GRPCHTTP2Core.SocketAddress
     private let config: Config
     private let eventLoopGroup: NIOTSEventLoopGroup
@@ -120,7 +122,7 @@ extension HTTP2ServerTransport {
       }
     }
 
-    private let listeningAddressState: LockedValueBox<State>
+    private let listeningAddressState: Mutex<State>
 
     /// The listening address for this server transport.
     ///
@@ -132,7 +134,7 @@ extension HTTP2ServerTransport {
     public var listeningAddress: GRPCHTTP2Core.SocketAddress {
       get async throws {
         try await self.listeningAddressState
-          .withLockedValue { try $0.listeningAddressFuture }
+          .withLock { try $0.listeningAddressFuture }
           .get()
       }
     }
@@ -154,14 +156,14 @@ extension HTTP2ServerTransport {
       self.serverQuiescingHelper = ServerQuiescingHelper(group: self.eventLoopGroup)
 
       let eventLoop = eventLoopGroup.any()
-      self.listeningAddressState = LockedValueBox(.idle(eventLoop.makePromise()))
+      self.listeningAddressState = Mutex(.idle(eventLoop.makePromise()))
     }
 
     public func listen(
       _ streamHandler: @escaping @Sendable (RPCStream<Inbound, Outbound>) async -> Void
     ) async throws {
       defer {
-        switch self.listeningAddressState.withLockedValue({ $0.close() }) {
+        switch self.listeningAddressState.withLock({ $0.close() }) {
         case .failPromise(let promise, let error):
           promise.fail(error)
         case .doNothing:
@@ -194,7 +196,7 @@ extension HTTP2ServerTransport {
           }
         }
 
-      let action = self.listeningAddressState.withLockedValue {
+      let action = self.listeningAddressState.withLock {
         $0.addressBound(serverChannel.channel.localAddress)
       }
       switch action {

Sources/GRPCInProcessTransport/InProcessClientTransport.swift

@@ -15,6 +15,7 @@
  */
 
 public import GRPCCore
+private import Synchronization
 
 /// An in-process implementation of a ``ClientTransport``.
 ///
@@ -34,7 +35,7 @@ public import GRPCCore
 ///
 /// - SeeAlso: ``ClientTransport``
 @available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
-public struct InProcessClientTransport: ClientTransport {
+public final class InProcessClientTransport: ClientTransport {
   private enum State: Sendable {
     struct UnconnectedState {
       var serverTransport: InProcessServerTransport
@@ -101,7 +102,7 @@ public struct InProcessClientTransport: ClientTransport {
   public let retryThrottle: RetryThrottle?
 
   private let methodConfig: MethodConfigs
-  private let state: LockedValueBox<State>
+  private let state: Mutex<State>
 
   /// Creates a new in-process client transport.
   ///
@@ -114,7 +115,7 @@ public struct InProcessClientTransport: ClientTransport {
   ) {
     self.retryThrottle = serviceConfig.retryThrottling.map { RetryThrottle(policy: $0) }
     self.methodConfig = MethodConfigs(serviceConfig: serviceConfig)
-    self.state = LockedValueBox(.unconnected(.init(serverTransport: server)))
+    self.state = Mutex(.unconnected(.init(serverTransport: server)))
   }
 
   /// Establish and maintain a connection to the remote destination.
@@ -129,7 +130,7 @@ public struct InProcessClientTransport: ClientTransport {
   /// task this function runs in.
   public func connect() async throws {
     let (stream, continuation) = AsyncStream<Void>.makeStream()
-    try self.state.withLockedValue { state in
+    try self.state.withLock { state in
       switch state {
       case .unconnected(let unconnectedState):
         state = .connected(
@@ -164,7 +165,7 @@ public struct InProcessClientTransport: ClientTransport {
 
     // If at this point there are any open streams, it's because Cancellation
     // occurred and all open streams must now be closed.
-    let openStreams = self.state.withLockedValue { state in
+    let openStreams = self.state.withLock { state in
       switch state {
       case .unconnected:
         // We have transitioned to connected, and we can't transition back.
@@ -190,7 +191,7 @@ public struct InProcessClientTransport: ClientTransport {
   ///
   /// If you want to forcefully cancel all active streams then cancel the task running ``connect()``.
   public func close() {
-    let maybeContinuation: AsyncStream<Void>.Continuation? = self.state.withLockedValue { state in
+    let maybeContinuation: AsyncStream<Void>.Continuation? = self.state.withLock { state in
       switch state {
       case .unconnected:
         state = .closed(.init())
@@ -246,7 +247,7 @@ public struct InProcessClientTransport: ClientTransport {
       outbound: RPCWriter.Closable(wrapping: response.continuation)
     )
 
-    let waitForConnectionStream: AsyncStream<Void>? = self.state.withLockedValue { state in
+    let waitForConnectionStream: AsyncStream<Void>? = self.state.withLock { state in
       if case .unconnected(var unconnectedState) = state {
         let (stream, continuation) = AsyncStream<Void>.makeStream()
         unconnectedState.pendingStreams.append(continuation)
@@ -264,7 +265,7 @@ public struct InProcessClientTransport: ClientTransport {
       try Task.checkCancellation()
     }
 
-    let streamID = try self.state.withLockedValue { state in
+    let streamID = try self.state.withLock { state in
       switch state {
       case .unconnected:
         // The state cannot be unconnected because if it was, then the above
@@ -305,7 +306,7 @@ public struct InProcessClientTransport: ClientTransport {
     defer {
       clientStream.outbound.finish()
 
-      let maybeEndContinuation = self.state.withLockedValue { state in
+      let maybeEndContinuation = self.state.withLock { state in
         switch state {
         case .unconnected:
           // The state cannot be unconnected at this point, because if we made

Sources/Services/Health/HealthService.swift

@@ -15,6 +15,7 @@
  */
 
 internal import GRPCCore
+private import Synchronization
 
 @available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 internal struct HealthService: Grpc_Health_V1_HealthServiceProtocol {
@@ -67,21 +68,21 @@ internal struct HealthService: Grpc_Health_V1_HealthServiceProtocol {
 
 @available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 extension HealthService {
-  private struct State: Sendable {
+  private final class State: Sendable {
     // The state of each service keyed by the fully qualified service name.
-    private let lockedStorage = LockedValueBox([String: ServiceState]())
+    private let lockedStorage = Mutex([String: ServiceState]())
 
     fileprivate func currentStatus(
       ofService service: String
     ) -> Grpc_Health_V1_HealthCheckResponse.ServingStatus? {
-      return self.lockedStorage.withLockedValue { $0[service]?.currentStatus }
+      return self.lockedStorage.withLock { $0[service]?.currentStatus }
     }
 
     fileprivate func updateStatus(
       _ status: Grpc_Health_V1_HealthCheckResponse.ServingStatus,
       forService service: String
     ) {
-      self.lockedStorage.withLockedValue { storage in
+      self.lockedStorage.withLock { storage in
         storage[service, default: ServiceState(status: status)].updateStatus(status)
       }
     }
@@ -90,7 +91,7 @@ extension HealthService {
       _ continuation: AsyncStream<Grpc_Health_V1_HealthCheckResponse.ServingStatus>.Continuation,
       forService service: String
     ) {
-      self.lockedStorage.withLockedValue { storage in
+      self.lockedStorage.withLock { storage in
         storage[service, default: ServiceState(status: .serviceUnknown)]
           .addContinuation(continuation)
       }

Tests/GRPCCoreTests/Transport/RetryThrottleTests.swift

@@ -18,6 +18,7 @@ import XCTest
 
 @testable import GRPCCore
 
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 final class RetryThrottleTests: XCTestCase {
   func testThrottleOnInit() {
     let throttle = RetryThrottle(maximumTokens: 10, tokenRatio: 0.1)

Tests/GRPCHTTP2CoreTests/Client/Connection/Connection+Equatable.swift

@@ -19,15 +19,15 @@ import GRPCHTTP2Core
 
 // Equatable conformance for these types is 'best effort', this is sufficient for testing but not
 // for general use.
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 extension Connection.Event: Equatable {}
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 extension Connection.CloseReason: Equatable {}
 
 extension ClientConnectionEvent: Equatable {}
 extension ClientConnectionEvent.CloseReason: Equatable {}
 
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 extension Connection.Event {
   package static func == (lhs: Connection.Event, rhs: Connection.Event) -> Bool {
     switch (lhs, rhs) {
@@ -47,7 +47,7 @@ extension Connection.Event {
   }
 }
 
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 extension Connection.CloseReason {
   package static func == (lhs: Connection.CloseReason, rhs: Connection.CloseReason) -> Bool {
     switch (lhs, rhs) {

Tests/GRPCHTTP2CoreTests/Client/Connection/ConnectionTests.swift

@@ -17,14 +17,13 @@
 import DequeModule
 import GRPCCore
 import GRPCHTTP2Core
-import NIOConcurrencyHelpers
 import NIOCore
 import NIOHPACK
 import NIOHTTP2
 import NIOPosix
 import XCTest
 
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 final class ConnectionTests: XCTestCase {
   func testConnectThenClose() async throws {
     try await ConnectionTest.run(connector: .posix()) { context, event in

Tests/GRPCHTTP2CoreTests/Client/Connection/LoadBalancers/LoadBalancerTest.swift

@@ -17,7 +17,7 @@
 import GRPCHTTP2Core
 import XCTest
 
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 enum LoadBalancerTest {
   struct Context {
     let servers: [(server: TestServer, address: GRPCHTTP2Core.SocketAddress)]
@@ -163,7 +163,7 @@ enum LoadBalancerTest {
   }
 }
 
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 extension LoadBalancerTest.Context {
   var roundRobin: RoundRobinLoadBalancer? {
     switch self.loadBalancer {

Tests/GRPCHTTP2CoreTests/Client/Connection/LoadBalancers/PickFirstLoadBalancerTests.swift

@@ -21,7 +21,7 @@ import NIOHTTP2
 import NIOPosix
 import XCTest
 
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 final class PickFirstLoadBalancerTests: XCTestCase {
   func testPickFirstConnectsToServer() async throws {
     try await LoadBalancerTest.pickFirst(servers: 1, connector: .posix()) { context, event in

Tests/GRPCHTTP2CoreTests/Client/Connection/LoadBalancers/RoundRobinLoadBalancerTests.swift

@@ -21,7 +21,7 @@ import NIOHTTP2
 import NIOPosix
 import XCTest
 
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 final class RoundRobinLoadBalancerTests: XCTestCase {
   func testMultipleConnectionsAreEstablished() async throws {
     try await LoadBalancerTest.roundRobin(servers: 3, connector: .posix()) { context, event in

Tests/GRPCHTTP2CoreTests/Client/Connection/LoadBalancers/SubchannelTests.swift

@@ -21,7 +21,7 @@ import NIOHTTP2
 import NIOPosix
 import XCTest
 
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 final class SubchannelTests: XCTestCase {
   func testMakeStreamOnIdleSubchannel() async throws {
     let subchannel = self.makeSubchannel(

Tests/GRPCHTTP2CoreTests/Client/Connection/RequestQueueTests.swift

@@ -15,11 +15,12 @@
  */
 
 import GRPCCore
+import Synchronization
 import XCTest
 
 @testable import GRPCHTTP2Core
 
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 final class RequestQueueTests: XCTestCase {
   struct AnErrorToAvoidALeak: Error {}
 
@@ -45,7 +46,7 @@ final class RequestQueueTests: XCTestCase {
 
   func testPopFirstMultiple() async {
     await withTaskGroup(of: QueueEntryID.self) { group in
-      let queue = LockedValueBox(RequestQueue())
+      let queue = SharedRequestQueue()
       let signal1 = AsyncStream.makeStream(of: Void.self)
       let signal2 = AsyncStream.makeStream(of: Void.self)
 
@@ -54,7 +55,7 @@ final class RequestQueueTests: XCTestCase {
 
       group.addTask {
         _ = try? await withCheckedThrowingContinuation { continuation in
-          queue.withLockedValue {
+          queue.withQueue {
             $0.append(continuation: continuation, waitForReady: false, id: id1)
           }
 
@@ -70,7 +71,7 @@ final class RequestQueueTests: XCTestCase {
         for await _ in signal1.stream {}
 
         _ = try? await withCheckedThrowingContinuation { continuation in
-          queue.withLockedValue {
+          queue.withQueue {
             $0.append(continuation: continuation, waitForReady: false, id: id2)
           }
 
@@ -85,7 +86,7 @@ final class RequestQueueTests: XCTestCase {
       for await _ in signal2.stream {}
 
       for id in [id1, id2] {
-        let continuation = queue.withLockedValue { $0.popFirst() }
+        let continuation = queue.withQueue { $0.popFirst() }
         continuation?.resume(throwing: AnErrorToAvoidALeak())
         let actual = await group.next()
         XCTAssertEqual(id, actual)
@@ -110,7 +111,7 @@ final class RequestQueueTests: XCTestCase {
 
   func testRemoveEntryByIDMultiple() async {
     await withTaskGroup(of: QueueEntryID.self) { group in
-      let queue = LockedValueBox(RequestQueue())
+      let queue = SharedRequestQueue()
       let signal1 = AsyncStream.makeStream(of: Void.self)
       let signal2 = AsyncStream.makeStream(of: Void.self)
 
@@ -119,7 +120,7 @@ final class RequestQueueTests: XCTestCase {
 
       group.addTask {
         _ = try? await withCheckedThrowingContinuation { continuation in
-          queue.withLockedValue {
+          queue.withQueue {
             $0.append(continuation: continuation, waitForReady: false, id: id1)
           }
 
@@ -135,7 +136,7 @@ final class RequestQueueTests: XCTestCase {
         for await _ in signal1.stream {}
 
         _ = try? await withCheckedThrowingContinuation { continuation in
-          queue.withLockedValue {
+          queue.withQueue {
             $0.append(continuation: continuation, waitForReady: false, id: id2)
           }
 
@@ -150,7 +151,7 @@ final class RequestQueueTests: XCTestCase {
       for await _ in signal2.stream {}
 
       for id in [id1, id2] {
-        let continuation = queue.withLockedValue { $0.removeEntry(withID: id) }
+        let continuation = queue.withQueue { $0.removeEntry(withID: id) }
         continuation?.resume(throwing: AnErrorToAvoidALeak())
         let actual = await group.next()
         XCTAssertEqual(id, actual)
@@ -159,7 +160,7 @@ final class RequestQueueTests: XCTestCase {
   }
 
   func testRemoveFastFailingEntries() async throws {
-    let queue = LockedValueBox(RequestQueue())
+    let queue = SharedRequestQueue()
     let enqueued = AsyncStream.makeStream(of: Void.self)
 
     try await withThrowingTaskGroup(of: Void.self) { group in
@@ -177,7 +178,7 @@ final class RequestQueueTests: XCTestCase {
           group.addTask {
             do {
               _ = try await withCheckedThrowingContinuation { continuation in
-                queue.withLockedValue {
+                queue.withQueue {
                   $0.append(continuation: continuation, waitForReady: waitForReady, id: id)
                 }
                 enqueued.continuation.yield()
@@ -199,7 +200,7 @@ final class RequestQueueTests: XCTestCase {
       }
 
       // Remove all fast-failing continuations.
-      let continuations = queue.withLockedValue {
+      let continuations = queue.withQueue {
         $0.removeFastFailingEntries()
       }
 
@@ -208,13 +209,13 @@ final class RequestQueueTests: XCTestCase {
       }
 
       for id in failFastIDs {
-        queue.withLockedValue {
+        queue.withQueue {
           XCTAssertNil($0.removeEntry(withID: id))
         }
       }
 
       for id in waitForReadyIDs {
-        let maybeContinuation = queue.withLockedValue { $0.removeEntry(withID: id) }
+        let maybeContinuation = queue.withQueue { $0.removeEntry(withID: id) }
         let continuation = try XCTUnwrap(maybeContinuation)
         continuation.resume(throwing: AnErrorToAvoidALeak())
       }
@@ -222,14 +223,14 @@ final class RequestQueueTests: XCTestCase {
   }
 
   func testRemoveAll() async throws {
-    let queue = LockedValueBox(RequestQueue())
+    let queue = SharedRequestQueue()
     let enqueued = AsyncStream.makeStream(of: Void.self)
 
     await withThrowingTaskGroup(of: Void.self) { group in
       for _ in 0 ..< 10 {
         group.addTask {
           _ = try await withCheckedThrowingContinuation { continuation in
-            queue.withLockedValue {
+            queue.withQueue {
               $0.append(continuation: continuation, waitForReady: false, id: QueueEntryID())
             }
 
@@ -247,13 +248,27 @@ final class RequestQueueTests: XCTestCase {
         }
       }
 
-      let continuations = queue.withLockedValue { $0.removeAll() }
+      let continuations = queue.withQueue { $0.removeAll() }
       XCTAssertEqual(continuations.count, 10)
-      XCTAssertNil(queue.withLockedValue { $0.popFirst() })
+      XCTAssertNil(queue.withQueue { $0.popFirst() })
 
       for continuation in continuations {
         continuation.resume(throwing: AnErrorToAvoidALeak())
       }
     }
   }
+
+  final class SharedRequestQueue: Sendable {
+    private let protectedQueue: Mutex<RequestQueue>
+
+    init() {
+      self.protectedQueue = Mutex(RequestQueue())
+    }
+
+    func withQueue<T>(_ body: @Sendable (inout RequestQueue) throws -> T) rethrows -> T {
+      try self.protectedQueue.withLock {
+        try body(&$0)
+      }
+    }
+  }
 }

Tests/GRPCHTTP2CoreTests/Client/Connection/Utilities/ConnectionTest.swift

@@ -21,7 +21,7 @@ import NIOCore
 import NIOHTTP2
 import NIOPosix
 
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 enum ConnectionTest {
   struct Context {
     var server: Server
@@ -61,7 +61,7 @@ enum ConnectionTest {
   }
 }
 
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 extension ConnectionTest {
   /// A server which only expected to accept a single connection.
   final class Server {
@@ -142,7 +142,7 @@ extension ConnectionTest {
   }
 }
 
-@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 extension ConnectionTest {
   /// Succeeds a promise when a SETTINGS frame ack has been read.
   private final class SucceedOnSettingsAck: ChannelInboundHandler {

Tests/GRPCHTTP2CoreTests/Client/Connection/Utilities/TestServer.swift

@@ -15,31 +15,31 @@
  */
 
 import GRPCCore
-import NIOConcurrencyHelpers
 import NIOCore
 import NIOHTTP2
 import NIOPosix
+import Synchronization
 import XCTest
 
 @testable import GRPCHTTP2Core
 
-@available(macOS 13.0, iOS 16.0, watchOS 9.0, tvOS 16.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 final class TestServer: Sendable {
   private let eventLoopGroup: any EventLoopGroup
   private typealias Stream = NIOAsyncChannel<RPCRequestPart, RPCResponsePart>
   private typealias Multiplexer = NIOHTTP2AsyncSequence<Stream>
 
-  private let connected: NIOLockedValueBox<[any Channel]>
+  private let connected: Mutex<[any Channel]>
 
   typealias Inbound = NIOAsyncChannelInboundStream<RPCRequestPart>
   typealias Outbound = NIOAsyncChannelOutboundWriter<RPCResponsePart>
 
-  private let server: NIOLockedValueBox<NIOAsyncChannel<Multiplexer, Never>?>
+  private let server: Mutex<NIOAsyncChannel<Multiplexer, Never>?>
 
   init(eventLoopGroup: any EventLoopGroup) {
     self.eventLoopGroup = eventLoopGroup
-    self.server = NIOLockedValueBox(nil)
-    self.connected = NIOLockedValueBox([])
+    self.server = Mutex(nil)
+    self.connected = Mutex([])
   }
 
   enum Target {
@@ -48,20 +48,20 @@ final class TestServer: Sendable {
   }
 
   var clients: [any Channel] {
-    return self.connected.withLockedValue { $0 }
+    return self.connected.withLock { $0 }
   }
 
   func bind(to target: Target = .localhost) async throws -> GRPCHTTP2Core.SocketAddress {
-    precondition(self.server.withLockedValue { $0 } == nil)
+    precondition(self.server.withLock { $0 } == nil)
 
     @Sendable
     func configure(_ channel: any Channel) -> EventLoopFuture<Multiplexer> {
-      self.connected.withLockedValue {
+      self.connected.withLock {
         $0.append(channel)
       }
 
       channel.closeFuture.whenSuccess {
-        self.connected.withLockedValue { connected in
+        self.connected.withLock { connected in
           guard let index = connected.firstIndex(where: { $0 === channel }) else { return }
           connected.remove(at: index)
         }
@@ -112,12 +112,12 @@ final class TestServer: Sendable {
       address = .unixDomainSocket(path: server.channel.localAddress!.pathname!)
     }
 
-    self.server.withLockedValue { $0 = server }
+    self.server.withLock { $0 = server }
     return address
   }
 
   func run(_ handle: @Sendable @escaping (Inbound, Outbound) async throws -> Void) async throws {
-    guard let server = self.server.withLockedValue({ $0 }) else {
+    guard let server = self.server.withLock({ $0 }) else {
       fatalError("bind() must be called first")
     }
 
@@ -145,7 +145,7 @@ final class TestServer: Sendable {
   }
 }
 
-@available(macOS 13.0, iOS 16.0, watchOS 9.0, tvOS 16.0, *)
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 extension TestServer {
   enum RunHandler {
     case echo

Tests/GRPCHTTP2CoreTests/Internal/TimerTests.swift

@@ -18,30 +18,30 @@ import Atomics
 import GRPCCore
 import GRPCHTTP2Core
 import NIOEmbedded
+import Synchronization
 import XCTest
 
+@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *)
 internal final class TimerTests: XCTestCase {
   func testScheduleOneOffTimer() {
     let loop = EmbeddedEventLoop()
     defer { try! loop.close() }
 
-    let value = LockedValueBox(0)
+    let value = Atomic(0)
     var timer = Timer(delay: .seconds(1), repeat: false)
     timer.schedule(on: loop) {
-      value.withLockedValue {
-        XCTAssertEqual($0, 0)
-        $0 += 1
-      }
+      let (old, _) = value.add(1, ordering: .releasing)
+      XCTAssertEqual(old, 0)
     }
 
     loop.advanceTime(by: .milliseconds(999))
-    XCTAssertEqual(value.withLockedValue { $0 }, 0)
+    XCTAssertEqual(value.load(ordering: .acquiring), 0)
     loop.advanceTime(by: .milliseconds(1))
-    XCTAssertEqual(value.withLockedValue { $0 }, 1)
+    XCTAssertEqual(value.load(ordering: .acquiring), 1)
 
     // Run again to make sure the task wasn't repeated.
     loop.advanceTime(by: .seconds(1))
-    XCTAssertEqual(value.withLockedValue { $0 }, 1)
+    XCTAssertEqual(value.load(ordering: .acquiring), 1)
   }
 
   func testCancelOneOffTimer() {
@@ -62,25 +62,25 @@ internal final class TimerTests: XCTestCase {
     let loop = EmbeddedEventLoop()
     defer { try! loop.close() }
 
-    let values = LockedValueBox([Int]())
+    let value = Atomic(0)
     var timer = Timer(delay: .seconds(1), repeat: true)
     timer.schedule(on: loop) {
-      values.withLockedValue { $0.append($0.count) }
+      value.add(1, ordering: .releasing)
     }
 
     loop.advanceTime(by: .milliseconds(999))
-    XCTAssertEqual(values.withLockedValue { $0 }, [])
+    XCTAssertEqual(value.load(ordering: .acquiring), 0)
     loop.advanceTime(by: .milliseconds(1))
-    XCTAssertEqual(values.withLockedValue { $0 }, [0])
+    XCTAssertEqual(value.load(ordering: .acquiring), 1)
 
     loop.advanceTime(by: .seconds(1))
-    XCTAssertEqual(values.withLockedValue { $0 }, [0, 1])
+    XCTAssertEqual(value.load(ordering: .acquiring), 2)
     loop.advanceTime(by: .seconds(1))
-    XCTAssertEqual(values.withLockedValue { $0 }, [0, 1, 2])
+    XCTAssertEqual(value.load(ordering: .acquiring), 3)
 
     timer.cancel()
     loop.advanceTime(by: .seconds(1))
-    XCTAssertEqual(values.withLockedValue { $0 }, [0, 1, 2])
+    XCTAssertEqual(value.load(ordering: .acquiring), 3)
   }
 
   func testCancelRepeatedTimer() {

Tests/GRPCInProcessTransportTests/InProcessServerTransportTests.swift

@@ -39,20 +39,17 @@ final class InProcessServerTransportTests: XCTestCase {
       outbound: RPCWriter.Closable(wrapping: outbound.continuation)
     )
 
-    let messages = LockedValueBox<[RPCRequestPart]?>(nil)
     try await withThrowingTaskGroup(of: Void.self) { group in
       group.addTask {
         try await transport.listen { stream in
           let partValue = try? await stream.inbound.reduce(into: []) { $0.append($1) }
-          messages.withLockedValue { $0 = partValue }
+          XCTAssertEqual(partValue, [.message([42])])
           transport.stopListening()
         }
       }
 
       try transport.acceptStream(stream)
     }
-
-    XCTAssertEqual(messages.withLockedValue { $0 }, [.message([42])])
   }
 
   func testStopListening() async throws {