From f0204ecc3934e03cf9743688608b0341a8b8ecf6 Mon Sep 17 00:00:00 2001 From: George Barnett Date: Wed, 21 Aug 2024 10:59:43 +0100 Subject: [PATCH 1/6] Remove swift-atomics from v2 Motivation: Swift 6 includes built-in support for atomics in the Synchronization module. Currently we depend on the swift-atomics package which is no longer necessary. Modifications: Remove usages of swift-atomics from v2 Result: Fewer dependencies --- Package@swift-6.swift | 3 - Sources/GRPCCore/GRPCClient.swift | 148 ++++++++---------- Sources/GRPCCore/GRPCServer.swift | 106 ++++++------- .../Internal/AsyncIteratorSequence.swift | 8 +- .../Client/Connection/GRPCChannel.swift | 3 +- .../Internal/ProcessUniqueID.swift | 11 +- .../performance-worker/BenchmarkClient.swift | 16 +- .../performance-worker/BenchmarkService.swift | 30 +++- ...PCExecutorTestHarness+ServerBehavior.swift | 8 +- ...ientRPCExecutorTestHarness+Transport.swift | 2 +- .../ClientRPCExecutorTestHarness.swift | 2 +- .../Internal/ServerRPCExecutorTests.swift | 17 +- Tests/GRPCCoreTests/GRPCClientTests.swift | 10 +- Tests/GRPCCoreTests/GRPCServerTests.swift | 14 +- .../Test Utilities/AtomicCounter.swift | 35 +++++ .../Call/Client/ClientInterceptors.swift | 10 +- .../Call/Server/ServerInterceptors.swift | 10 +- .../Transport/StreamCountingTransport.swift | 22 +-- .../PickFirstLoadBalancerTests.swift | 17 +- .../RoundRobinLoadBalancerTests.swift | 13 +- .../Client/Connection/RequestQueueTests.swift | 2 +- .../Connection/Utilities/TestServer.swift | 2 +- .../Internal/ProcessUniqueIDTests.swift | 1 + .../Internal/TimerTests.swift | 17 +- .../Test Utilities/AtomicCounter.swift | 35 +++++ 25 files changed, 296 insertions(+), 246 deletions(-) create mode 100644 Tests/GRPCCoreTests/Test Utilities/AtomicCounter.swift create mode 100644 Tests/GRPCHTTP2CoreTests/Test Utilities/AtomicCounter.swift diff --git a/Package@swift-6.swift b/Package@swift-6.swift index 99206ce42..49f3f24a6 100644 --- a/Package@swift-6.swift +++ b/Package@swift-6.swift @@ -192,7 +192,6 @@ extension Target { name: "GRPCCore", dependencies: [ .dequeModule, - .atomics ], path: "Sources/GRPCCore", swiftSettings: [ @@ -242,7 +241,6 @@ extension Target { .nioTLS, .cgrpcZlib, .dequeModule, - .atomics ], swiftSettings: [ .swiftLanguageMode(.v6), @@ -401,7 +399,6 @@ extension Target { .grpcCore, .grpcInProcessTransport, .dequeModule, - .atomics, .protobuf, ], swiftSettings: [.swiftLanguageMode(.v6), .enableUpcomingFeature("ExistentialAny")] diff --git a/Sources/GRPCCore/GRPCClient.swift b/Sources/GRPCCore/GRPCClient.swift index e65699932..aa8b730f8 100644 --- a/Sources/GRPCCore/GRPCClient.swift +++ b/Sources/GRPCCore/GRPCClient.swift @@ -14,7 +14,7 @@ * limitations under the License. */ -internal import Atomics +private import Synchronization /// A gRPC client. /// @@ -110,7 +110,7 @@ internal import Atomics /// additional resources that need their lifecycles managed you should consider using [Swift Service /// Lifecycle](https://github.com/swift-server/swift-service-lifecycle). @available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *) -public struct GRPCClient: Sendable { +public final class GRPCClient: Sendable { /// The transport which provides a bidirectional communication channel with the server. private let transport: any ClientTransport @@ -123,10 +123,10 @@ public struct GRPCClient: Sendable { private let interceptors: [any ClientInterceptor] /// The current state of the client. - private let state: ManagedAtomic + private let state: Mutex /// The state of the client. - private enum State: UInt8, AtomicValue { + private enum State: Sendable { /// The client hasn't been started yet. Can transition to `running` or `stopped`. case notStarted /// The client is running and can send RPCs. Can transition to `stopping`. @@ -137,6 +137,56 @@ public struct GRPCClient: Sendable { /// The client has stopped, no RPCs are in flight and no more will be accepted. This state /// is terminal. case stopped + + mutating func run() throws { + switch self { + case .notStarted: + self = .running + + case .running: + throw RuntimeError( + code: .clientIsAlreadyRunning, + message: "The client is already running and can only be started once." + ) + + case .stopping, .stopped: + throw RuntimeError( + code: .clientIsStopped, + message: "The client has stopped and can only be started once." + ) + } + } + + mutating func stopped() { + self = .stopped + } + + mutating func beginGracefulShutdown() -> Bool { + switch self { + case .notStarted: + self = .stopped + return false + case .running: + self = .stopping + return true + case .stopping, .stopped: + return false + } + } + + func checkExecutable() throws { + switch self { + case .notStarted, .running: + // Allow .notStarted as making a request can race with 'run()'. Transports should tolerate + // queuing the request if not yet started. + () + case .stopping, .stopped: + throw RuntimeError( + code: .clientIsStopped, + message: "Client has been stopped. Can't make any more RPCs." + ) + } + } } /// Creates a new client with the given transport, interceptors and configuration. @@ -154,7 +204,7 @@ public struct GRPCClient: Sendable { ) { self.transport = transport self.interceptors = interceptors - self.state = ManagedAtomic(.notStarted) + self.state = Mutex(.notStarted) } /// Start the client. @@ -165,33 +215,11 @@ public struct GRPCClient: Sendable { /// The client, and by extension this function, can only be run once. If the client is already /// running or has already been closed then a ``RuntimeError`` is thrown. public func run() async throws { - let (wasNotStarted, original) = self.state.compareExchange( - expected: .notStarted, - desired: .running, - ordering: .sequentiallyConsistent - ) - - guard wasNotStarted else { - switch original { - case .notStarted: - // The value wasn't exchanged so the original value can't be 'notStarted'. - fatalError() - case .running: - throw RuntimeError( - code: .clientIsAlreadyRunning, - message: "The client is already running and can only be started once." - ) - case .stopping, .stopped: - throw RuntimeError( - code: .clientIsStopped, - message: "The client has stopped and can only be started once." - ) - } - } + try self.state.withLock { try $0.run() } - // When we exit this function we must have stopped. + // When this function exits the client must have stopped. defer { - self.state.store(.stopped, ordering: .sequentiallyConsistent) + self.state.withLock { $0.stopped() } } do { @@ -211,50 +239,9 @@ public struct GRPCClient: Sendable { /// in-flight RPCs to finish executing, but no new RPCs will be accepted. You can cancel the task /// executing ``run()`` if you want to abruptly stop in-flight RPCs. public func close() { - while true { - let (wasRunning, actualState) = self.state.compareExchange( - expected: .running, - desired: .stopping, - ordering: .sequentiallyConsistent - ) - - // Transition from running to stopping: close the transport. - if wasRunning { - self.transport.close() - return - } - - // The expected state wasn't 'running'. There are two options: - // 1. The client isn't running yet. - // 2. The client is already stopping or stopped. - switch actualState { - case .notStarted: - // Not started: try going straight to stopped. - let (wasNotStarted, _) = self.state.compareExchange( - expected: .notStarted, - desired: .stopped, - ordering: .sequentiallyConsistent - ) - - // If the exchange happened then just return: the client wasn't started so there's no - // transport to start. - // - // If the exchange didn't happen then continue looping: the client must've been started by - // another thread. - if wasNotStarted { - return - } else { - continue - } - - case .running: - // Unreachable: the value was exchanged and this was the expected value. - fatalError() - - case .stopping, .stopped: - // No exchange happened but the client is already stopping. - return - } + let wasRunning = self.state.withLock { $0.beginGracefulShutdown() } + if wasRunning { + self.transport.close() } } @@ -371,18 +358,7 @@ public struct GRPCClient: Sendable { options: CallOptions, handler: @Sendable @escaping (ClientResponse.Stream) async throws -> ReturnValue ) async throws -> ReturnValue { - switch self.state.load(ordering: .sequentiallyConsistent) { - case .notStarted, .running: - // Allow .notStarted as making a request can race with 'run()'. Transports should tolerate - // queuing the request if not yet started. - () - case .stopping, .stopped: - throw RuntimeError( - code: .clientIsStopped, - message: "Client has been stopped. Can't make any more RPCs." - ) - } - + try self.state.withLock { try $0.checkExecutable() } let methodConfig = self.transport.configuration(forMethod: descriptor) var options = options options.formUnion(with: methodConfig) diff --git a/Sources/GRPCCore/GRPCServer.swift b/Sources/GRPCCore/GRPCServer.swift index c4e2534a5..8ff31129e 100644 --- a/Sources/GRPCCore/GRPCServer.swift +++ b/Sources/GRPCCore/GRPCServer.swift @@ -14,7 +14,7 @@ * limitations under the License. */ -internal import Atomics +private import Synchronization /// A gRPC server. /// @@ -70,7 +70,7 @@ internal import Atomics /// that need their lifecycles managed you should consider using [Swift Service /// Lifecycle](https://github.com/swift-server/swift-service-lifecycle). @available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *) -public struct GRPCServer: Sendable { +public final class GRPCServer: Sendable { typealias Stream = RPCStream /// The ``ServerTransport`` implementation that the server uses to listen for new requests. @@ -88,9 +88,9 @@ public struct GRPCServer: Sendable { private let interceptors: [any ServerInterceptor] /// The state of the server. - private let state: ManagedAtomic + private let state: Mutex - private enum State: UInt8, AtomicValue { + private enum State: Sendable { /// The server hasn't been started yet. Can transition to `running` or `stopped`. case notStarted /// The server is running and accepting RPCs. Can transition to `stopping`. @@ -101,6 +101,43 @@ public struct GRPCServer: Sendable { /// The server has stopped, no RPCs are in flight and no more will be accepted. This state /// is terminal. case stopped + + mutating func startServing() throws { + switch self { + case .notStarted: + self = .running + + case .running: + throw RuntimeError( + code: .serverIsAlreadyRunning, + message: "The server is already running and can only be started once." + ) + + case .stopping, .stopped: + throw RuntimeError( + code: .serverIsStopped, + message: "The server has stopped and can only be started once." + ) + } + } + + mutating func beginGracefulShutdown() -> Bool { + switch self { + case .notStarted: + self = .stopped + return false + case .running: + self = .stopping + return true + case .stopping, .stopped: + // Already stopping/stopped, ignore. + return false + } + } + + mutating func stopped() { + self = .stopped + } } /// Creates a new server with no resources. @@ -113,7 +150,7 @@ public struct GRPCServer: Sendable { /// are called. The first interceptor added will be the first interceptor to intercept each /// request. The last interceptor added will be the final interceptor to intercept each /// request before calling the appropriate handler. - public init( + public convenience init( transport: any ServerTransport, services: [any RegistrableRPCService], interceptors: [any ServerInterceptor] = [] @@ -141,7 +178,7 @@ public struct GRPCServer: Sendable { router: RPCRouter, interceptors: [any ServerInterceptor] = [] ) { - self.state = ManagedAtomic(.notStarted) + self.state = Mutex(.notStarted) self.transport = transport self.router = router self.interceptors = interceptors @@ -161,33 +198,11 @@ public struct GRPCServer: Sendable { /// - Note: You can only call this function once, repeated calls will result in a /// ``RuntimeError`` being thrown. public func run() async throws { - let (wasNotStarted, actualState) = self.state.compareExchange( - expected: .notStarted, - desired: .running, - ordering: .sequentiallyConsistent - ) - - guard wasNotStarted else { - switch actualState { - case .notStarted: - fatalError() - case .running: - throw RuntimeError( - code: .serverIsAlreadyRunning, - message: "The server is already running and can only be started once." - ) + try self.state.withLock { try $0.startServing() } - case .stopping, .stopped: - throw RuntimeError( - code: .serverIsStopped, - message: "The server has stopped and can only be started once." - ) - } - } - - // When we exit this function we must have stopped. + // When we exit this function the server must have stopped. defer { - self.state.store(.stopped, ordering: .sequentiallyConsistent) + self.state.withLock { $0.stopped() } } do { @@ -210,36 +225,9 @@ public struct GRPCServer: Sendable { /// /// Calling this on a server which is already stopping or has stopped has no effect. public func stopListening() { - let (wasRunning, actual) = self.state.compareExchange( - expected: .running, - desired: .stopping, - ordering: .sequentiallyConsistent - ) - + let wasRunning = self.state.withLock { $0.beginGracefulShutdown() } if wasRunning { self.transport.stopListening() - } else { - switch actual { - case .notStarted: - let (exchanged, _) = self.state.compareExchange( - expected: .notStarted, - desired: .stopped, - ordering: .sequentiallyConsistent - ) - - // Lost a race with 'run()', try again. - if !exchanged { - self.stopListening() - } - - case .running: - // Unreachable, this branch only happens when the initial exchange didn't take place. - fatalError() - - case .stopping, .stopped: - // Already stopping/stopped, ignore. - () - } } } } diff --git a/Sources/GRPCCore/Streaming/Internal/AsyncIteratorSequence.swift b/Sources/GRPCCore/Streaming/Internal/AsyncIteratorSequence.swift index 05596c6d4..f75421dd6 100644 --- a/Sources/GRPCCore/Streaming/Internal/AsyncIteratorSequence.swift +++ b/Sources/GRPCCore/Streaming/Internal/AsyncIteratorSequence.swift @@ -14,12 +14,12 @@ * limitations under the License. */ -public import Atomics // should be @usableFromInline +public import Synchronization // should be @usableFromInline -@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *) +@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *) @usableFromInline /// An `AsyncSequence` which wraps an existing async iterator. -struct AsyncIteratorSequence: AsyncSequence { +final class AsyncIteratorSequence: AsyncSequence { @usableFromInline typealias Element = Base.Element @@ -29,7 +29,7 @@ struct AsyncIteratorSequence: AsyncSequence { /// Set to `true` when an iterator has been made. @usableFromInline - let _hasMadeIterator = ManagedAtomic(false) + let _hasMadeIterator = Atomic(false) @inlinable init(_ base: Base) { diff --git a/Sources/GRPCHTTP2Core/Client/Connection/GRPCChannel.swift b/Sources/GRPCHTTP2Core/Client/Connection/GRPCChannel.swift index 00fde9312..6942ee213 100644 --- a/Sources/GRPCHTTP2Core/Client/Connection/GRPCChannel.swift +++ b/Sources/GRPCHTTP2Core/Client/Connection/GRPCChannel.swift @@ -14,8 +14,7 @@ * limitations under the License. */ -internal import Atomics -internal import DequeModule +private import DequeModule package import GRPCCore private import Synchronization diff --git a/Sources/GRPCHTTP2Core/Internal/ProcessUniqueID.swift b/Sources/GRPCHTTP2Core/Internal/ProcessUniqueID.swift index d80658e00..5f6f32f7e 100644 --- a/Sources/GRPCHTTP2Core/Internal/ProcessUniqueID.swift +++ b/Sources/GRPCHTTP2Core/Internal/ProcessUniqueID.swift @@ -14,15 +14,17 @@ * limitations under the License. */ -internal import Atomics +private import Synchronization /// An ID which is unique within this process. +@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *) struct ProcessUniqueID: Hashable, Sendable, CustomStringConvertible { - private static let source = ManagedAtomic(UInt64(0)) + private static let source = Atomic(UInt64(0)) private let rawValue: UInt64 init() { - self.rawValue = Self.source.loadThenWrappingIncrement(ordering: .relaxed) + let (_, newValue) = Self.source.add(1, ordering: .relaxed) + self.rawValue = newValue } var description: String { @@ -31,6 +33,7 @@ struct ProcessUniqueID: Hashable, Sendable, CustomStringConvertible { } /// A process-unique ID for a subchannel. +@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *) package struct SubchannelID: Hashable, Sendable, CustomStringConvertible { private let id = ProcessUniqueID() package init() {} @@ -40,6 +43,7 @@ package struct SubchannelID: Hashable, Sendable, CustomStringConvertible { } /// A process-unique ID for a load-balancer. +@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *) struct LoadBalancerID: Hashable, Sendable, CustomStringConvertible { private let id = ProcessUniqueID() var description: String { @@ -48,6 +52,7 @@ struct LoadBalancerID: Hashable, Sendable, CustomStringConvertible { } /// A process-unique ID for an entry in a queue. +@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *) struct QueueEntryID: Hashable, Sendable, CustomStringConvertible { private let id = ProcessUniqueID() var description: String { diff --git a/Sources/performance-worker/BenchmarkClient.swift b/Sources/performance-worker/BenchmarkClient.swift index 0541cc4cb..5081bd2ab 100644 --- a/Sources/performance-worker/BenchmarkClient.swift +++ b/Sources/performance-worker/BenchmarkClient.swift @@ -14,14 +14,14 @@ * limitations under the License. */ -private import Atomics private import Foundation internal import GRPCCore private import NIOConcurrencyHelpers +private import Synchronization @available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *) -struct BenchmarkClient { - private let _isShuttingDown = ManagedAtomic(false) +final class BenchmarkClient: Sendable { + private let _isShuttingDown = Atomic(false) /// Whether the benchmark client is shutting down. Used to control when to stop sending messages /// or creating new RPCs. @@ -30,17 +30,17 @@ struct BenchmarkClient { } /// The underlying client. - private var client: GRPCClient + private let client: GRPCClient /// The number of concurrent RPCs to run. - private var concurrentRPCs: Int + private let concurrentRPCs: Int /// The type of RPC to make against the server. - private var rpcType: RPCType + private let rpcType: RPCType /// The max number of messages to send on a stream before replacing the RPC with a new one. A /// value of zero means there is no limit. - private var messagesPerStream: Int + private let messagesPerStream: Int private var noMessageLimit: Bool { self.messagesPerStream == 0 } /// The message to send for all RPC types to the server. @@ -206,7 +206,7 @@ struct BenchmarkClient { let nanos = now.uptimeNanoseconds - lastMessageSendTime.uptimeNanoseconds lastMessageSendTime = now self.record(latencyNanos: Double(nanos), errorCode: nil) - try await writer.write(message) + try await writer.write(self.message) } else { break } diff --git a/Sources/performance-worker/BenchmarkService.swift b/Sources/performance-worker/BenchmarkService.swift index 694ad51b7..c90eb2fd7 100644 --- a/Sources/performance-worker/BenchmarkService.swift +++ b/Sources/performance-worker/BenchmarkService.swift @@ -14,15 +14,15 @@ * limitations under the License. */ -private import Atomics internal import GRPCCore +private import Synchronization import struct Foundation.Data @available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *) -struct BenchmarkService: Grpc_Testing_BenchmarkService.ServiceProtocol { +final class BenchmarkService: Grpc_Testing_BenchmarkService.ServiceProtocol { /// Used to check if the server can be streaming responses. - private let working = ManagedAtomic(true) + private let working = Atomic(true) /// One request followed by one response. /// The server returns a client payload with the size requested by the client. @@ -127,8 +127,24 @@ struct BenchmarkService: Grpc_Testing_BenchmarkService.ServiceProtocol { } } + final class InboundStreamingSignal: Sendable { + private let _isStreaming: Atomic + + init() { + self._isStreaming = Atomic(true) + } + + var isStreaming: Bool { + self._isStreaming.load(ordering: .relaxed) + } + + func stop() { + self._isStreaming.store(false, ordering: .relaxed) + } + } + // Marks if the inbound streaming is ongoing or finished. - let inboundStreaming = ManagedAtomic(true) + let inbound = InboundStreamingSignal() return ServerResponse.Stream { writer in try await withThrowingTaskGroup(of: Void.self) { group in @@ -138,13 +154,11 @@ struct BenchmarkService: Grpc_Testing_BenchmarkService.ServiceProtocol { try self.checkOkStatus(message.responseStatus) } } - inboundStreaming.store(false, ordering: .relaxed) + inbound.stop() } group.addTask { - while inboundStreaming.load(ordering: .relaxed) - && self.working.load(ordering: .acquiring) - { + while inbound.isStreaming && self.working.load(ordering: .acquiring) { try await writer.write(response) } } diff --git a/Tests/GRPCCoreTests/Call/Client/Internal/ClientRPCExecutorTestSupport/ClientRPCExecutorTestHarness+ServerBehavior.swift b/Tests/GRPCCoreTests/Call/Client/Internal/ClientRPCExecutorTestSupport/ClientRPCExecutorTestHarness+ServerBehavior.swift index a2bed0d7d..4f0ab1b14 100644 --- a/Tests/GRPCCoreTests/Call/Client/Internal/ClientRPCExecutorTestSupport/ClientRPCExecutorTestHarness+ServerBehavior.swift +++ b/Tests/GRPCCoreTests/Call/Client/Internal/ClientRPCExecutorTestSupport/ClientRPCExecutorTestHarness+ServerBehavior.swift @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -import Atomics + import XCTest @testable import GRPCCore @@ -104,10 +104,10 @@ extension ClientRPCExecutorTestHarness.ServerStreamHandler { } static func attemptBased(_ onAttempt: @Sendable @escaping (_ attempt: Int) -> Self) -> Self { - let attempts = ManagedAtomic(1) + let attempts = AtomicCounter(1) return Self { stream in - let attempt = attempts.loadThenWrappingIncrement(ordering: .sequentiallyConsistent) - let handler = onAttempt(attempt) + let (oldAttemptCount, _) = attempts.increment() + let handler = onAttempt(oldAttemptCount) try await handler.handle(stream: stream) } } diff --git a/Tests/GRPCCoreTests/Call/Client/Internal/ClientRPCExecutorTestSupport/ClientRPCExecutorTestHarness+Transport.swift b/Tests/GRPCCoreTests/Call/Client/Internal/ClientRPCExecutorTestSupport/ClientRPCExecutorTestHarness+Transport.swift index 30f0e1000..850b006d5 100644 --- a/Tests/GRPCCoreTests/Call/Client/Internal/ClientRPCExecutorTestSupport/ClientRPCExecutorTestHarness+Transport.swift +++ b/Tests/GRPCCoreTests/Call/Client/Internal/ClientRPCExecutorTestSupport/ClientRPCExecutorTestHarness+Transport.swift @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -import Atomics + import GRPCCore import GRPCInProcessTransport diff --git a/Tests/GRPCCoreTests/Call/Client/Internal/ClientRPCExecutorTestSupport/ClientRPCExecutorTestHarness.swift b/Tests/GRPCCoreTests/Call/Client/Internal/ClientRPCExecutorTestSupport/ClientRPCExecutorTestHarness.swift index 9a23448ac..e3f089893 100644 --- a/Tests/GRPCCoreTests/Call/Client/Internal/ClientRPCExecutorTestSupport/ClientRPCExecutorTestHarness.swift +++ b/Tests/GRPCCoreTests/Call/Client/Internal/ClientRPCExecutorTestSupport/ClientRPCExecutorTestHarness.swift @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -import Atomics + import GRPCInProcessTransport import XCTest diff --git a/Tests/GRPCCoreTests/Call/Server/Internal/ServerRPCExecutorTests.swift b/Tests/GRPCCoreTests/Call/Server/Internal/ServerRPCExecutorTests.swift index e552991fa..0393de2c9 100644 --- a/Tests/GRPCCoreTests/Call/Server/Internal/ServerRPCExecutorTests.swift +++ b/Tests/GRPCCoreTests/Call/Server/Internal/ServerRPCExecutorTests.swift @@ -14,7 +14,6 @@ * limitations under the License. */ -import Atomics import XCTest @testable import GRPCCore @@ -290,8 +289,8 @@ final class ServerRPCExecutorTests: XCTestCase { } func testMultipleInterceptorsAreCalled() async throws { - let counter1 = ManagedAtomic(0) - let counter2 = ManagedAtomic(0) + let counter1 = AtomicCounter() + let counter2 = AtomicCounter() // The interceptor skips the handler altogether. let harness = ServerRPCExecutorTestHarness( @@ -309,13 +308,13 @@ final class ServerRPCExecutorTests: XCTestCase { XCTAssertEqual(parts, [.metadata([:]), .status(.ok, [:])]) } - XCTAssertEqual(counter1.load(ordering: .sequentiallyConsistent), 1) - XCTAssertEqual(counter2.load(ordering: .sequentiallyConsistent), 1) + XCTAssertEqual(counter1.value, 1) + XCTAssertEqual(counter2.value, 1) } func testInterceptorsAreCalledInOrder() async throws { - let counter1 = ManagedAtomic(0) - let counter2 = ManagedAtomic(0) + let counter1 = AtomicCounter() + let counter2 = AtomicCounter() // The interceptor skips the handler altogether. let harness = ServerRPCExecutorTestHarness( @@ -334,9 +333,9 @@ final class ServerRPCExecutorTests: XCTestCase { XCTAssertEqual(parts, [.status(Status(code: .unavailable, message: ""), [:])]) } - XCTAssertEqual(counter1.load(ordering: .sequentiallyConsistent), 1) + XCTAssertEqual(counter1.value, 1) // Zero because the RPC should've been rejected by the second interceptor. - XCTAssertEqual(counter2.load(ordering: .sequentiallyConsistent), 0) + XCTAssertEqual(counter2.value, 0) } func testThrowingInterceptor() async throws { diff --git a/Tests/GRPCCoreTests/GRPCClientTests.swift b/Tests/GRPCCoreTests/GRPCClientTests.swift index f360936b4..15363c35b 100644 --- a/Tests/GRPCCoreTests/GRPCClientTests.swift +++ b/Tests/GRPCCoreTests/GRPCClientTests.swift @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -import Atomics + import GRPCCore import GRPCInProcessTransport import XCTest @@ -230,8 +230,8 @@ final class GRPCClientTests: XCTestCase { } func testInterceptorsAreAppliedInOrder() async throws { - let counter1 = ManagedAtomic(0) - let counter2 = ManagedAtomic(0) + let counter1 = AtomicCounter() + let counter2 = AtomicCounter() try await self.withInProcessConnectedClient( services: [BinaryEcho()], @@ -254,8 +254,8 @@ final class GRPCClientTests: XCTestCase { } } - XCTAssertEqual(counter1.load(ordering: .sequentiallyConsistent), 1) - XCTAssertEqual(counter2.load(ordering: .sequentiallyConsistent), 0) + XCTAssertEqual(counter1.value, 1) + XCTAssertEqual(counter2.value, 0) } func testNoNewRPCsAfterClientClose() async throws { diff --git a/Tests/GRPCCoreTests/GRPCServerTests.swift b/Tests/GRPCCoreTests/GRPCServerTests.swift index 6c66d7ca0..c67c16084 100644 --- a/Tests/GRPCCoreTests/GRPCServerTests.swift +++ b/Tests/GRPCCoreTests/GRPCServerTests.swift @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -import Atomics + import GRPCCore import GRPCInProcessTransport import XCTest @@ -215,8 +215,8 @@ final class GRPCServerTests: XCTestCase { } func testInterceptorsAreAppliedInOrder() async throws { - let counter1 = ManagedAtomic(0) - let counter2 = ManagedAtomic(0) + let counter1 = AtomicCounter() + let counter2 = AtomicCounter() try await self.withInProcessClientConnectedToServer( services: [BinaryEcho()], @@ -240,12 +240,12 @@ final class GRPCServerTests: XCTestCase { } } - XCTAssertEqual(counter1.load(ordering: .sequentiallyConsistent), 1) - XCTAssertEqual(counter2.load(ordering: .sequentiallyConsistent), 0) + XCTAssertEqual(counter1.value, 1) + XCTAssertEqual(counter2.value, 0) } func testInterceptorsAreNotAppliedToUnimplementedMethods() async throws { - let counter = ManagedAtomic(0) + let counter = AtomicCounter() try await self.withInProcessClientConnectedToServer( services: [BinaryEcho()], @@ -265,7 +265,7 @@ final class GRPCServerTests: XCTestCase { } } - XCTAssertEqual(counter.load(ordering: .sequentiallyConsistent), 0) + XCTAssertEqual(counter.value, 0) } func testNoNewRPCsAfterServerStopListening() async throws { diff --git a/Tests/GRPCCoreTests/Test Utilities/AtomicCounter.swift b/Tests/GRPCCoreTests/Test Utilities/AtomicCounter.swift new file mode 100644 index 000000000..7aae0bb5a --- /dev/null +++ b/Tests/GRPCCoreTests/Test Utilities/AtomicCounter.swift @@ -0,0 +1,35 @@ +/* + * Copyright 2024, gRPC Authors All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +private import Synchronization + +@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *) +final class AtomicCounter: Sendable { + private let counter: Atomic + + init(_ initialValue: Int = 0) { + self.counter = Atomic(initialValue) + } + + var value: Int { + self.counter.load(ordering: .sequentiallyConsistent) + } + + @discardableResult + func increment() -> (oldValue: Int, newValue: Int) { + self.counter.add(1, ordering: .sequentiallyConsistent) + } +} diff --git a/Tests/GRPCCoreTests/Test Utilities/Call/Client/ClientInterceptors.swift b/Tests/GRPCCoreTests/Test Utilities/Call/Client/ClientInterceptors.swift index 1b35a0704..d8d355969 100644 --- a/Tests/GRPCCoreTests/Test Utilities/Call/Client/ClientInterceptors.swift +++ b/Tests/GRPCCoreTests/Test Utilities/Call/Client/ClientInterceptors.swift @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -import Atomics + import GRPCCore @available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *) @@ -30,7 +30,7 @@ extension ClientInterceptor where Self == RejectAllClientInterceptor { @available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *) extension ClientInterceptor where Self == RequestCountingClientInterceptor { - static func requestCounter(_ counter: ManagedAtomic) -> Self { + static func requestCounter(_ counter: AtomicCounter) -> Self { return RequestCountingClientInterceptor(counter: counter) } } @@ -68,9 +68,9 @@ struct RejectAllClientInterceptor: ClientInterceptor { @available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *) struct RequestCountingClientInterceptor: ClientInterceptor { /// The number of requests made. - let counter: ManagedAtomic + let counter: AtomicCounter - init(counter: ManagedAtomic) { + init(counter: AtomicCounter) { self.counter = counter } @@ -82,7 +82,7 @@ struct RequestCountingClientInterceptor: ClientInterceptor { ClientInterceptorContext ) async throws -> ClientResponse.Stream ) async throws -> ClientResponse.Stream { - self.counter.wrappingIncrement(ordering: .sequentiallyConsistent) + self.counter.increment() return try await next(request, context) } } diff --git a/Tests/GRPCCoreTests/Test Utilities/Call/Server/ServerInterceptors.swift b/Tests/GRPCCoreTests/Test Utilities/Call/Server/ServerInterceptors.swift index c70970eed..97c68aba9 100644 --- a/Tests/GRPCCoreTests/Test Utilities/Call/Server/ServerInterceptors.swift +++ b/Tests/GRPCCoreTests/Test Utilities/Call/Server/ServerInterceptors.swift @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -import Atomics + import GRPCCore @available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *) @@ -29,7 +29,7 @@ extension ServerInterceptor where Self == RejectAllServerInterceptor { @available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *) extension ServerInterceptor where Self == RequestCountingServerInterceptor { - static func requestCounter(_ counter: ManagedAtomic) -> Self { + static func requestCounter(_ counter: AtomicCounter) -> Self { return RequestCountingServerInterceptor(counter: counter) } } @@ -67,9 +67,9 @@ struct RejectAllServerInterceptor: ServerInterceptor { @available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *) struct RequestCountingServerInterceptor: ServerInterceptor { /// The number of requests made. - let counter: ManagedAtomic + let counter: AtomicCounter - init(counter: ManagedAtomic) { + init(counter: AtomicCounter) { self.counter = counter } @@ -81,7 +81,7 @@ struct RequestCountingServerInterceptor: ServerInterceptor { ServerInterceptorContext ) async throws -> ServerResponse.Stream ) async throws -> ServerResponse.Stream { - self.counter.wrappingIncrement(ordering: .sequentiallyConsistent) + self.counter.increment() return try await next(request, context) } } diff --git a/Tests/GRPCCoreTests/Test Utilities/Transport/StreamCountingTransport.swift b/Tests/GRPCCoreTests/Test Utilities/Transport/StreamCountingTransport.swift index 28111bd15..cc634b1a4 100644 --- a/Tests/GRPCCoreTests/Test Utilities/Transport/StreamCountingTransport.swift +++ b/Tests/GRPCCoreTests/Test Utilities/Transport/StreamCountingTransport.swift @@ -13,7 +13,6 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -import Atomics @testable import GRPCCore @@ -23,20 +22,22 @@ struct StreamCountingClientTransport: ClientTransport, Sendable { typealias Outbound = RPCWriter.Closable private let transport: AnyClientTransport - private let _streamsOpened = ManagedAtomic(0) - private let _streamFailures = ManagedAtomic(0) + private let _streamsOpened: AtomicCounter + private let _streamFailures: AtomicCounter var streamsOpened: Int { - self._streamsOpened.load(ordering: .sequentiallyConsistent) + self._streamsOpened.value } var streamFailures: Int { - self._streamFailures.load(ordering: .sequentiallyConsistent) + self._streamFailures.value } init(wrapping transport: Transport) where Transport.Inbound == Inbound, Transport.Outbound == Outbound { self.transport = AnyClientTransport(wrapping: transport) + self._streamsOpened = AtomicCounter() + self._streamFailures = AtomicCounter() } var retryThrottle: RetryThrottle? { @@ -61,11 +62,11 @@ struct StreamCountingClientTransport: ClientTransport, Sendable { descriptor: descriptor, options: options ) { stream in - self._streamsOpened.wrappingIncrement(ordering: .sequentiallyConsistent) + self._streamsOpened.increment() return try await closure(stream) } } catch { - self._streamFailures.wrappingIncrement(ordering: .sequentiallyConsistent) + self._streamFailures.increment() throw error } } @@ -83,21 +84,22 @@ struct StreamCountingServerTransport: ServerTransport, Sendable { typealias Outbound = RPCWriter.Closable private let transport: AnyServerTransport - private let _acceptedStreams = ManagedAtomic(0) + private let _acceptedStreams: AtomicCounter var acceptedStreamsCount: Int { - self._acceptedStreams.load(ordering: .sequentiallyConsistent) + self._acceptedStreams.value } init(wrapping transport: Transport) { self.transport = AnyServerTransport(wrapping: transport) + self._acceptedStreams = AtomicCounter() } func listen( _ streamHandler: @escaping @Sendable (RPCStream) async -> Void ) async throws { try await self.transport.listen { stream in - self._acceptedStreams.wrappingIncrement(ordering: .sequentiallyConsistent) + self._acceptedStreams.increment() await streamHandler(stream) } } diff --git a/Tests/GRPCHTTP2CoreTests/Client/Connection/LoadBalancers/PickFirstLoadBalancerTests.swift b/Tests/GRPCHTTP2CoreTests/Client/Connection/LoadBalancers/PickFirstLoadBalancerTests.swift index cc26898dd..29764adb5 100644 --- a/Tests/GRPCHTTP2CoreTests/Client/Connection/LoadBalancers/PickFirstLoadBalancerTests.swift +++ b/Tests/GRPCHTTP2CoreTests/Client/Connection/LoadBalancers/PickFirstLoadBalancerTests.swift @@ -14,7 +14,6 @@ * limitations under the License. */ -import Atomics import GRPCCore import GRPCHTTP2Core import NIOHTTP2 @@ -194,7 +193,7 @@ final class PickFirstLoadBalancerTests: XCTestCase { func testPickOnIdleTriggersConnect() async throws { // Tests that picking a subchannel when the load balancer is idle triggers a reconnect and // becomes ready again. Uses a very short idle time to re-enter the idle state. - let idle = ManagedAtomic(0) + let idle = AtomicCounter() try await LoadBalancerTest.pickFirst( servers: 1, @@ -202,7 +201,7 @@ final class PickFirstLoadBalancerTests: XCTestCase { ) { context, event in switch event { case .connectivityStateChanged(.idle): - let idleCount = idle.wrappingIncrementThenLoad(ordering: .sequentiallyConsistent) + let (_, idleCount) = idle.increment() switch idleCount { case 1: @@ -242,12 +241,13 @@ final class PickFirstLoadBalancerTests: XCTestCase { func testPickFirstConnectionDropReturnsToIdle() async throws { // Checks that when the load balancers connection is unexpectedly dropped when there are no // open streams that it returns to the idle state. - let idleCount = ManagedAtomic(0) + let idleCount = AtomicCounter() try await LoadBalancerTest.pickFirst(servers: 1, connector: .posix()) { context, event in switch event { case .connectivityStateChanged(.idle): - switch idleCount.wrappingIncrementThenLoad(ordering: .sequentiallyConsistent) { + let (_, newIdleCount) = idleCount.increment() + switch newIdleCount { case 1: let endpoint = Endpoint(addresses: context.servers.map { $0.address }) context.pickFirst!.updateEndpoint(endpoint) @@ -277,11 +277,12 @@ final class PickFirstLoadBalancerTests: XCTestCase { } func testPickFirstReceivesGoAway() async throws { - let idleCount = ManagedAtomic(0) + let idleCount = AtomicCounter() try await LoadBalancerTest.pickFirst(servers: 2, connector: .posix()) { context, event in switch event { case .connectivityStateChanged(.idle): - switch idleCount.wrappingIncrementThenLoad(ordering: .sequentiallyConsistent) { + let (_, newIdleCount) = idleCount.increment() + switch newIdleCount { case 1: // Provide the address of the first server. context.pickFirst!.updateEndpoint(Endpoint(context.servers[0].address)) @@ -293,7 +294,7 @@ final class PickFirstLoadBalancerTests: XCTestCase { } case .connectivityStateChanged(.ready): - switch idleCount.load(ordering: .sequentiallyConsistent) { + switch idleCount.value { case 1: // Must be connected to server 1, send a GOAWAY frame. let channel = context.servers[0].server.clients.first! diff --git a/Tests/GRPCHTTP2CoreTests/Client/Connection/LoadBalancers/RoundRobinLoadBalancerTests.swift b/Tests/GRPCHTTP2CoreTests/Client/Connection/LoadBalancers/RoundRobinLoadBalancerTests.swift index db15b52c8..b53e038b7 100644 --- a/Tests/GRPCHTTP2CoreTests/Client/Connection/LoadBalancers/RoundRobinLoadBalancerTests.swift +++ b/Tests/GRPCHTTP2CoreTests/Client/Connection/LoadBalancers/RoundRobinLoadBalancerTests.swift @@ -14,7 +14,6 @@ * limitations under the License. */ -import Atomics import GRPCCore import GRPCHTTP2Core import NIOHTTP2 @@ -322,8 +321,8 @@ final class RoundRobinLoadBalancerTests: XCTestCase { } func testPickOnIdleLoadBalancerTriggersConnect() async throws { - let idle = ManagedAtomic(0) - let ready = ManagedAtomic(0) + let idle = AtomicCounter() + let ready = AtomicCounter() try await LoadBalancerTest.roundRobin( servers: 1, @@ -331,9 +330,9 @@ final class RoundRobinLoadBalancerTests: XCTestCase { ) { context, event in switch event { case .connectivityStateChanged(.idle): - let idleCount = idle.wrappingIncrementThenLoad(ordering: .sequentiallyConsistent) + let (_, newIdleCount) = idle.increment() - switch idleCount { + switch newIdleCount { case 1: // The first idle happens when the load balancer in started, give it a set of addresses // which it will connect to. Wait for it to be ready and then idle again. @@ -354,9 +353,9 @@ final class RoundRobinLoadBalancerTests: XCTestCase { } case .connectivityStateChanged(.ready): - let readyCount = ready.wrappingIncrementThenLoad(ordering: .sequentiallyConsistent) + let (_, newReadyCount) = ready.increment() - if readyCount == 2 { + if newReadyCount == 2 { XCTAssertNotNil(context.loadBalancer.pickSubchannel()) } diff --git a/Tests/GRPCHTTP2CoreTests/Client/Connection/RequestQueueTests.swift b/Tests/GRPCHTTP2CoreTests/Client/Connection/RequestQueueTests.swift index 31b046571..23bec0dab 100644 --- a/Tests/GRPCHTTP2CoreTests/Client/Connection/RequestQueueTests.swift +++ b/Tests/GRPCHTTP2CoreTests/Client/Connection/RequestQueueTests.swift @@ -15,7 +15,7 @@ */ import GRPCCore -import Synchronization +private import Synchronization import XCTest @testable import GRPCHTTP2Core diff --git a/Tests/GRPCHTTP2CoreTests/Client/Connection/Utilities/TestServer.swift b/Tests/GRPCHTTP2CoreTests/Client/Connection/Utilities/TestServer.swift index 4063a06e9..642a495a1 100644 --- a/Tests/GRPCHTTP2CoreTests/Client/Connection/Utilities/TestServer.swift +++ b/Tests/GRPCHTTP2CoreTests/Client/Connection/Utilities/TestServer.swift @@ -18,7 +18,7 @@ import GRPCCore import NIOCore import NIOHTTP2 import NIOPosix -import Synchronization +private import Synchronization import XCTest @testable import GRPCHTTP2Core diff --git a/Tests/GRPCHTTP2CoreTests/Internal/ProcessUniqueIDTests.swift b/Tests/GRPCHTTP2CoreTests/Internal/ProcessUniqueIDTests.swift index d510031e3..180bb030f 100644 --- a/Tests/GRPCHTTP2CoreTests/Internal/ProcessUniqueIDTests.swift +++ b/Tests/GRPCHTTP2CoreTests/Internal/ProcessUniqueIDTests.swift @@ -18,6 +18,7 @@ import XCTest @testable import GRPCHTTP2Core +@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *) final class ProcessUniqueIDTests: XCTestCase { func testProcessUniqueIDIsUnique() { var ids: Set = [] diff --git a/Tests/GRPCHTTP2CoreTests/Internal/TimerTests.swift b/Tests/GRPCHTTP2CoreTests/Internal/TimerTests.swift index 65f9013c5..5f9ad6e82 100644 --- a/Tests/GRPCHTTP2CoreTests/Internal/TimerTests.swift +++ b/Tests/GRPCHTTP2CoreTests/Internal/TimerTests.swift @@ -14,11 +14,10 @@ * limitations under the License. */ -import Atomics import GRPCCore import GRPCHTTP2Core import NIOEmbedded -import Synchronization +private import Synchronization import XCTest @available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *) @@ -62,25 +61,25 @@ internal final class TimerTests: XCTestCase { let loop = EmbeddedEventLoop() defer { try! loop.close() } - let value = Atomic(0) + let counter = AtomicCounter(0) var timer = Timer(delay: .seconds(1), repeat: true) timer.schedule(on: loop) { - value.add(1, ordering: .releasing) + counter.increment() } loop.advanceTime(by: .milliseconds(999)) - XCTAssertEqual(value.load(ordering: .acquiring), 0) + XCTAssertEqual(counter.value, 0) loop.advanceTime(by: .milliseconds(1)) - XCTAssertEqual(value.load(ordering: .acquiring), 1) + XCTAssertEqual(counter.value, 1) loop.advanceTime(by: .seconds(1)) - XCTAssertEqual(value.load(ordering: .acquiring), 2) + XCTAssertEqual(counter.value, 2) loop.advanceTime(by: .seconds(1)) - XCTAssertEqual(value.load(ordering: .acquiring), 3) + XCTAssertEqual(counter.value, 3) timer.cancel() loop.advanceTime(by: .seconds(1)) - XCTAssertEqual(value.load(ordering: .acquiring), 3) + XCTAssertEqual(counter.value, 3) } func testCancelRepeatedTimer() { diff --git a/Tests/GRPCHTTP2CoreTests/Test Utilities/AtomicCounter.swift b/Tests/GRPCHTTP2CoreTests/Test Utilities/AtomicCounter.swift new file mode 100644 index 000000000..7aae0bb5a --- /dev/null +++ b/Tests/GRPCHTTP2CoreTests/Test Utilities/AtomicCounter.swift @@ -0,0 +1,35 @@ +/* + * Copyright 2024, gRPC Authors All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +private import Synchronization + +@available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *) +final class AtomicCounter: Sendable { + private let counter: Atomic + + init(_ initialValue: Int = 0) { + self.counter = Atomic(initialValue) + } + + var value: Int { + self.counter.load(ordering: .sequentiallyConsistent) + } + + @discardableResult + func increment() -> (oldValue: Int, newValue: Int) { + self.counter.add(1, ordering: .sequentiallyConsistent) + } +} From d5a6c1eace7421065a05a2c089b578f77e42d3d6 Mon Sep 17 00:00:00 2001 From: George Barnett Date: Mon, 2 Sep 2024 17:46:41 +0100 Subject: [PATCH 2/6] Add missing imports to v1 --- Package.swift | 4 +++- Package@swift-6.swift | 4 +++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/Package.swift b/Package.swift index 658d97e83..9a9c48637 100644 --- a/Package.swift +++ b/Package.swift @@ -146,6 +146,7 @@ extension Target { .logging, .protobuf, .dequeModule, + .atomics ].appending( .nioSSL, if: includeNIOSSL ), @@ -198,7 +199,8 @@ extension Target { .nioEmbedded, .nioTransportServices, .logging, - .reflectionService + .reflectionService, + .atomics ].appending( .nioSSL, if: includeNIOSSL ), diff --git a/Package@swift-6.swift b/Package@swift-6.swift index 49f3f24a6..f42e48fa5 100644 --- a/Package@swift-6.swift +++ b/Package@swift-6.swift @@ -179,6 +179,7 @@ extension Target { .logging, .protobuf, .dequeModule, + .atomics ].appending( .nioSSL, if: includeNIOSSL ), @@ -381,7 +382,8 @@ extension Target { .nioEmbedded, .nioTransportServices, .logging, - .reflectionService + .reflectionService, + .atomics ].appending( .nioSSL, if: includeNIOSSL ), From 98498cc81cc49760e225fcbebb43c6c620d75d77 Mon Sep 17 00:00:00 2001 From: George Barnett Date: Tue, 3 Sep 2024 10:08:28 +0100 Subject: [PATCH 3/6] add missing target --- Package.swift | 1 + 1 file changed, 1 insertion(+) diff --git a/Package.swift b/Package.swift index 9a9c48637..96d38e4bd 100644 --- a/Package.swift +++ b/Package.swift @@ -123,6 +123,7 @@ extension Target.Dependency { name: "SwiftProtobufPluginLibrary", package: "swift-protobuf" ) + static let atomics: Self = .product(name: "Atomics", package: "swift-atomics") static let dequeModule: Self = .product(name: "DequeModule", package: "swift-collections") } From 35551cc4e028bde56721c463a2411c2417a45c0d Mon Sep 17 00:00:00 2001 From: George Barnett Date: Tue, 3 Sep 2024 10:45:01 +0100 Subject: [PATCH 4/6] add missing package dep --- Package.swift | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/Package.swift b/Package.swift index 96d38e4bd..43d1b987c 100644 --- a/Package.swift +++ b/Package.swift @@ -50,6 +50,10 @@ let packageDependencies: [Package.Dependency] = [ url: "https://github.com/apple/swift-collections.git", from: "1.0.5" ), + .package( + url: "https://github.com/apple/swift-atomics.git", + from: "1.2.0" + ), .package( url: "https://github.com/apple/swift-protobuf.git", from: "1.27.0" From 70368b410c97f4210de40352067d47a55f31da6c Mon Sep 17 00:00:00 2001 From: George Barnett Date: Thu, 5 Sep 2024 13:26:10 +0100 Subject: [PATCH 5/6] Apply suggestions from code review Co-authored-by: Gus Cairo --- Sources/performance-worker/BenchmarkService.swift | 2 +- Tests/GRPCHTTP2CoreTests/Internal/TimerTests.swift | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/Sources/performance-worker/BenchmarkService.swift b/Sources/performance-worker/BenchmarkService.swift index c90eb2fd7..112b1f1f0 100644 --- a/Sources/performance-worker/BenchmarkService.swift +++ b/Sources/performance-worker/BenchmarkService.swift @@ -22,7 +22,7 @@ import struct Foundation.Data @available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *) final class BenchmarkService: Grpc_Testing_BenchmarkService.ServiceProtocol { /// Used to check if the server can be streaming responses. - private let working = Atomic(true) + private let working = Atomic(true) /// One request followed by one response. /// The server returns a client payload with the size requested by the client. diff --git a/Tests/GRPCHTTP2CoreTests/Internal/TimerTests.swift b/Tests/GRPCHTTP2CoreTests/Internal/TimerTests.swift index 5f9ad6e82..4332eac86 100644 --- a/Tests/GRPCHTTP2CoreTests/Internal/TimerTests.swift +++ b/Tests/GRPCHTTP2CoreTests/Internal/TimerTests.swift @@ -61,7 +61,7 @@ internal final class TimerTests: XCTestCase { let loop = EmbeddedEventLoop() defer { try! loop.close() } - let counter = AtomicCounter(0) + let counter = AtomicCounter() var timer = Timer(delay: .seconds(1), repeat: true) timer.schedule(on: loop) { counter.increment() From a022027878957a83c721bffe1e81e7cb6dd328ef Mon Sep 17 00:00:00 2001 From: George Barnett Date: Thu, 5 Sep 2024 13:27:24 +0100 Subject: [PATCH 6/6] remove acl on test imports --- Tests/GRPCCoreTests/Test Utilities/AtomicCounter.swift | 2 +- .../Client/Connection/RequestQueueTests.swift | 2 +- .../Client/Connection/Utilities/TestServer.swift | 2 +- Tests/GRPCHTTP2CoreTests/Internal/TimerTests.swift | 2 +- Tests/GRPCHTTP2CoreTests/Test Utilities/AtomicCounter.swift | 2 +- 5 files changed, 5 insertions(+), 5 deletions(-) diff --git a/Tests/GRPCCoreTests/Test Utilities/AtomicCounter.swift b/Tests/GRPCCoreTests/Test Utilities/AtomicCounter.swift index 7aae0bb5a..b9e9fb5b8 100644 --- a/Tests/GRPCCoreTests/Test Utilities/AtomicCounter.swift +++ b/Tests/GRPCCoreTests/Test Utilities/AtomicCounter.swift @@ -14,7 +14,7 @@ * limitations under the License. */ -private import Synchronization +import Synchronization @available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *) final class AtomicCounter: Sendable { diff --git a/Tests/GRPCHTTP2CoreTests/Client/Connection/RequestQueueTests.swift b/Tests/GRPCHTTP2CoreTests/Client/Connection/RequestQueueTests.swift index 23bec0dab..31b046571 100644 --- a/Tests/GRPCHTTP2CoreTests/Client/Connection/RequestQueueTests.swift +++ b/Tests/GRPCHTTP2CoreTests/Client/Connection/RequestQueueTests.swift @@ -15,7 +15,7 @@ */ import GRPCCore -private import Synchronization +import Synchronization import XCTest @testable import GRPCHTTP2Core diff --git a/Tests/GRPCHTTP2CoreTests/Client/Connection/Utilities/TestServer.swift b/Tests/GRPCHTTP2CoreTests/Client/Connection/Utilities/TestServer.swift index 642a495a1..4063a06e9 100644 --- a/Tests/GRPCHTTP2CoreTests/Client/Connection/Utilities/TestServer.swift +++ b/Tests/GRPCHTTP2CoreTests/Client/Connection/Utilities/TestServer.swift @@ -18,7 +18,7 @@ import GRPCCore import NIOCore import NIOHTTP2 import NIOPosix -private import Synchronization +import Synchronization import XCTest @testable import GRPCHTTP2Core diff --git a/Tests/GRPCHTTP2CoreTests/Internal/TimerTests.swift b/Tests/GRPCHTTP2CoreTests/Internal/TimerTests.swift index 4332eac86..eb920976c 100644 --- a/Tests/GRPCHTTP2CoreTests/Internal/TimerTests.swift +++ b/Tests/GRPCHTTP2CoreTests/Internal/TimerTests.swift @@ -17,7 +17,7 @@ import GRPCCore import GRPCHTTP2Core import NIOEmbedded -private import Synchronization +import Synchronization import XCTest @available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *) diff --git a/Tests/GRPCHTTP2CoreTests/Test Utilities/AtomicCounter.swift b/Tests/GRPCHTTP2CoreTests/Test Utilities/AtomicCounter.swift index 7aae0bb5a..b9e9fb5b8 100644 --- a/Tests/GRPCHTTP2CoreTests/Test Utilities/AtomicCounter.swift +++ b/Tests/GRPCHTTP2CoreTests/Test Utilities/AtomicCounter.swift @@ -14,7 +14,7 @@ * limitations under the License. */ -private import Synchronization +import Synchronization @available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *) final class AtomicCounter: Sendable {