Add NIOAsyncChannel benchmark

Benchmarks/Benchmarks/NIOPosixBenchmarks/Benchmarks.swift

@@ -13,6 +13,9 @@
 //===----------------------------------------------------------------------===//
 
 import Benchmark
+import NIOPosix
+
+private let eventLoop = MultiThreadedEventLoopGroup(numberOfThreads: 1).next()
 
 let benchmarks = {
     let defaultMetrics: [BenchmarkMetric] = [
@@ -27,6 +30,35 @@ let benchmarks = {
             scalingFactor: .mega
         )
     ) { benchmark in
-        try runTCPEcho(numberOfWrites: benchmark.scaledIterations.upperBound)
+        try runTCPEcho(
+            numberOfWrites: benchmark.scaledIterations.upperBound,
+            eventLoop: eventLoop
+        )
+    }
+
+    // This benchmark is only available above 5.9 since our EL conformance
+    // to serial executor is also gated behind 5.9.
+    #if compiler(>=5.9)
+    Benchmark(
+        "TCPEchoAsyncChannel",
+        configuration: .init(
+            metrics: defaultMetrics,
+            timeUnits: .milliseconds,
+            scalingFactor: .mega,
+            setup: {
+                swiftTaskEnqueueGlobalHook = { job, _ in
+                    eventLoop.executor.enqueue(job)
+                }
+            },
+            teardown: {
+                swiftTaskEnqueueGlobalHook = nil
+            }
+        )
+    ) { benchmark in
+        try await runTCPEchoAsyncChannel(
+            numberOfWrites: benchmark.scaledIterations.upperBound,
+            eventLoop: eventLoop
+        )
     }
+    #endif
 }

Benchmarks/Benchmarks/NIOPosixBenchmarks/TCPEcho.swift

@@ -55,8 +55,7 @@ private final class EchoRequestChannelHandler: ChannelInboundHandler {
     }
 }
 
