使 Swift 并发中的任务串行运行

我有一个基于文档的应用程序，它使用结构作为其主要数据/模型。由于模型是（的子类）的属性NSDocument需要从主线程访问它。到目前为止一切都很好。

但对数据的某些操作可能需要相当长的时间，我想为用户提供一个进度条。这就是问题开始的地方。特别是当用户从 GUI 快速连续启动两个操作时。

如果我在模型上同步运行操作（或以“正常”方式）Task {}）我得到了正确的串行行为，但主线程被阻止，因此我无法显示进度条。 （选项A）

如果我在模型上运行操作Task.detached {}我可以更新进度条，但根据模型上操作的运行时间，用户的第二个操作可能会在第一个操作之前完成，从而导致模型的无效/意外状态。这是由于await分离任务中需要的语句（我认为）。 （选项B）。

所以我想要 a) 释放主线程来更新 GUI，b) 确保每个任务在另一个（排队的）任务开始之前运行完全完成。使用后台串行调度队列很有可能实现这一点，但我正在尝试切换到新的 Swift 并发系统，该系统也用于在访问模型之前执行任何准备工作。

我尝试使用全局演员，因为这似乎是某种串行后台队列，但它也需要await声明。尽管模型中出现意外状态的可能性降低了，但这仍然是可能的。

我写了一些小代码来演示这个问题：

该模型：

struct Model {
    var doneA = false
    var doneB = false

    mutating func updateA() {
        Thread.sleep(forTimeInterval: 5)
        doneA = true
    }

    mutating func updateB() {
        Thread.sleep(forTimeInterval: 1)
        doneB = true
    }
}

和文件（省略标准NSDocument覆盖）：

@globalActor
struct ModelActor {
    actor ActorType { }

    static let shared: ActorType = ActorType()
}

class Document: NSDocument {
    var model = Model() {
        didSet {
            Swift.print(model)
        }
    }

    func update(model: Model) {
        self.model = model
    }

    @ModelActor
    func updateModel(with operation: (Model) -> Model) async {
        var model = await self.model
        model = operation(model)
        await update(model: model)
    }

    @IBAction func operationA(_ sender: Any?) {
        //Option A
//        Task {
//            Swift.print("Performing some A work...")
//            self.model.updateA()
//        }

        //Option B
//        Task.detached {
//            Swift.print("Performing some A work...")
//            var model = await self.model
//            model.updateA()
//            await self.update(model: model)
//        }

        //Option C
        Task.detached {
            Swift.print("Performing some A work...")
            await self.updateModel { model in
                var model = model
                model.updateA()
                return model
            }
        }
    }

    @IBAction func operationB(_ sender: Any?) {
        //Option A
//        Task {
//            Swift.print("Performing some B work...")
//            self.model.updateB()
//        }

        //Option B
//        Task.detached {
//            Swift.print("Performing some B work...")
//            var model = await self.model
//            model.updateB()
//            await self.update(model: model)
//        }

        //Option C
        Task.detached {
            Swift.print("Performing some B work...")
            await self.updateModel { model in
                var model = model
                model.updateB()
                return model
            }
        }
    }
}

单击“操作 A”，然后单击“操作 B”应该会生成一个包含两个的模型true的。但情况并非总是如此。

有没有办法确保操作 A 在我进行操作 B 之前完成并让主线程可用于 GUI 更新？

EDIT根据罗布的回答，我得出以下结论。我这样修改它是因为我可以等待创建的操作并向原始调用者报告任何错误。我认为通过将所有代码包含在一个单一的代码中更容易理解正在发生的事情update函数，所以我选择执行独立任务而不是actor。我还从任务中返回中间模型，否则可能会使用旧模型。

