addTask和addTaskUnlessCancelled Swift的结构化并发性差异

Question

我在打电话给addTask()和addTaskUnlessCancelled之间有混淆。根据定义，组中的addTask()将无条件地向group添加一个新任务。

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()，它也在将所有的任务添加到组中。那么，这里和返回值有什么区别，只有当true抛出错误时才返回Task？

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是很好的建议。尽管如此，如果您不取消组(例如，取消任务比取消组( IMHO)更常见)，那么还不完全清楚是否需要addTaskUnlessCancelled。

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

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()
    }
}