Use NIOThreadPoolTaskExecutor for NIOAsyncWriter test that hangs on Android emulator

The testSuspendingBufferedYield_whenWriterFinished test requires at least
two threads in the concurrency thread pool because it blocks one task,
which waits for another task to set a condition. In environments where
the global concurrency thread pool doesn’t have at least two threads
available, the test will fail, as observed on the Android emulator running
with a single virtual core (see discussion in #3044).

Using a custom task executor guarantees that at least two threads are
available for the test, regardless of the width of the global concurrency
thread pool.

Author: orobio

Diff for: Package.swift

@@ -441,6 +441,7 @@ let package = Package(
                 "NIOCore",
                 "NIOEmbedded",
                 "NIOFoundationCompat",
+                "NIOTestUtils",
                 swiftAtomics,
             ],
             swiftSettings: strictConcurrencySettings

Diff for: Tests/NIOCoreTests/AsyncSequences/NIOAsyncWriterTests.swift

@@ -14,6 +14,7 @@
 
 import DequeModule
 import NIOConcurrencyHelpers
+import NIOTestUtils
 import XCTest
 
 @testable import NIOCore
@@ -607,50 +608,56 @@ final class NIOAsyncWriterTests: XCTestCase {
     }
 
     func testSuspendingBufferedYield_whenWriterFinished() async throws {
-        self.sink.setWritability(to: false)
-
-        let bothSuspended = expectation(description: "suspended on both yields")
-        let suspendedAgain = ConditionLock(value: false)
-        self.delegate.didSuspendHandler = {
-            if self.delegate.didSuspendCallCount == 2 {
-                bothSuspended.fulfill()
-            } else if self.delegate.didSuspendCallCount > 2 {
-                suspendedAgain.lock()
-                suspendedAgain.unlock(withValue: true)
-            }
-        }
+        #if compiler(>=6)
+        try await withNIOThreadPoolTaskExecutor(numberOfThreads: 2) { taskExecutor in
+            try await withThrowingTaskGroup(of: Void.self) { group in
+                self.sink.setWritability(to: false)
+
+                let bothSuspended = expectation(description: "suspended on both yields")
+                let suspendedAgain = ConditionLock(value: false)
+                self.delegate.didSuspendHandler = {
+                    if self.delegate.didSuspendCallCount == 2 {
+                        bothSuspended.fulfill()
+                    } else if self.delegate.didSuspendCallCount > 2 {
+                        suspendedAgain.lock()
+                        suspendedAgain.unlock(withValue: true)
+                    }
+                }
 
-        self.delegate.didYieldHandler = { _ in
-            if self.delegate.didYieldCallCount == 1 {
-                // Delay this yield until the other yield is suspended again.
-                if suspendedAgain.lock(whenValue: true, timeoutSeconds: 5) {
-                    suspendedAgain.unlock()
-                } else {
-                    XCTFail("Timeout while waiting for other yield to suspend again.")
+                self.delegate.didYieldHandler = { _ in
+                    if self.delegate.didYieldCallCount == 1 {
+                        // Delay this yield until the other yield is suspended again.
+                        if suspendedAgain.lock(whenValue: true, timeoutSeconds: 5) {
+                            suspendedAgain.unlock()
+                        } else {
+                            XCTFail("Timeout while waiting for other yield to suspend again.")
+                        }
+                    }
                 }
-            }
-        }
 
-        let task1 = Task { [writer] in
-            try await writer!.yield("message1")
-        }
-        let task2 = Task { [writer] in
-            try await writer!.yield("message2")
-        }
+                group.addTask(executorPreference: taskExecutor) { [writer] in
+                    try await writer!.yield("message1")
+                }
+                group.addTask(executorPreference: taskExecutor) { [writer] in
+                    try await writer!.yield("message2")
+                }
 
-        await fulfillment(of: [bothSuspended], timeout: 5)
-        self.writer.finish()
+                await fulfillment(of: [bothSuspended], timeout: 5)
+                self.writer.finish()
 
-        self.assert(suspendCallCount: 2, yieldCallCount: 0, terminateCallCount: 0)
+                self.assert(suspendCallCount: 2, yieldCallCount: 0, terminateCallCount: 0)
 
-        // We have to become writable again to unbuffer the yields
-        // The first call to didYield will pause, so that the other yield will be suspended again.
-        self.sink.setWritability(to: true)
+                // We have to become writable again to unbuffer the yields
+                // The first call to didYield will pause, so that the other yield will be suspended again.
+                self.sink.setWritability(to: true)
 
-        await XCTAssertNoThrow(try await task1.value)
-        await XCTAssertNoThrow(try await task2.value)
+                await XCTAssertNoThrow(try await group.next())
+                await XCTAssertNoThrow(try await group.next())
 
-        self.assert(suspendCallCount: 3, yieldCallCount: 2, terminateCallCount: 1)
+                self.assert(suspendCallCount: 3, yieldCallCount: 2, terminateCallCount: 1)
+            }
+        }
+        #endif  // compiler(>=6)
     }
 
     func testWriterFinish_whenFinished() {