Swift Concurrency Part 2

大綱(1/2)

Unstructured Task
- Task
- Cancellation
  - Task Cancellation
  - Cancellation handlers
- Task Piority
- Attached Task
- Executor
- ~~UnsafeCurrentTask~~

大綱(2/2)

Structured Task
- TaskGroup
- async let
- Child Task

Task

Concurrency 中最基礎的單元



Task {
    // do something...
}

所有異步函數皆執行在 `Task` 內



// Task {
    await something()
// }

Task State

digraph Task {
    rankdir = LR;
	suspended -> running [label = "schedule"]
    running -> completed [label = "return/throw"]
    running -> suspended [label = "await"]
}

Task State 實際

await 是潛在暫停點

digraph Task {
    rankdir = LR;
	suspended -> running [label = "schedule"]
    running -> completed [label = "return/throw"]
    running -> suspended [label = "await"]
    running -> running [label = "await"]
}

同步函數








func main() {
    all()
}
func all() {
    a1()
    a2()
    a3()
}

gantt
    title all
    dateFormat YYYY-MM-DD ss
    axisFormat %S

    section Main Thread
    a2               :a2, after a1, 2s
    a3               :a3, after a2, 2s
    a1               :a1, 2000-01-01 00, 2s
    all              :all, 2000-01-01 00, 6s
    main             :main, 2000-01-01 00, 6s

call stack

異步函數1





func all() async {
    a1()
    await a2()
    a3()
}

異步函數1-1

可在任意 thread 切換執行(executor)

Task: all
Job: a1 a2 a3

gantt
    title all
    dateFormat YYYY-MM-DD ss
    axisFormat %S

    section Global Executor1
    a1               :a1, 2000-01-01 00, 2s
    
    section Global Executor2
    a2               :a2, after a1, 2s
    
    section Global Executor3
    a3               :a3, after a2, 2s

異步函數1-2

await 是潛在暫停點，也有機會不暫停

Task: all
Job: all

gantt
    title all
    dateFormat YYYY-MM-DD ss
    axisFormat %S

    section Global Executor
    all              :2000-01-01 00, 6s

異步函數2(with actor)











actor A {
    let b = B()
    func all() async {
        a1()
        await b.a2()
        a3()
    }
}
actor B {
    func a2() {}
}

每個 actor 有各自的 executor

gantt
    title all
    dateFormat YYYY-MM-DD ss
    axisFormat %S

    section A
    a1               :a1, 2000-01-01 00, 2s
    a3               :a3, after a2, 2s
    
    section B
    a2               :a2, after a1, 2s

Task is Cooperative

Task Cancellation

Task Cancellation

只標記 Task 為取消


task.cancel() // task.isCancelled = true







Task {
    while !Task.isCancelled {
        // your works
    }
    
    return
}







func doSomething() async {
    while !Task.isCancelled {
        // your works
    }
    
    return
}



Task {
    try Task.checkCancellation()
}

上次的小結

cancel 只能停止

~~尚未開始的 Task~~
有支援取消流程的 Task

Default Executor

預設情況下

Task 排程到 default global concurrent executor

Executor

Executor 是一個服務(service)
- 接收提交過來的 Jobs 並且調度 Thread 來執行 Job
- 系統假設 executor 是可靠且執行 Job 時永不失敗

Executor 執行時是獨佔的(exclusive)，表示同時多個 job 提交時，並不會併發執行
不應該依照提交順序執行
- 實際上應以, 優先權 > 提交順序(submission order)

Swift 提供預設 Executor 實作


actor SomeActor {}

actor 以及 global actor 可替換其實作










@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
public protocol Actor : AnyObject, Sendable {
    nonisolated var unownedExecutor: UnownedSerialExecutor { get }
}

@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
public protocol GlobalActor {
    // ...
    static var sharedUnownedExecutor: UnownedSerialExecutor { get }
}

Cancellation handlers

一般 Task cancel

需要自行實作 cancel 流程

TaskCancellationHandler