-func runTCPEcho(numberOfWrites: Int) throws {
-    let eventLoop = MultiThreadedEventLoopGroup.singleton.next()
+func runTCPEcho(numberOfWrites: Int, eventLoop: any EventLoop) throws {
     let serverChannel = try ServerBootstrap(group: eventLoop)
         .childChannelInitializer { channel in
             channel.eventLoop.makeCompletedFuture {

Benchmarks/Benchmarks/NIOPosixBenchmarks/TCPEchoAsyncChannel.swift

@@ -0,0 +1,89 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the SwiftNIO open source project
+//
+// Copyright (c) 2023 Apple Inc. and the SwiftNIO project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of SwiftNIO project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+@_spi(AsyncChannel) import NIOCore
+@_spi(AsyncChannel) import NIOPosix
+
+func runTCPEchoAsyncChannel(numberOfWrites: Int, eventLoop: EventLoop) async throws {
+    let serverChannel = try await ServerBootstrap(group: eventLoop)
+        .bind(
+            host: "127.0.0.1",
+            port: 0
+        ) { channel in
+            channel.eventLoop.makeCompletedFuture {
+                return try NIOAsyncChannel(
+                    synchronouslyWrapping: channel,
+                    configuration: .init(
+                        inboundType: ByteBuffer.self,
+                        outboundType: ByteBuffer.self
+                    )
+                )
+            }
+        }
+
+    let clientChannel = try await ClientBootstrap(group: eventLoop)
+        .connect(
+            host: "127.0.0.1",
+            port: serverChannel.channel.localAddress!.port!
+        ) { channel in
+            channel.eventLoop.makeCompletedFuture {
+                return try NIOAsyncChannel(
+                    synchronouslyWrapping: channel,
+                    configuration: .init(
+                        inboundType: ByteBuffer.self,
+                        outboundType: ByteBuffer.self
+                    )
+                )
+            }
+        }
+
+    let bufferSize = 10000
+
+    try await withThrowingTaskGroup(of: Void.self) { group in
+        // This child task is echoing back the data on the server.
+        group.addTask {
+            for try await connectionChannel in serverChannel.inboundStream {
+                for try await inboundData in connectionChannel.inboundStream {
+                    try await connectionChannel.outboundWriter.write(inboundData)
+                }
+            }
+        }
+
+        // This child task is collecting the echoed back responses.
+        group.addTask {
+            var receivedData = 0
+            for try await inboundData in clientChannel.inboundStream  {
+                receivedData += inboundData.readableBytes
+
+                if receivedData == numberOfWrites * bufferSize {
+                    return
+                }
+            }
+        }
+
+        // Let's start sending data.
+        let data = ByteBuffer(repeating: 0, count: bufferSize)
+        for _ in 0..<numberOfWrites {
+            try await clientChannel.outboundWriter.write(data)
+        }
+
+        // Waiting for the child task that collects the responses to finish.
+        try await group.next()
+
+        // Cancelling the server child task.
+        group.cancelAll()
+        try await serverChannel.channel.closeFuture.get()
+        try await clientChannel.channel.closeFuture.get()
+    }
+}

Benchmarks/Benchmarks/NIOPosixBenchmarks/Util/GlobalExecutor.swift

@@ -0,0 +1,33 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the SwiftNIO open source project
+//
+// Copyright (c) 2023 Apple Inc. and the SwiftNIO project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of SwiftNIO project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+#if canImport(Darwin)
+import Darwin.C
+#elseif canImport(Glibc)
+import Glibc
+#else
+#error("Unsupported platform.")
+#endif
+
+// This file allows us to hook the global executor which
+// we can use to mimic task executors for now.
+typealias EnqueueGlobalHook = @convention(thin) (UnownedJob, @convention(thin) (UnownedJob) -> Void) -> Void
+
+var swiftTaskEnqueueGlobalHook: EnqueueGlobalHook? {
+  get { _swiftTaskEnqueueGlobalHook.pointee }
+  set { _swiftTaskEnqueueGlobalHook.pointee = newValue }
+}
+
+private let _swiftTaskEnqueueGlobalHook: UnsafeMutablePointer<EnqueueGlobalHook?> =
+  dlsym(dlopen(nil, RTLD_LAZY), "swift_task_enqueueGlobal_hook").assumingMemoryBound(to: EnqueueGlobalHook?.self)

Benchmarks/Package.swift

@@ -18,7 +18,7 @@ import PackageDescription
 let package = Package(
     name: "benchmarks",
     platforms: [
-        .macOS(.v13),
+        .macOS("14"),
     ],
     dependencies: [
         .package(path: "../"),

Benchmarks/Thresholds/5.10/NIOPosixBenchmarks.TCPEcho.p90.json

@@ -1,3 +1,3 @@
 {
-  "mallocCountTotal" : 93
+  "mallocCountTotal" : 90
 }

Benchmarks/Thresholds/5.10/NIOPosixBenchmarks.TCPEchoAsyncChannel.p90.json

@@ -0,0 +1,3 @@
+{
+  "mallocCountTotal" : 5554895
+}

Benchmarks/Thresholds/5.7/NIOPosixBenchmarks.TCPEcho.p90.json

@@ -1,3 +1,3 @@
 {
-  "mallocCountTotal" : 95
+  "mallocCountTotal" : 92
 }

Benchmarks/Thresholds/5.8/NIOPosixBenchmarks.TCPEcho.p90.json

@@ -1,3 +1,3 @@
 {
-  "mallocCountTotal" : 95
+  "mallocCountTotal" : 92
 }

Benchmarks/Thresholds/5.9/NIOPosixBenchmarks.TCPEcho.p90.json

@@ -1,3 +1,3 @@
 {
-  "mallocCountTotal" : 95
+  "mallocCountTotal" : 92
 }

Benchmarks/Thresholds/5.9/NIOPosixBenchmarks.TCPEchoAsyncChannel.p90.json

@@ -0,0 +1,3 @@
+{
+  "mallocCountTotal" : 5636901
+}

Benchmarks/Thresholds/main/NIOPosixBenchmarks.TCPEcho.p90.json

@@ -1,3 +1,3 @@
 {
-  "mallocCountTotal" : 93
+  "mallocCountTotal" : 90
 }

Benchmarks/Thresholds/main/NIOPosixBenchmarks.TCPEchoAsyncChannel.p90.json

@@ -0,0 +1,3 @@
+{
+  "mallocCountTotal" : 5554895
+}