标签: structured-concurrency

我看过Explore structured concurrency in Swift视频和我能找到的其他相关视频/文章/书籍（Sundell 的 swift、用 swift 进行黑客攻击、Ray Renderlich），但所有示例都非常琐碎 - 异步函数通常只有 1 个异步调用。这在现实生活中的代码中应该如何工作？

例如：

...
        task = Task {
            var longRunningWorker: LongRunningWorker? = nil

            do {
                var fileURL = state.fileURL
                if state.needsCompression {
                    longRunningWorker = LongRunningWorker(inputURL: fileURL)
                    fileURL = try await longRunningWorker!.doAsyncWork()
                }

                let urls = try await ApiService.i.fetchUploadUrls()

                if let image = state.image, let imageData = image.jpegData(compressionQuality: 0.8) {
                    guard let imageUrl = urls.signedImageUrl else {
                        fatalError("Cover art supplied but art upload URL is nil")
                    } …

在 GCD 中我只是调用：

DispatchQueue.main.asyncAfter(deadline: .now() + someTimeInterval) { ... }

但我们开始迁移到结构化并发。

我尝试了以下代码：

extension Task where Failure == Error {
    static func delayed(
        byTimeInterval delayInterval: TimeInterval,
        priority: TaskPriority? = nil,
        operation: @escaping @Sendable () async throws -> Success
    ) -> Task {
        Task(priority: priority) {
            let delay = UInt64(delayInterval * 1_000_000_000)
            try await Task<Never, Never>.sleep(nanoseconds: delay)
            return try await operation()
        }
    }
}

用法：

Task.delayed(byTimeInterval: someTimeInterval) {
    await MainActor.run { ... }
}

但它似乎相当于：

DispatchQueue.global().asyncAfter(deadline: .now() + someTimeInterval) { …

以下代码在 Swift 5.5（测试版）中是合法的：

class Dog {
    var name = "rover"
    var friend : Dog? = nil
}
actor MyActor {
    let dog = Dog()
}
func test() async {
    let act = MyActor()
    act.dog.name = "fido"
    act.dog.friend = Dog()
    act.dog.friend?.name = "fido"
}

为什么这是合法的？dog 属性是共享状态，不是吗？我们不是有同时在不同线程上访问演员的狗的危险吗？这不是演员应该保护我们的吗？

奇怪的是，如果演员的dog属性是用var而不是声明的let，我们将不得不await在访问期间说出来。为什么会有所不同？Dog 是一个引用类型；它就地可变，并且无论是否用let或声明，它都以完全相同的方式可变var。

我希望能够让该函数doSomething()不class B存在async并且不阻止它的调用者线程。但使用以下代码我收到此错误：

无法将“() async -> Void”类型的函数传递给需要同步函数类型的参数

并且 xcode 尝试强制该doSomething()函数class B成为async

class A {
    func doSomething() async throws {
       //perform 
    }
}

class B {
   let a = A()
   func doSomething() {
      DispatchQueue.global().async {
        do {
           await try a.doSomething()
        } catch {
           print(error)
        }
     }
  }
}

在 Kotlin 中启动协程有多种方法。我发现了几个使用GlobalScope和的例子CoroutineScope。但后一个是在启动协程时直接创建的：

1. 使用GlobalScope：

   fun loadConfiguration() {
       GlobalScope.launch(Dispatchers.Main) {
          val config = fetchConfigFromServer() // network request
          updateConfiguration(config)
       }
   }
   

   Run Code Online (Sandbox Code Playgroud)

2. 使用CoroutineScope实例，在启动协程时直接创建：

   fun loadConfiguration() {
       CoroutineScope(Dispatchers.Main).launch {
           val config = fetchConfigFromServer() // network request
           updateConfiguration(config)
       }
   }
   

   Run Code Online (Sandbox Code Playgroud)

在这种情况下，这两种方法有区别吗？

第二种情况不是违反了结构化并发的原则吗？

今天，我将 SwiftUI 视图的 ViewModel 重构为结构化并发。它会触发网络请求，当请求返回时，更新@Published属性以更新 UI。由于我使用 aTask来执行网络请求，因此我必须返回来MainActor更新我的属性，并且我正在探索不同的方法来执行此操作。MainActor.run一种简单的方法是在 my 内部使用Task，效果很好。然后我尝试使用@MainActor，但不太理解这里的行为。

稍微简化一下，我的 ViewModel 看起来有点像这样：

class ContentViewModel: ObservableObject {
    
    @Published var showLoadingIndicator = false
    
    @MainActor func reload() {
        showLoadingIndicator = true
        
        Task {
            try await doNetworkRequest()
            showLoadingIndicator = false
        }
    }
    
    @MainActor func someOtherMethod() {
        // does UI work
    }
    
}

我原以为这不能正常工作。

首先，我预计 SwiftUI 会抱怨showLoadingIndicator = false主线程之外发生的情况。事实并非如此。所以我设置了一个断点，似乎连Taska 内的内容@MainActor都在主线程上运行。为什么这可能是另一天的问题，我想我还没有完全弄清楚Task。现在，让我们接受这一点。

因此，我预计 UI 在我的网络请求期间会被阻止 - 毕竟，它是在主线程上运行的。但事实也并非如此。网络请求运行，并且 …

例如，是否可以使用 RecursiveAction 与虚拟线程池（而不是 fork/join 池）结合使用（在我尝试设计不良的自定义工作之前）？

假设我们有一些异步代码。在某些时候，我们必须将其包装在 a 中Task {\xe2\x80\xa6}，以便从同步上下文运行它。\n那么规范的方法在哪里呢？ViewModel或者ViewController？

如果我们用Task {\xe2\x80\xa6}in包装它ViewModel，ViewModel函数就会变得有效同步，并且调用它们ViewController仍然需要所有这些完成/委托/闭包/RX 在异步工作完成后完成一些 UI 更新。

另一方面，如果我们将ViewModel函数标记为并从主体内部async调用它们，似乎可以解决问题。那么这是一条路吗？ViewControllerTask {\xe2\x80\xa6}

我试图理解 Kotlin 中的结构化并发，但无法理解这段代码。

fun main(): Unit = runBlocking {
    other(this)
}

suspend fun other(scope: CoroutineScope) {
    val job = scope.launch {
        scope.launch {
            delay(200)
            println("e")
        }
        println("a")
    }
    job.invokeOnCompletion {
        println("Complete")
    }
}

代码打印出来

a
Complete
e

如果我用 替换内部scope.launch调用launch，就像这样

suspend fun other(scope: CoroutineScope) {
    val job = scope.launch {
       launch {
            delay(200)
            println("e")
        }
        println("a")
    }
    job.invokeOnCompletion {
        println("Complete")
    }
}

它打印

a
e
Complete

这表明第一个示例不遵循结构化并发，因为父作业先于子作业完成。我的困惑是，为什么会发生这种情况？

我觉得在这种情况下scope.launch可能相当于调用launch（应该相当于this.launch并且 this 指的是scope …

我正在尝试使用现代 Swift Concurrency 构建分块文件上传机制。有一个流式文件读取器，我用它来逐块读取 1mb 大小的文件。它有两个闭包nextChunk: (DataChunk) -> Void和completion: () - Void。InputStream第一个被调用的次数与从块大小读取的数据一样多。

为了使该阅读器兼容 Swift Concurrency，我进行了扩展并创建了AsyncStream 似乎最适合这种情况的扩展。

public extension StreamedFileReader {
    func read() -> AsyncStream<DataChunk> {
        AsyncStream { continuation in
            self.read(nextChunk: { chunk in
                continuation.yield(chunk)
            }, completion: {
                continuation.finish()
            })
        }
    }
}

使用它，AsyncStream我迭代地读取一些文件并进行如下网络调用：

func process(_ url: URL) async {
    // ...
    do {
        for await chunk in reader.read() {
            let request = // ...
            _ = try await service.upload(data: chunk.data, request: …