class Document {
    func updateModel(operation: @escaping (Model) throws -> Model) async throws {
        //Update the model in the background
        let modelTask = Task.detached { [previousTask, model] () throws -> Model in
            var model = model

            //Check whether we're cancelled
            try Task.checkCancellation()

            //Check whether we need to wait on earlier task(s)
            if let previousTask = previousTask {
                //If the preceding task succeeds we use its model
                do {
                    model = try await previousTask.value
                } catch {
                    throw CancellationError()
                }
            }

            return try operation(model)
        }


        previousTask = modelTask
        defer { previousTask = nil } //Make sure a later task can always start if we throw

        //Wait for the operation to finish and store the model
        do {
            self.model = try await modelTask.value
        } catch {
            if error is CancellationError { return }
            else { throw error }
        }
    }
}

呼叫方：

@IBAction func operationA(_ sender: Any?) {
    //Option D
    Task {
        do {
            try await updateModel { model in
                var model = model
                model.updateA()
                return model
            }
        } catch {
            presentError(error)
        }
    }
}

它似乎做了我需要的任何事情，即对文档上的属性进行排队更新，可以等待并返回错误，就像所有事情都发生在主线程上一样。 唯一的缺点似乎是在调用方，由于需要使模型成为一个闭包，所以闭包非常冗长。var并明确返回它。

显然，如果您的任务没有任何await或其他暂停点，您只需使用演员，而不是制作该方法async，它会自动按顺序执行它们。

但是，在处理异步 Actor 方法时，必须认识到 Actor 是可重入的（请参阅SE-0306：演员 - 演员重入 https://github.com/apple/swift-evolution/blob/main/proposals/0306-actors.md#actor-reentrancy）。如果您确实尝试串行运行一系列异步任务，您将需要手动让每个后续任务等待前一个任务。例如。，

actor Foo {
    private var previousTask: Task<Void, Error>?

    func add(block: @Sendable @escaping () async throws -> Void) {
        previousTask = Task { [previousTask] in
            let _ = await previousTask?.result

            return try await block()
        }
    }
}

上述内容有两个微妙的方面：

1. 我使用的捕获列表[previousTask]确保获得先前任务的副本。

2. 我表演await previousTask?.value inside新任务，而不是之前的任务。

   如果您在创建新任务之前等待，则会发生竞赛，如果您启动三个任务，第二个和第三个任务都将等待first任务，即第三个任务不等待第二个任务。

而且，也许不用说，因为这是在一个 actor 内，所以它避免了对分离任务的需要，同时保持主线程空闲。

注意，使用时非结构化并发 https://docs.swift.org/swift-book/documentation/the-swift-programming-language/concurrency#Unstructured-Concurrency (i.e., Task {…} or Task.detached {…})，您自行承担处理取消的责任，例如使用withTaskCancellationHandler https://developer.apple.com/documentation/swift/withtaskcancellationhandler(operation:oncancel:)/:

actor Foo<Value: Sendable> {
    private var previousTask: Task<Value, Error>?

    func add(block: @Sendable @escaping () async throws -> Value) async throws -> Value {
        let task = Task { [previousTask] in
            try await withTaskCancellationHandler {
                let _ = try await previousTask?.value
            } onCancel: {
                previousTask?.cancel()
            }

            return try await block()
        }

        previousTask = task

        return try await withTaskCancellationHandler {
            try await task.value
        } onCancel: {
            task.cancel()
        }
    }
}

我还将其扩展为可能返回值的块。

因此，例如，我在这里添加了四个任务（即Task.sleep两秒然后返回一个随机值）：

或者，如果您在第三个任务中途取消第四个任务：

（不用说，这假设你添加的任务支持取消，抛出CancellationError https://developer.apple.com/documentation/swift/cancellationerror如果取消等。标准Apple API，例如URLSession，执行所有这些操作，但正如您所看到的，如果引入非结构化并发，则需要小心。）

上面的内容有点脆弱，所以我可能建议异步序列（例如，任何符合AsyncSequence https://developer.apple.com/documentation/swift/asyncsequence/协议，例如AsyncStream https://developer.apple.com/documentation/swift/asyncstream或您自己的自定义异步序列），这也可以为您提供串行行为。

