Call NIOAsyncWriterSinkDelegate outside of the lock

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

@@ -1,3 +1,3 @@
 {
-  "mallocCountTotal" : 1317015
-}
+  "mallocCountTotal" : 164419
+}

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

@@ -1,3 +1,3 @@
 {
-  "mallocCountTotal" : 1317022
+  "mallocCountTotal" : 164426
 }

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

@@ -1,3 +1,3 @@
 {
-  "mallocCountTotal" : 1317015
+  "mallocCountTotal" : 164419
 }

Sources/NIOCore/AsyncChannel/AsyncChannelOutboundWriterHandler.swift

@@ -170,25 +170,34 @@ extension NIOAsyncChannelOutboundWriterHandler {
 
         @inlinable
         func didYield(contentsOf sequence: Deque<OutboundOut>) {
-            // This always called from an async context, so we must loop-hop.
-            self.eventLoop.execute {
+            if self.eventLoop.inEventLoop {
                 self.handler._didYield(sequence: sequence)
+            } else {
+                self.eventLoop.execute {
+                    self.handler._didYield(sequence: sequence)
+                }
             }
         }
 
         @inlinable
         func didYield(_ element: OutboundOut) {
-            // This always called from an async context, so we must loop-hop.
-            self.eventLoop.execute {
+            if self.eventLoop.inEventLoop {
                 self.handler._didYield(element: element)
+            } else {
+                self.eventLoop.execute {
+                    self.handler._didYield(element: element)
+                }
             }
         }
 
         @inlinable
         func didTerminate(error: Error?) {
-            // This always called from an async context, so we must loop-hop.
-            self.eventLoop.execute {
+            if self.eventLoop.inEventLoop {
                 self.handler._didTerminate(error: error)
+            } else {
+                self.eventLoop.execute {
+                    self.handler._didTerminate(error: error)
+                }
             }
         }
     }

Tests/NIOCoreTests/AsyncChannel/AsyncChannelTests.swift

@@ -85,7 +85,7 @@ final class AsyncChannelTests: XCTestCase {
             inboundReader = wrapped.inboundStream
 
             try await channel.testingEventLoop.executeInContext {
-                XCTAssertEqual(0, closeRecorder.outboundCloses)
+                XCTAssertEqual(1, closeRecorder.outboundCloses)
             }
         }
 
@@ -159,7 +159,7 @@ final class AsyncChannelTests: XCTestCase {
                 inboundReader = wrapped.inboundStream
 
                 try await channel.testingEventLoop.executeInContext {
-                    XCTAssertEqual(0, closeRecorder.allCloses)
+                    XCTAssertEqual(1, closeRecorder.allCloses)
                 }
             }
 

Tests/NIOCoreTests/AsyncSequences/NIOAsyncWriterTests.swift

@@ -15,25 +15,32 @@
 import DequeModule
 import NIOCore
 import XCTest
+import NIOConcurrencyHelpers
 
 private struct SomeError: Error, Hashable {}
 
 private final class MockAsyncWriterDelegate: NIOAsyncWriterSinkDelegate, @unchecked Sendable {
     typealias Element = String
 
-    var didYieldCallCount = 0
+    var _didYieldCallCount = NIOLockedValueBox(0)
+    var didYieldCallCount: Int {
+        self._didYieldCallCount.withLockedValue { $0 }
+    }
     var didYieldHandler: ((Deque<String>) -> Void)?
     func didYield(contentsOf sequence: Deque<String>) {
-        self.didYieldCallCount += 1
+        self._didYieldCallCount.withLockedValue { $0 += 1 }
         if let didYieldHandler = self.didYieldHandler {
             didYieldHandler(sequence)
         }
     }
 
-    var didTerminateCallCount = 0
+    var _didTerminateCallCount = NIOLockedValueBox(0)
+    var didTerminateCallCount: Int {
+        self._didTerminateCallCount.withLockedValue { $0 }
+    }
     var didTerminateHandler: ((Error?) -> Void)?
     func didTerminate(error: Error?) {
-        self.didTerminateCallCount += 1
+        self._didTerminateCallCount.withLockedValue { $0 += 1 }
         if let didTerminateHandler = self.didTerminateHandler {
             didTerminateHandler(error)
         }
@@ -68,6 +75,8 @@ final class NIOAsyncWriterTests: XCTestCase {
     }
 
     func testMultipleConcurrentWrites() async throws {
+        var elements = 0
+        self.delegate.didYieldHandler = { elements += $0.count }
         let task1 = Task { [writer] in
             for i in 0...9 {
                 try await writer!.yield("message\(i)")
@@ -88,7 +97,33 @@ final class NIOAsyncWriterTests: XCTestCase {
         try await task2.value
         try await task3.value
 
-        XCTAssertEqual(self.delegate.didYieldCallCount, 30)
+        XCTAssertEqual(elements, 30)
+    }
+
+    func testMultipleConcurrentBatchWrites() async throws {
+        var elements = 0
+        self.delegate.didYieldHandler = { elements += $0.count }
+        let task1 = Task { [writer] in
+            for i in 0...9 {
+                try await writer!.yield(contentsOf: ["message\(i).1", "message\(i).2"])
+            }
+        }
+        let task2 = Task { [writer] in
+            for i in 10...19 {
+                try await writer!.yield(contentsOf: ["message\(i).1", "message\(i).2"])
+            }
+        }
+        let task3 = Task { [writer] in
+            for i in 20...29 {
+                try await writer!.yield(contentsOf: ["message\(i).1", "message\(i).2"])
+            }
+        }
+
+        try await task1.value
+        try await task2.value
+        try await task3.value
+
+        XCTAssertEqual(elements, 60)
     }
 
     func testWriterCoalescesWrites() async throws {