• structured concurrency recap

    • 비동기 코드를 이해하기 쉬운 형태로 구조화하기 좋다.

      • 어디서 동시성 실행 흐름이 시작되는가
        • async let
        • TaskGroup
        • Task, Task.detached
      • 그 결과가 어떻게 다시 합쳐지는가
        • await으로 나타나는 suspention point에서
    • 모든 작업이 structured인 건 아니다.

      • async let, taskgroup은 structured
        • structured는 선언된 스코프내에서만 유효하고, 스코프를 벗어나면 자동으로 캔슬된다.
      • Task와 Task.detached는 unstructured
    • 가능하면 structured Task를 사용하자.

      // bad
      func makeSoup(order: Order) async throws -> Soup {
          let boilingPot = Task { try await stove.boilBroth() }
          let choppedIngredients = Task { try await chopIngredients(order.ingredients) }
          let meat = Task { await marinate(meat: .chicken) }
          let soup = await Soup(meat: meat.value, ingredients: choppedIngredients.value)
          return await stove.cook(pot: boilingPot.value, soup: soup, duration: .minutes(10))
      }
      
      // good
      
      func makeSoup(order: Order) async throws -> Soup {
          async let pot = stove.boilBroth()
          async let choppedIngredients = chopIngredients(order.ingredients)
          async let meat = marinate(meat: .chicken)
          let soup = try await Soup(meat: meat, ingredients: choppedIngredients)
          return try await stove.cook(pot: pot, soup: soup, duration: .minutes(10))
      }
      
    • 동시 실행 작업수를 특정할 수 있으면 async let을, 특정할 수 없으면 taskgroup을 사용한다.

      func chopIngredients(_ ingredients: [any Ingredient]) async -> [any ChoppedIngredient] {
          return await withTaskGroup(of: (ChoppedIngredient?).self,
                                     returning: [any ChoppedIngredient].self) { group in
               // Concurrently chop ingredients
               for ingredient in ingredients {
                   group.addTask { await chop(ingredient) }
               }
               // Collect chopped vegetables
               var choppedIngredients: [any ChoppedIngredient] = []
               for await choppedIngredient in group {
                   if choppedIngredient != nil {
                      choppedIngredients.append(choppedIngredient!)
                   }
               }
               return choppedIngredients
          }
      }
      
  • Task hierarchy

    스크린샷 2023-06-14 오후 11.04.44.png

  • Task Cancellation

    • Unstructured는 명시적으로 취소를 해야만 취소가 된다.

      task.cancel()
      
    • structured 부모가 취소되면 자식도 알아서 취소가 된다.

    • 다만 취소는 cooperative 형태기 때문에 플래그를 설정해놓기만 하고, 자식이 이를 확인하고 대응해야 한다.

      • 자식이 확인안하면 그냥 작업 끝날때까지 돌아간다는 것
      • 이것도 타이밍 문제기 때문에 취소 검사를 이미 했는데 실제로 취소가 되버리면 그냥 작업이 되버릴 수도 있다.
      • 무거운 작업을 하기 전에 반드시 취소 검사를 할 것. 취소 검사는 동기적이기 때문에 어디서든 할 수 있다.
      func makeSoup(order: Order) async throws -> Soup {
          async let pot = stove.boilBroth()
      
          guard !Task.isCancelled else { // bool 값 확인 후 흐름 끊기
              throw SoupCancellationError()
          }
      
          async let choppedIngredients = chopIngredients(order.ingredients)
          async let meat = marinate(meat: .chicken)
          let soup = try await Soup(meat: meat, ingredients: choppedIngredients)
          return try await stove.cook(pot: pot, soup: soup, duration: .minutes(10))
      }
      
      func chopIngredients(_ ingredients: [any Ingredient]) async throws -> [any ChoppedIngredient] {
          return try await withThrowingTaskGroup(of: (ChoppedIngredient?).self,
                                         returning: [any ChoppedIngredient].self) { group in
              try Task.checkCancellation() // 취소시 에러 던지기
              
              // Concurrently chop ingredients
              for ingredient in ingredients {
                  group.addTask { await chop(ingredient) }
              }
      
              // Collect chopped vegetables
              var choppedIngredients: [any ChoppedIngredient] = []
              for try await choppedIngredient in group {
                  if let choppedIngredient {
                      choppedIngredients.append(choppedIngredient)
                  }
              }
              return choppedIngredients
          }
      }
      
    • 이벤트 기반 취소 검사

      • withTaskCancellationHandler(operation: onCancel:)

        스크린샷 2023-06-14 오후 11.24.26.png

      • 특히 asyncsequence를 구현할 때 유용하다.

        • asyncSequence의 다음값을 기다리고 있기 때문에 현재 task는 멈춰있어서 polling이 불가능하다.
        actor Cook {
            func handleShift<Orders>(orders: Orders) async throws
               where Orders: AsyncSequence,
                     Orders.Element == Order {
        
                for try await order in orders {
                    let soup = try await makeSoup(order)
                    // ...
                }
            }
        }
        
        • 그래서 cancellationHandler를 사용해서 취소를 감지하고, for await-loop를 강제로 탈줄하게 해줘야 한다.

          • 실제 구현은 AsyncIterator의 next 함수에서 한다.
          • 많은 AsyncSequence가 state machine으로 구현되기 때문에 이를 사용해서 sequence를 취소하자.
          public func next() async -> Order? {
              return await withTaskCancellationHandler {
                  let result = await kitchen.generateOrder()
                  guard state.isRunning else {
                      return nil
                  }
                  return result
              } onCancel: {
                  state.cancel()
              }
          }
          
        • 근데 onCancel은 동기적으로 돌고, 실제 작업은 비동기라서 여기서 변경 가능한 state를 공유하게 되면 문제가 될 수있다.

          • actor가 상태를 캡슐화하기는 좋지만, 여기서는 개별 프로퍼티를 읽고 변경해야 하기 때문에 좋은 선택이 아니다. 심지어 actor는 돌아가는 작업의 순서도 보장할 수 없다.
          • 그래서 다른 동기화 도구를 사용해야 한다.
            • Swift-Atomic을 예시에서는 썼지만, lock이나 dispatchQueue도 쓸 수 있다.

              private final class OrderState: Sendable {
                  let protectedIsRunning = ManagedAtomic<Bool>(true)
                  var isRunning: Bool {
                      get { protectedIsRunning.load(ordering: .acquiring) }
                      set { protectedIsRunning.store(newValue, ordering: .relaxed) }
                  }
                  func cancel() { isRunning = false }
              }
              
  • Task Priority

    • 시스템에 해당 작업이 얼마나 시급한지를 알려주는 값이다.
      • 사용자 입력은 빠르게 처리하지 않으면 앱이 멈춘 것으로 보이기 떄문에 빠르게 처리해야한다.
      • 컨텐츠 prefetching 같은 경우는 백그라운드에서 해도 된다.
    • priority inversion: 높은 priority의 작업이 낮은 priority의 결과를 기다리는 현상
      • 자식 작업은 부모 작업의 priority를 이어받는다.

        스크린샷 2023-06-14 오후 11.57.09.png

      • priority가 높은 작업이 await을 하게 되면, 그 자식 작업들은 모두 부모의 높은 priority로 승격된다.

        • 한번 올라간 priority는 다시 내려가지 않는다.

        스크린샷 2023-06-14 오후 11.59.14.png

      • 동시성 런타임은 우선순위 큐기 때문에 priority가 높은 작업을 우선적으로 수행한다.

  • Task group patterns

    • taskgroup으로 너무 많은 작업을 실행하면 다른 작업을 실행하기가 어렵게 된다.

    • 그래서 최대 작업 갯수를 제한해주는 것도 좋다.(예시는 최대 3개)

      func chopIngredients(_ ingredients: [any Ingredient]) async -> [any ChoppedIngredient] {
          return await withTaskGroup(of: (ChoppedIngredient?).self,
                                     returning: [any ChoppedIngredient].self) { group in
              // Concurrently chop ingredients
              let maxChopTasks = min(3, ingredients.count)
              for ingredientIndex in 0..<maxChopTasks {
                  group.addTask { await chop(ingredients[ingredientIndex]) }
              }
      
              // Collect chopped vegetables
              var choppedIngredients: [any ChoppedIngredient] = []
              var nextIngredientIndex = maxChopTasks
              for await choppedIngredient in group {
                  if nextIngredientIndex < ingredients.count {
                      group.addTask { await chop(ingredients[nextIngredientIndex]) }
                      nextIngredientIndex += 1
                  }
                  if let choppedIngredient {
                      choppedIngredients.append(choppedIngredient)
                  }
              }
              return choppedIngredients
          }
      }
      
    • 이를 패턴화 할 수 있다.

      • 최대 작업수만큼만 넣고, 하나 끝나는대로 나머지 하나 더 넣고
      withTaskGroup(of: Something.self) { group in
          for _ in 0..<maxConcurrentTasks {
              group.addTask { }
          }
          while let <partial result> = await group.next() {
              if !shouldStop { 
                  group.addTask { }
              }
          }
      }
      
    • withDiscardingTaskGroup

      • taskGroup으로 여러 작업을 돌리지만 작업의 결과는 필요하지 않을 수 있다.

        func run() async throws {
            try await withThrowingTaskGroup(of: Void.self) { group in
                for cook in staff.keys {
                    group.addTask { try await cook.handleShift() }
                }
        
                group.addTask {
                    // keep the restaurant going until closing time
                    try await Task.sleep(for: shiftDuration)
                }
        
                try await group.next()
                // cancel all ongoing shifts
                group.cancelAll()
            }
        }
        
      • Swift 5.9에 추가된 withDiscardingTaskGroup을 쓰면, 각 작업이 쓰는 리소스가 작업이 끝나는 즉시 해제된다.

        • 자식 중에 하나라도 에러를 던지면 남은 모든 작업도 자동으로 취소된다.
        func run() async throws {
            try await withThrowingDiscardingTaskGroup { group in
                for cook in staff.keys {
                    group.addTask { try await cook.handleShift() }
                }
        
                group.addTask { // keep the restaurant going until closing time
                    try await Task.sleep(for: shiftDuration)
                    throw TimeToCloseError()
                }
            }
        }
        
  • Task-local values

    • 특정 작업(정확히는 작업 계층)과 연관된 데이터 조각

    • 전역적이지만, 여기에 바인딩 된 값은 특정 작업 계층에서만 보인다.

    • TaskLocal은 static 프로퍼티에 @TaskLocal 프로퍼티 래퍼를 붙여서 선언한다.

      • 기본값을 항상 요구하기 때문에 옵셔널로 두는게 편하다.
    • tasklocal은 직접 값을 할당할 수는 없고, 반드시 특정 스코프 안에서만 쓸 수 있다.

      actor Kitchen {
          @TaskLocal static var orderID: Int?
          @TaskLocal static var cook: String?
          func logStatus() {
              print("Current cook: \\(Kitchen.cook ?? "none")")
          }
      }
      
      let kitchen = Kitchen()
      await kitchen.logStatus()
      await Kitchen.$cook.withValue("Sakura") {
          await kitchen.logStatus() // 이 안에서만 cook 값이 Sakura
      }
      await kitchen.logStatus()
      
    • 각 작업은 taskLocal을 저장해놓을 수 있는 연관 저장소인 TaskLocalStorage를 가진다.

    • 자식은 해당 값에 접근할 때 자신에게 값이 없으면 계층을 타고 올라가면서 해당 값이 있는지를 찾는다.

      • 루트까지 올라가도 못찾으면 기본값을 준다.

      스크린샷 2023-06-15 오전 1.02.22.png

    • Swift 런타임은 이를 최적화해서 계층을 올라가지 않고 해당 값이 있는 곳의 포인터를 가지고 있어 바로 거기로 간다.

      스크린샷 2023-06-15 오전 1.03.46.png

    • ex. 서버에서의 로깅

      • 단계를 추적하기 위해서, 로그에 여러 정보를 넣는다.

      • 근데 이 부분은 장황하고 반복적이여서 실수하기도 쉽다.

        func makeSoup(order: Order) async throws -> Soup {
             log.debug("Preparing dinner", [
               "cook": "\\(self.name)",
               "order-id": "\\(order.id)",
               "vegetable": "\\(vegetable)",
             ])
             // ... 
        }
        
         func chopVegetables(order: Order) async throws -> [Vegetable] {
             log.debug("Chopping ingredients", [
               "cook": "\\(self.name)",
               "order-id": "\\(order.id)",
               "vegetable": "\\(vegetable)",
             ])
             
             async let choppedCarrot = try chop(.carrot)
             async let choppedPotato = try chop(.potato)
             return try await [choppedCarrot, choppedPotato]
        }
        
        func chop(_ vegetable: Vegetable, order: Order) async throws -> Vegetable {
            log.debug("Chopping vegetable", [
              "cook": "\\(self.name)",
              "order-id": "\\(order)", // order.id 여야 한다!
              "vegetable": "\\(vegetable)",
            ])
            // ...
        }
        
      • Apple 디바이스면 OSLog를 직접 쓰겠지만, 이미 돌아가는 앱을 클라우드에서 실행하는 경우에는 다른 솔루션이 필요하다.

        • 그래서 있는게 Swift-Log다. 플랫폼별 구현을 추상화해서 플랫폼 독립적인 로깅을 가능하게 한다.
      • MetadataProvider API가 Swift-Log 1.5 버전에 추가되었다.

        • 관련된 이벤트를 로깅할 때 필요한 데이터를 일관적으로 실어보낼 수 있게 해준다.
        • dictonary 같은 구조를 가진다.
        • 라이브러리마다 각자 Metadata를 가질 수 있기 때문에 이를 합쳐주고 이를 기반으로 로깅 시스템을 초기화한다.
        • 로그를 남길 때마다 메타데이터를 자동으로 함께 넣어준다.
    • 자식 작업 뿐 아니라 Task{ } 로 만들어도 taskLocal 값을 이어받는다. Task.detached {} 만 이어받지 않는다.

  • Task traces