Or, AsyncChannel https://github.com/apple/swift-async-algorithms/blob/main/Sources/AsyncAlgorithms/AsyncAlgorithms.docc/Guides/Channel.md from Swift 异步算法 https://github.com/apple/swift-async-algorithms是处理触发某些代码块的串行执行的请求管道的另一种好方法。

例如，这是一个使用的串行下载管理器AsyncChannel和一个简单的for-await-in循环实现串行行为：

actor SerialDownloadManager {
    static let shared = SerialDownloadManager()

    private let session: URLSession = …
    private let urls = AsyncChannel<URL>()

    private init() {
        Task { try await startDownloader() }
    }

    // this sends URLs on the channel

    func append(_ url: URL) async {
        await urls.send(url)
    }
}

private extension SerialDownloadManager {
    func startDownloader() async throws {
        let folder = try FileManager.default
            .url(for: .applicationSupportDirectory, in: .userDomainMask, appropriateFor: nil, create: true)
            .appending(component: "downloads")

        try? FileManager.default.createDirectory(at: folder, withIntermediateDirectories: true)

        // this consumes the URLs on the channel

        for await url in urls {
            // if you want to observe in "points of interest"
            //
            // let id = OSSignpostID(log: poi)
            // os_signpost(.begin, log: poi, name: "Download", signpostID: id, "%{public}@", url.lastPathComponent)
            // defer { os_signpost(.end, log: poi, name: "Download", signpostID: id) }

            // download

            let (location, response) = try await self.session.download(from: url, delegate: nil)

            if let response = response as? HTTPURLResponse, 200 ..< 300 ~= response.statusCode {
                let destination = folder.appending(component: url.lastPathComponent)
                try? FileManager.default.removeItem(at: destination)
                try FileManager.default.moveItem(at: location, to: destination)
            }
        }
    }
}

然后你可以做这样的事情：

func appendUrls() async {
    for i in 0 ..< 10 {
        await SerialDownloadManager.shared.append(baseUrl.appending(component: "\(i).jpg"))
    }
}

产量：

或者，如果您愿意，您可以允许任务组进行受限并发，例如，一次执行 4 个任务：

actor DownloadManager {
    static let shared = DownloadManager()

    private let session: URLSession = …
    private let urls = AsyncChannel<URL>()
    private var count = 0
    private let maxConcurrency = 4       // change to 1 for serial downloads, but 4-6 is a good balance between benefits of concurrency, but not overtaxing server

    private init() {
        Task {
            do {
                try await startDownloader()
            } catch {
                logger.error("\(error, privacy: .public)")
            }
        }
    }

    func append(_ url: URL) async {
        await urls.send(url)
    }
}

private extension DownloadManager {
    func startDownloader() async throws {
        let folder = try FileManager.default
            .url(for: .applicationSupportDirectory, in: .userDomainMask, appropriateFor: nil, create: true)
            .appending(component: "downloads")

        try? FileManager.default.createDirectory(at: folder, withIntermediateDirectories: true)

        try await withThrowingTaskGroup(of: Void.self) { group in
            for await url in urls {
                count += 1
                if count > maxConcurrency { try await group.next() }

                group.addTask {
                    // if you want to observe in "points of interest"
                    //
                    // let id = OSSignpostID(log: poi)
                    // os_signpost(.begin, log: poi, name: "Download", signpostID: id, "%{public}@", url.lastPathComponent)
                    // defer { os_signpost(.end, log: poi, name: "Download", signpostID: id) }

                    // download

                    let (location, response) = try await self.session.download(from: url, delegate: nil)

                    if let response = response as? HTTPURLResponse, 200 ..< 300 ~= response.statusCode {
                        let destination = folder.appending(component: url.lastPathComponent)
                        try? FileManager.default.removeItem(at: destination)
                        try FileManager.default.moveItem(at: location, to: destination)
                    }
                }
            }

            try await group.waitForAll()
        }
    }
}

产量：

有关异步序列的更多信息，一般而言，请参阅 WWDC 2021 视频认识异步序列 https://developer.apple.com/videos/play/wwdc2021/10058/.