ios - addTask 和 addTaskUnlessCancelled Swift 之间的结构化并发差异

Question

我对调用 addTask() 和 addTaskUnlessCancelled 感到困惑。根据定义，您的组上的 addTask() 将无条件地将新任务添加到组

func testCancellation() async {
    do {
        try await withThrowingTaskGroup(of: Void.self) { group -> Void in
            group.addTaskUnlessCancelled {
                print("added")
                try await Task.sleep(nanoseconds: 1_000_000_000)
                throw ExampleError.badURL
            }
            group.addTaskUnlessCancelled {
                print("added")
                try await Task.sleep(nanoseconds: 2_000_000_000)
                print("Task is cancelled: \(Task.isCancelled)")
            }

            group.addTaskUnlessCancelled {
                print("added")
                try await Task.sleep(nanoseconds: 5_000_000_000)
                print("Task is cancelled: \(Task.isCancelled)")
            }
            group.cancelAll()
            try await group.next()

        }
    } catch {
        print("Error thrown: \(error.localizedDescription)")
    }
}

如果您想避免将任务添加到已取消的组中，我们必须使用 addTaskUnlessCancelled() 方法。但即使使用group.cancelAll()，它也会将所有任务添加到组中。那么这里和仅当 Task 抛出一些错误时才返回 true 的返回值有什么区别？

Answer 1

简短回答

addTaskUnlessCancelled的美德是:

• 如果组被取消，它甚至不会启动任务；和
• 它允许您在团体取消时添加提前退出的功能。

有兴趣的话可以对比 addTask 的源码与 addTaskUnlessCancelled 的，就在它的下面。

---

长答案

我认为如果我们改变任务组随着时间的推移处理请求(例如消耗 AsyncSequence )，这个问题就得到了最好的例证。

考虑以下因素

func a() async {
    await withTaskGroup(of: Void.self) { group in
        for await i in tickSequence() {
            group.addTask(operation: { await self.b() })

            if i == 3 {
                group.cancelAll()
            }
        }
    }
}

func b() async {
    let start = CACurrentMediaTime()
    while CACurrentMediaTime() - start < 3 { }
}

func tickSequence() -> AsyncStream<Int> {
    AsyncStream { continuation in
        Task {
            for i in 0 ..< 12 {
                try await Task.sleep(nanoseconds: NSEC_PER_SEC / 2)

                continuation.yield(i)
            }
            continuation.finish()
        }
    }
}

在此示例中，b正在等待价格变动序列事件，正在调用c为每一个。但我们在添加第四个任务后取消了该组。结果如下:

所以我们可以看到路标ⓢ，在添加第四个任务后，该组被取消，但它继续进行并没有受到影响。

但那是因为b不响应取消。假设您解决了这个问题:

func b() async {
    let start = CACurrentMediaTime()
    while CACurrentMediaTime() - start < 3 {
        if Task.isCancelled { return }
    }
}

您现在会看到如下行为:

这样更好，如 b现在正在取消，但是 a尽管任务已被取消，但仍在尝试将任务添加到组中。你可以看到b正在重复运行(尽管至少现在立即终止)并且由 a 启动的任务没有及时完成。

但是，如果您使用 group.addTaskUnlessCancelled它不仅不会向组中添加新任务(即，它不依赖于任务的取消功能)，而且还允许您在组被取消时退出。

func a() async {
    await withTaskGroup(of: Void.self) { group in
        for await i in tickSequence() {
            guard group.addTaskUnlessCancelled(operation: { await self.b() })
            else { break }

            if i == 3 {
                group.cancelAll()
            }
        }
    }
}

这会产生所需的行为:

显然，这是一个相当人为的示例，明确构建是为了说明差异，但在许多情况下， addTask 之间的差异和addTaskUnlessCancelled不那么明显。但希望上面的内容能够说明其中的差异。

底线，如果您可能在向该组添加其他任务的过程中取消该组，addTaskUnlessCancelled是很好的建议。话虽如此，如果您不取消组(例如，取消任务比取消组更常见，恕我直言)，则不完全清楚 addTaskUnlessCancelled 的频率是多少。确实会被需要。

---

注意，在上面，我删除了所有 OSLog和路标代码，用于在 Xcode Instruments 中生成所有上述间隔和路标(因为我不想分散对手头问题的注意力)。但是，如果您有兴趣，这是实际的代码:

import os.log

private let log = OSLog(subsystem: "Test", category: .pointsOfInterest)

func a() async {
    await withTaskGroup(of: Void.self) { group in
        let id = log.begin(name: #function, "begin")
        defer { log.end(name: #function, "end", id: id) }

        for await i in tickSequence() {
            guard group.addTaskUnlessCancelled(operation: { await self.b() }) else { break }

            if i == 3 {
                log.event(name: #function, "Cancel")
                group.cancelAll()
            }
        }
    }
}

func b() async {
    let id = log.begin(name: #function, "begin")
    defer { log.end(name: #function, "end", id: id) }

    let start = CACurrentMediaTime()
    while CACurrentMediaTime() - start < 3 {
        if Task.isCancelled { return }
    }
}

func tickSequence() -> AsyncStream<Int> {
    AsyncStream { continuation in
        Task {
            for i in 0 ..< 12 {
                try await Task.sleep(nanoseconds: NSEC_PER_SEC / 2)

                continuation.yield(i)
            }
            continuation.finish()
        }
    }
}

与

extension OSLog {
    func event(name: StaticString = "Points", _ string: String) {
        os_signpost(.event, log: self, name: name, "%{public}@", string)
    }

    /// Manually begin an interval
    func begin(name: StaticString = "Intervals", _ string: String) -> OSSignpostID {
        let id = OSSignpostID(log: self)
        os_signpost(.begin, log: self, name: name, signpostID: id, "%{public}@", string)
        return id
    }

    /// Manually end an interval
    func end(name: StaticString = "Intervals", _ string: String, id: OSSignpostID) {
        os_signpost(.end, log: self, name: name, signpostID: id, "%{public}@", string)
    }

    func interval<T>(name: StaticString = "Intervals", _ string: String, block: () throws -> T) rethrows -> T {
        let id = OSSignpostID(log: self)

        os_signpost(.begin, log: self, name: name, signpostID: id, "%{public}@", string)
        defer { os_signpost(.end, log: self, name: name, signpostID: id, "%{public}@", string) }
        return try block()
    }
}