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 apple#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

Tests/NIOCoreTests/AsyncSequences/NIOAsyncWriterTests.swift

@@ -14,6 +14,7 @@
 
 import DequeModule
 import NIOConcurrencyHelpers
+import NIOTestUtils
 import XCTest
 
 @testable import NIOCore
@@ -607,50 +608,54 @@ 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
+            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")
-        }
+            let task1 = Task(executorPreference: taskExecutor) { [writer] in
+                try await writer!.yield("message1")
+            }
+            let task2 = Task(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 task1.value)
+            await XCTAssertNoThrow(try await task2.value)
 
-        self.assert(suspendCallCount: 3, yieldCallCount: 2, terminateCallCount: 1)
+            self.assert(suspendCallCount: 3, yieldCallCount: 2, terminateCallCount: 1)
+        }
+        #endif  // compiler(>=6)
     }
 
     func testWriterFinish_whenFinished() {