使用结构化并发测试并行执行

Question

I\xe2\x80\x99m 测试使用 actor 的代码，并且 I\xe2\x80\x99d 喜欢测试 I\xe2\x80\x99m 是否正确处理并发访问和重入。我常用的方法之一是用来DispatchQueue.concurrentPerform触发来自不同线程的一堆请求，并确保我的值按预期解析。但是，由于参与者使用结构化并发，因此 I\xe2\x80\x99m 不确定如何实际等待任务完成。

\n

我\xe2\x80\x99d 喜欢做的是这样的：

\n

let iterationCount = 100\nlet allTasksComplete = expectation(description: "allTasksComplete")\nallTasksComplete.expectedFulfillmentCount = iterationCount\nDispatchQueue.concurrentPerform(iterations: iterationCount) { _ in\n    Task {\n        // Do some async work here, and assert\n        allTasksComplete.fulfill()\n    }\n}\nwait(for: [allTasksComplete], timeout: 1.0)\n

\n

然而，期望的超时allTasksComplete每次都会到期，无论迭代计数是 1 还是 100，也无论超时的长度如何。I\xe2\x80\x99m 假设这与混合结构化并发和 DispatchQueue 式并发是禁忌有关？

\n

如何正确测试并发访问 \xe2\x80\x94 特别是如何保证从不同线程访问参与者，并等待测试完成直到满足所有期望？

\n

Answer 1

一些观察：

\n

\n

1. 在测试 Swift 并发时，我们不再需要依赖期望。我们可以将我们的测试标记为async方法。请参阅异步测试和期望。这是根据该示例改编的异步测试：

   \n

   func testDownloadWebDataWithConcurrency() async throws {\n    let url = try XCTUnwrap(URL(string: "https://apple.com"), "Expected valid URL.")\n\n    let (_, response) = try await URLSession.shared.data(from: url)\n\n    let httpResponse = try XCTUnwrap(response as? HTTPURLResponse, "Expected an HTTPURLResponse.")\n    XCTAssertEqual(httpResponse.statusCode, 200, "Expected a 200 OK response.")\n}\n

   \n
\n

2. FWIW，虽然我们现在可以async在测试 Swift 并发时使用测试，但我们仍然可以使用期望：

   \n

   func testWithExpectation() {\n    let iterations = 100\n    let experiment = ExperimentActor()\n\n    let e = self.expectation(description: #function)\n    e.expectedFulfillmentCount = iterations\n\n    for i in 0 ..< iterations {\n        Task.detached {\n            let result = await experiment.reentrantCalculation(i)\n            let success = await experiment.isAcceptable(result)\n            XCTAssert(success, "Incorrect value")\n            e.fulfill()\n        }\n    }\n\n    wait(for: [e], timeout: 10)\n}\n

   \n
\n

3. 你说：

   \n

   \n

   然而，期望的超时allTasksComplete每次都会到期，无论迭代计数是 1 还是 100，也无论超时的长度如何。

   \n

   \n

   如果没有看到用注释 \xe2\x80\x9c 替换的代码的可重现示例，我们就无法发表评论。在这里做一些异步工作，并断言 \xe2\x80\x9d。我们不需要看到您的实际实现，而是构建最简单的示例来体现您所描述的行为。请参阅如何创建最小的、可重现的示例。

   \n

   我个人怀疑您还有其他一些不相关的僵局。例如，假设concurrentPerform阻塞了您调用它的线程，也许您正在做一些需要被阻塞线程的事情。另外，请小心Task { ... }，它在当前参与者上运行任务，因此如果您在其中执行缓慢且同步的操作，则可能会导致问题。相反，我们可以使用分离任务。

   \n

   简而言之，如果没有最小的、可重现的示例，我们就无法诊断问题。

   \n
\n

4. 作为一种更普遍的观察，人们应该警惕将 GCD（或信号量或长期锁或其他什么）与 Swift 并发混合，因为后者使用协作线程池，它依赖于其线程能够取得进展的假设。但如果您有 GCD API 阻塞线程，这些假设可能不再有效。它可能不是问题的根源，但我提到它作为警告。

   \n
\n

5. 顺便说一句，concurrentPerform（限制并行度）只有在正在执行的工作同步运行时才有意义。使用concurrentPerform来启动一系列异步任务根本不会限制并发性。（协作线程池可能会，但concurrentPerform不会。）

   \n

   因此，例如，如果我们想并行测试一堆计算，而不是concurrentPerform，我们可以使用TaskGroup：

   \n

   func testWithStructuredConcurrency() async {\n    let iterations = 100\n    let experiment = ExperimentActor()\n\n    await withTaskGroup(of: Void.self) { group in\n        for i in 0 ..< iterations {\n            group.addTask {\n                let result = await experiment.reentrantCalculation(i)\n                let success = await experiment.isAcceptable(result)\n                XCTAssert(success, "Incorrect value")\n            }\n        }\n    }\n\n    let count = await experiment.count\n    XCTAssertEqual(count, iterations)\n}\n

   \n
\n

6. 现在，如果您想验证应用程序内的并发执行，通常我只需使用 Instruments 分析该应用程序（而不是单元测试），然后在 \xe2\x80\x9cPoints of Interest\xe2\x80\x9d 工具中观察间隔，或者查看在 WWDC 2022 中描述的新 \xe2\x80\x9cSwift Tasks\xe2\x80\x9d 工具\xe2\x80\x99s可视化和优化 Swift 并发视频。例如，在这里我启动了四十个任务，我可以看到我的设备一次运行六个任务：

   \n

   

   \n

   请参阅DTSendSignalFlag 的替代方案来识别仪器中的关键事件？有关 \xe2\x80\x9cPoints of Interest\xe2\x80\x9d 工具的参考。

   \n
\n

7. 如果您确实想编写单元测试来确认并发性，理论上您可以跟踪自己的计数器，例如，

   \n

   final class MyAppTests: XCTestCase {\n    func testWithStructuredConcurrency() async {\n        let iterations = 100\n        let experiment = ExperimentActor()\n\n        await withTaskGroup(of: Void.self) { group in\n            for i in 0 ..< iterations {\n                group.addTask {\n                    let result = await experiment.reentrantCalculation(i)\n                    let success = await experiment.isAcceptable(result)\n                    XCTAssert(success, "Incorrect value")\n                }\n            }\n        }\n\n        let count = await experiment.count\n        XCTAssertEqual(count, iterations, "Correct count")\n\n        let degreeOfConcurrency = await experiment.maxDegreeOfConcurrency\n        XCTAssertGreaterThan(degreeOfConcurrency, 1, "No concurrency")\n    }\n}\n

   \n

   在哪里：

   \n

   actor ExperimentActor {\n    var degreeOfConcurrency = 0\n    var maxDegreeOfConcurrency = 0\n    var count = 0\n\n    /// Calculate pi with Leibniz series\n    ///\n    /// Note: I am awaiting a detached task so that I can manifest actor reentrancy.\n\n    func reentrantCalculation(_ index: Int, decimalPlaces: Int = 8) async -> Double {\n        let task = Task.detached {\n            logger.log("starting \\(index)")                   // I wouldn\xe2\x80\x99t generally log in a unit test, but it\xe2\x80\x99s a quick visual confirmation that I\xe2\x80\x99m enjoying parallel execution\n            await self.increaseConcurrencyCount()\n\n            let threshold = pow(0.1, Double(decimalPlaces))\n            var isPositive = true\n            var denominator: Double = 1\n            var result: Double = 0\n            var increment: Double\n\n            repeat {\n                increment = 4 / denominator\n                if isPositive {\n                    result += increment\n                } else {\n                    result -= increment\n                }\n                isPositive.toggle()\n                denominator += 2\n            } while increment >= threshold\n\n            logger.log("finished \\(index)")\n            await self.decreaseConcurrencyCount()\n\n            return result\n        }\n\n        count += 1\n\n        return await task.value\n    }\n\n    func increaseConcurrencyCount() {\n        degreeOfConcurrency += 1\n        if degreeOfConcurrency > maxDegreeOfConcurrency { maxDegreeOfConcurrency = degreeOfConcurrency}\n    }\n\n    func decreaseConcurrencyCount() {\n        degreeOfConcurrency -= 1\n    }\n\n    func isAcceptable(_ result: Double) -> Bool {\n        return abs(.pi - result) < 0.0001\n    }\n}\n

   \n
\n

8. 请注意，如果在模拟器上测试/运行，协作线程池会受到一定限制，不会表现出与在实际设备上看到的相同程度的并发性。

   \n
\n

9. 另请注意，如果您正在测试特定测试是否表现出并行执行，您可能希望禁用测试本身的并行执行，以便其他测试不会占用您的核心，从而阻止任何给定的特定测试享受并行执行。

   \n

   

   \n
\n

\n