在 Task 被取消時，立即呼叫 cancel handler



















func download(url: URL) async throws -> Data? {
  var urlSessionTask: URLSessionTask?

  return try withTaskCancellationHandler {
    return try await withUnsafeThrowingContinuation { continuation in
      urlSessionTask = URLSession.shared.dataTask(with: url) { data, _, error in
        if let error = error {
          // Ideally translate NSURLErrorCancelled to CancellationError here
          continuation.resume(throwing: error)
        } else {
          continuation.resume(returning: data)
        }
      }
      urlSessionTask?.resume()
    }
  } onCancel: {
    urlSessionTask?.cancel() // runs immediately when cancelled
  }
}

Alamofire + Rx
















public var data: Observable<Data> {
  return Observable.create { observer -> Disposable in
    self.dataRequest.responseData { (res) in
      do {
		try observer.onNext(res.result.get())
	  	observer.onCompleted()
	  } catch {
	  	observer.onError(error)
	  }

      return Disposables.create {
  	    self.dataRequest.cancel()
      }
    }
  }
}

Alamofire + Continuation + TaskCancellationHandler


















public var data: Data {
  get async throws {
    return try withTaskCancellationHandler {
      return try await withCheckedThrowingContinuation { continuation in
        self.dataRequest.responseData { (res) in
          do {
          	let value = try res.result.get()
			continuation.resume(returning: value)
          } catch {
			continuation.resume(throwing: error)
          }
        }
      }
    } onCancel: {
      self.dataRequest.cancel()
    }
  }
}

Attached Task



Task {
    // ...
}

Context inheritance

跑在 Task
沒跑在 Task

跑在 Task

繼承 TaskPriority





func xxx() async {
    Task {
        // 繼承 xxx `TaskPriority`
    }
}

跑在 Task

TaskLocal values
- copy value to inner task








@TaskLocal
static var traceID: TraceID?

print("traceID: \(traceID)") // traceID: nil

$traceID.withValue(1234) { // bind the value
  print("traceID: \(traceID)") // traceID: 1234
}

跑在 Task & 跑在 Actor

















actor One {
    var counter = 1 // mutable properties
    var notSendable = NotSendable() // non-sendable values
    func xxx() {
        Task {
            // 繼承 actor 的 Execution Context
            // 此 task 排程到 actor's executer
            
            // closure 變成 `actor-isolated`
            // 可以存取 actor 的私有狀態(`actor-isolated` state)
            // mutable properties
            // non-sendable values
            counter += 1
            notSendable.xxx()
        }
    }
}

沒跑在 Task









func xxx() {
    // Main Thread
    Task {
        // 諮詢 `runtime` 並推導出最可能的 `TaskPriority`
        // 例如問 `current thread priority`
        
        // 排程到 `global concurrent executor`
    }
}

Actor context propagation

傳遞給 Task init 的閉包將隱式繼承 actor execution context 和形成閉包的上下文的隔離














func notOnActor(_: @Sendable () async -> Void) { }

actor A {
  func f() {
    notOnActor {
      await g() // must call g asynchronously, because it's a @Sendable closure
    }
    Task {
      g() // okay to call g synchronously, even though it's @Sendable
    }
  }
  
  func g() { }
}

Implicit "self"

要求使用 self 的意圖，是為了提醒開發者 capture self 會有潛在 reference cycle
傳遞給 Task init 的閉包會被立刻執行，而且只會在內部 reference，因此 explicit self
- 沒有傳達有效資訊
- 不需要(not be required)














func acceptEscaping(_: @escaping () -> Void) { }

class C {
  var counter: Int = 0
  
  func f() {
    acceptEscaping {
      counter = counter + 1   // error: must use "self." because the closure escapes
    }
    Task {
      counter = counter + 1   // okay: implicit "self" is allowed here
    }
  }
}

TaskPriority

Task 的 Priority 是用來協助 Executor 做出排程決策(scheduling decisions)

List

high
medium
low
userInitiated
utility
background