Strict concurrency for NIOPerformanceTester and NIOCrashTester (#3167)

Author: glbrntt

Package.swift

@@ -424,7 +424,8 @@ let package = Package(
                 "NIOHTTP1",
                 "NIOFoundationCompat",
                 "NIOWebSocket",
-            ]
+            ],
+            swiftSettings: strictConcurrencySettings
         ),
         .executableTarget(
             name: "NIOCrashTester",
@@ -435,7 +436,8 @@ let package = Package(
                 "NIOHTTP1",
                 "NIOWebSocket",
                 "NIOFoundationCompat",
-            ]
+            ],
+            swiftSettings: strictConcurrencySettings
         ),
         .testTarget(
             name: "NIOCoreTests",

Sources/NIOCrashTester/CrashTestSuites.swift

@@ -13,12 +13,14 @@
 //===----------------------------------------------------------------------===//
 
 #if !canImport(Darwin) || os(macOS)
-let crashTestSuites: [String: Any] = [
-    "EventLoopCrashTests": EventLoopCrashTests(),
-    "ByteBufferCrashTests": ByteBufferCrashTests(),
-    "SystemCrashTests": SystemCrashTests(),
-    "HTTPCrashTests": HTTPCrashTests(),
-    "StrictCrashTests": StrictCrashTests(),
-    "LoopBoundTests": LoopBoundTests(),
-]
+func makeCrashTestSuites() -> [String: Any] {
+    [
+        "EventLoopCrashTests": EventLoopCrashTests(),
+        "ByteBufferCrashTests": ByteBufferCrashTests(),
+        "SystemCrashTests": SystemCrashTests(),
+        "HTTPCrashTests": HTTPCrashTests(),
+        "StrictCrashTests": StrictCrashTests(),
+        "LoopBoundTests": LoopBoundTests(),
+    ]
+}
 #endif

Sources/NIOCrashTester/CrashTests+EventLoop.swift

@@ -82,9 +82,9 @@ struct EventLoopCrashTests {
                 exit(2)
             }
             func f() {
-                el.scheduleTask(in: .nanoseconds(0)) { [f] in
+                el.assumeIsolated().scheduleTask(in: .nanoseconds(0)) { [f] in
                     f()
-                }.futureResult.whenFailure { [f] error in
+                }.futureResult.assumeIsolated().whenFailure { [f] error in
                     guard case .some(.shutdown) = error as? EventLoopError else {
                         exit(3)
                     }

Sources/NIOCrashTester/main.swift

@@ -76,7 +76,8 @@ func main() throws {
     }
 
     func allTestsForSuite(_ testSuite: String) -> [(String, CrashTest)] {
-        crashTestSuites[testSuite].map { testSuiteObject in
+        let crashTestSuites = makeCrashTestSuites()
+        return crashTestSuites[testSuite].map { testSuiteObject in
             Mirror(reflecting: testSuiteObject)
                 .children
                 .filter { $0.label?.starts(with: "test") ?? false }
@@ -199,6 +200,7 @@ func main() throws {
 
     switch CommandLine.arguments.dropFirst().first {
     case .some("run-all"):
+        let crashTestSuites = makeCrashTestSuites()
         for testSuite in crashTestSuites {
             for test in allTestsForSuite(testSuite.key) {
                 try runAndEval(test.0, suite: testSuite.key)

Sources/NIOPerformanceTester/ChannelPipelineBenchmark.swift

@@ -15,10 +15,10 @@ import NIOCore
 import NIOEmbedded
 
 final class ChannelPipelineBenchmark: Benchmark {
-    private final class NoOpHandler: ChannelInboundHandler, RemovableChannelHandler {
+    private final class NoOpHandler: ChannelInboundHandler, RemovableChannelHandler, Sendable {
         typealias InboundIn = Any
     }
-    private final class ConsumingHandler: ChannelInboundHandler, RemovableChannelHandler {
+    private final class ConsumingHandler: ChannelInboundHandler, RemovableChannelHandler, Sendable {
         typealias InboundIn = Any
 
         func channelReadComplete(context: ChannelHandlerContext) {
@@ -28,7 +28,7 @@ final class ChannelPipelineBenchmark: Benchmark {
     private let channel: EmbeddedChannel
     private let runCount: Int
     private let extraHandlers = 4
-    private var handlers: [RemovableChannelHandler] = []
+    private var handlers: [RemovableChannelHandler & Sendable] = []
 
     init(runCount: Int) {
         self.channel = EmbeddedChannel()

Sources/NIOPerformanceTester/ExecuteBenchmark.swift

@@ -20,7 +20,6 @@ final class ExecuteBenchmark: Benchmark {
     private var group: MultiThreadedEventLoopGroup!
     private var loop: EventLoop!
     private var dg: DispatchGroup!
-    private var counter = 0
     private let numTasks: Int
 
     init(numTasks: Int) {
@@ -34,10 +33,10 @@ final class ExecuteBenchmark: Benchmark {
 
         // We are preheating the EL to avoid growing the `ScheduledTask` `PriorityQueue`
         // during the actual test
-        try! self.loop.submit {
+        try! self.loop.submit { [loop, numTasks] in
             var counter: Int = 0
-            for _ in 0..<self.numTasks {
-                self.loop.scheduleTask(in: .nanoseconds(0)) {
+            for _ in 0..<numTasks {
+                loop!.assumeIsolatedUnsafeUnchecked().scheduleTask(in: .nanoseconds(0)) {
                     counter &+= 1
                 }
             }
@@ -47,19 +46,21 @@ final class ExecuteBenchmark: Benchmark {
     func tearDown() {}
 
     func run() -> Int {
-        try! self.loop.submit {
-            for _ in 0..<self.numTasks {
-                self.dg.enter()
+        let counter = try! self.loop.submit { [dg, loop, numTasks] in
+            var counter = 0
+            for _ in 0..<numTasks {
+                dg!.enter()
 
-                self.loop.execute {
-                    self.counter &+= 1
-                    self.dg.leave()
+                loop!.assumeIsolatedUnsafeUnchecked().execute {
+                    counter &+= 1
+                    dg!.leave()
                 }
             }
+            return counter
         }.wait()
         self.dg.wait()
 
-        return self.counter
+        return counter
     }
 
 }

Sources/NIOPerformanceTester/LockBenchmark.swift

@@ -17,7 +17,9 @@ import NIOConcurrencyHelpers
 import NIOCore
 import NIOPosix
 
-final class NIOLockBenchmark: Benchmark {
+final class NIOLockBenchmark: Benchmark, @unchecked Sendable {
+    // mutable state is protected by the lock
+
     private let numberOfThreads: Int
     private let lockOperationsPerThread: Int
     private let threadPool: NIOThreadPool

Sources/NIOPerformanceTester/SchedulingAndRunningBenchmark.swift

@@ -20,7 +20,6 @@ final class SchedulingAndRunningBenchmark: Benchmark {
     private var group: MultiThreadedEventLoopGroup!
     private var loop: EventLoop!
     private var dg: DispatchGroup!
-    private var counter = 0
     private let numTasks: Int
 
     init(numTasks: Int) {
@@ -34,10 +33,10 @@ final class SchedulingAndRunningBenchmark: Benchmark {
 
         // We are preheating the EL to avoid growing the `ScheduledTask` `PriorityQueue`
         // during the actual test
-        try! self.loop.submit {
+        try! self.loop.submit { [loop, numTasks] in
             var counter: Int = 0
-            for _ in 0..<self.numTasks {
-                self.loop.scheduleTask(in: .nanoseconds(0)) {
+            for _ in 0..<numTasks {
+                loop!.assumeIsolatedUnsafeUnchecked().scheduleTask(in: .nanoseconds(0)) {
                     counter &+= 1
                 }
             }
@@ -47,15 +46,17 @@ final class SchedulingAndRunningBenchmark: Benchmark {
     func tearDown() {}
 
     func run() -> Int {
-        try! self.loop.submit {
-            for _ in 0..<self.numTasks {
-                self.dg.enter()
+        let counter = try! self.loop.submit { [dg, loop, numTasks] in
+            var counter: Int = 0
+            for _ in 0..<numTasks {
+                dg!.enter()
 
-                self.loop.scheduleTask(in: .nanoseconds(0)) {
-                    self.counter &+= 1
-                    self.dg.leave()
+                loop!.assumeIsolatedUnsafeUnchecked().scheduleTask(in: .nanoseconds(0)) {
+                    counter &+= 1
+                    dg!.leave()
                 }
             }
+            return counter
         }.wait()
         self.dg.wait()
 

Sources/NIOPerformanceTester/TCPThroughputBenchmark.swift

@@ -28,7 +28,7 @@ final class TCPThroughputBenchmark: Benchmark {
 
     private var group: EventLoopGroup!
     private var serverChannel: Channel!
-    private var serverHandler: ServerHandler!
+    private var serverHandler: NIOLoopBound<ServerHandler>!
     private var clientChannel: Channel!
 
     private var message: ByteBuffer!
@@ -39,10 +39,12 @@ final class TCPThroughputBenchmark: Benchmark {
         public typealias OutboundOut = ByteBuffer
 
         private let connectionEstablishedPromise: EventLoopPromise<EventLoop>
+        private let eventLoop: EventLoop
         private var context: ChannelHandlerContext!
 
-        init(_ connectionEstablishedPromise: EventLoopPromise<EventLoop>) {
+        init(_ connectionEstablishedPromise: EventLoopPromise<EventLoop>, eventLoop: EventLoop) {
             self.connectionEstablishedPromise = connectionEstablishedPromise
+            self.eventLoop = eventLoop
         }
 
         public func channelActive(context: ChannelHandlerContext) {
@@ -114,14 +116,20 @@ final class TCPThroughputBenchmark: Benchmark {
 
         let connectionEstablishedPromise: EventLoopPromise<EventLoop> = self.group.next().makePromise()
 
+        let promise = self.group.next().makePromise(of: NIOLoopBound<ServerHandler>.self)
         self.serverChannel = try ServerBootstrap(group: self.group)
             .childChannelInitializer { channel in
-                self.serverHandler = ServerHandler(connectionEstablishedPromise)
-                return channel.pipeline.addHandler(self.serverHandler)
+                channel.eventLoop.makeCompletedFuture {
+                    let serverHandler = ServerHandler(connectionEstablishedPromise, eventLoop: channel.eventLoop)
+                    promise.succeed(NIOLoopBound(serverHandler, eventLoop: channel.eventLoop))
+                    try channel.pipeline.syncOperations.addHandler(serverHandler)
+                }
             }
             .bind(host: "127.0.0.1", port: 0)
             .wait()
 
+        self.serverHandler = try promise.futureResult.wait()
+
         self.clientChannel = try ClientBootstrap(group: group)
             .channelInitializer { channel in
                 channel.eventLoop.makeCompletedFuture {
@@ -165,7 +173,7 @@ final class TCPThroughputBenchmark: Benchmark {
         let messages = self.messages
 
         self.serverEventLoop.execute {
-            serverHandler.send(message, times: messages)
+            serverHandler.value.send(message, times: messages)
         }
         try isDonePromise.futureResult.wait()
         return 0

Sources/NIOPerformanceTester/UDPBenchmark.swift

@@ -28,7 +28,7 @@ final class UDPBenchmark {
     private var group: EventLoopGroup!
     private var server: Channel!
     private var client: Channel!
-    private var clientHandler: EchoHandlerClient!
+    private var clientHandler: EchoHandlerClient.SendableView!
 
     init(data: ByteBuffer, numberOfRequests: Int, vectorReads: Int, vectorWrites: Int) {
         self.data = data
@@ -57,22 +57,25 @@ extension UDPBenchmark: Benchmark {
         self.client = try DatagramBootstrap(group: group)
             // zero is the same as not applying the option.
             .channelOption(.datagramVectorReadMessageCount, value: self.vectorReads)
-            .channelInitializer { channel in
-                let handler = EchoHandlerClient(
-                    eventLoop: channel.eventLoop,
-                    config: .init(
-                        remoteAddress: remoteAddress,
-                        request: self.data,
-                        requests: self.numberOfRequests,
-                        writesPerFlush: self.vectorWrites
+            .channelInitializer { [data, numberOfRequests, vectorWrites] channel in
+                channel.eventLoop.makeCompletedFuture {
+                    let handler = EchoHandlerClient(
+                        eventLoop: channel.eventLoop,
+                        config: .init(
+                            remoteAddress: remoteAddress,
+                            request: data,
+                            requests: numberOfRequests,
+                            writesPerFlush: vectorWrites
+                        )
                     )
-                )
-                return channel.pipeline.addHandler(handler)
+                    try channel.pipeline.syncOperations.addHandler(handler)
+                }
             }
             .bind(to: address)
             .wait()
 
-        self.clientHandler = try self.client.pipeline.handler(type: EchoHandlerClient.self).wait()
+        self.clientHandler = try self.client.pipeline.handler(type: EchoHandlerClient.self).map { $0.sendableView }
+            .wait()
     }
 
     func tearDown() {
@@ -87,7 +90,7 @@ extension UDPBenchmark: Benchmark {
 }
 
 extension UDPBenchmark {
-    final class EchoHandler: ChannelInboundHandler {
+    final class EchoHandler: ChannelInboundHandler, Sendable {
         typealias InboundIn = AddressedEnvelope<ByteBuffer>
         typealias OutboundOut = AddressedEnvelope<ByteBuffer>
 
@@ -232,12 +235,26 @@ extension UDPBenchmark {
             self.context = nil
         }
 
-        func run() -> EventLoopFuture<Void> {
-            let p = self.eventLoop.makePromise(of: Void.self)
-            self.eventLoop.execute {
-                self._run(promise: p)
+        var sendableView: SendableView {
+            SendableView(handler: self, eventLoop: self.eventLoop)
+        }
+
+        struct SendableView: Sendable {
+            private let handler: NIOLoopBound<EchoHandlerClient>
+            private let eventLoop: EventLoop
+
+            init(handler: EchoHandlerClient, eventLoop: EventLoop) {
+                self.handler = NIOLoopBound(handler, eventLoop: eventLoop)
+                self.eventLoop = eventLoop
+            }
+
+            func run() -> EventLoopFuture<Void> {
+                let p = self.eventLoop.makePromise(of: Void.self)
+                self.eventLoop.execute {
+                    self.handler.value._run(promise: p)
+                }
+                return p.futureResult
             }
-            return p.futureResult
         }
 
         private func _run(promise: EventLoopPromise<Void>) {

Sources/NIOPerformanceTester/WebSocketFrameEncoderBenchmark.swift

@@ -67,9 +67,9 @@ extension WebSocketFrameEncoderBenchmark {
 extension WebSocketFrameEncoderBenchmark: Benchmark {
     func setUp() throws {
         // We want the pipeline walk to have some cost.
-        try! self.channel.pipeline.addHandler(WriteConsumingHandler()).wait()
+        try! self.channel.pipeline.syncOperations.addHandler(WriteConsumingHandler())
         for _ in 0..<3 {
-            try! self.channel.pipeline.addHandler(NoOpOutboundHandler()).wait()
+            try! self.channel.pipeline.syncOperations.addHandler(NoOpOutboundHandler())
         }
         try! self.channel.pipeline.syncOperations.addHandler(WebSocketFrameEncoder())
         self.frame = WebSocketFrame(opcode: .binary, maskKey: self.maskingKey, data: self.data, extensionData: nil)