Modern Concurrency: Beyond the Basics

Transcript: 12. TaskGroup

You’ve used async let to run tasks concurrently, but what if you need to run a thousand tasks in parallel, or you don’t know until runtime how many tasks need to run in parallel? You need more than async let!

The answer is TaskGroup. You can create concurrency on the fly and safely process the results, while reducing the possibility of data races.

Here’s how you might use a task group:

//1
let images = try await withThrowingTaskGroup(
  of: Data.self
  returning: [UIImage].self
) { group in
  // 2
  for index in 0..<numberOfImages {
    let url = baseURL.appendingPathComponent("image\(index).png")
    // 3
    group.addTask {
      // 4
      return try await URLSession.shared
        .data(from: url, delegate: nil)
        .0
    }
  }
  // 5
  return try await group.reduce(into: [UIImage]()) { result, data in
    if let image = UIImage(data: data) {
      result.append(image)
    }
  }
}

You set each task’s return type as Data with the of argument. The group as a whole will return an array of UIImage. You could also have an explicit return type in the closure declaration and skip the returning argument.

Elsewhere in your code, you’ve calculated the number of images you want to fetch, and you loop through them here.

group is the ready-to-go ThrowingTaskGroup. Inside the for loop, you use group.addTask { ... } to add tasks to the group.

You perform the actual work of the task by fetching data from a URL.

Task groups conform to AsyncSequence so, as each task in the group completes, you collect the results into an array of images and return it.

Here’s a diagram of what the code does. The example code starts a variable number of concurrent tasks, and each one downloads an image. Finally, you assign the array with all the images to images.

You manage the group’s tasks with the following APIs:

addTask(priority:operation:) adds a task to the group for concurrent execution with the given (optional) priority.

addTaskUnlessCancelled(priority:operation:) is identical to addTask(...), except that it does nothing if the group is already canceled.

cancelAll() cancels the group. It cancels all currently running tasks, along with all tasks added in the future.

isCancelled returns true if the group is canceled.

isEmpty returns true if the group has completed all its tasks, or has no tasks to begin with.

waitForAll() waits until all tasks have completed. Use this when you need to execute some code after finishing the group’s work.

In this episode, you’ll work with the Sky app. It pretends to scan satellite images of the sky. Each image is divided into 20 sectors, and the app scans each sector, looking for signs of alien life. Each sector scan is independent of the other sector scans, so the app could do these concurrently. That’s what you’ll implement in this episode, using TaskGroup.

Open the starter project, then build and run. The view shows 3 indicators: number of scheduled tasks, current tasks-per-second ratio and number of completed scans.

By the end of this episode, the first indicator should be more than 1. Tap Engage systems

The button action should create and run scans for the 20 sectors, then display how long the complete scan took. At the moment, none of the action code is there. You’ll write it soon.

When you finish this episode, the duration should be much less than 20 seconds.

Stop the run, then open ScanModel.swift and see what’s here: ScanModel is an ObservableObject. It has several Published properties. The app’s views use these properties, so any updates must happen on the main queue. The @MainActor keyword places these properties on the main actor. You’ll learn more about actors later in this course.

/// Currently scheduled for execution tasks.
@MainActor @Published var scheduled = 0

/// Completed scan tasks per second.
@MainActor @Published var countPerSecond: Double = 0

/// Completed scan tasks.
@MainActor @Published var completed = 0

@Published var total: Int

@MainActor @Published var isCollaborating = false

Now, jump down to the extension: It already has two private utility methods to track task progress:

extension ScanModel {
  @MainActor
  private func onTaskCompleted() {
    completed += 1
    counted += 1
    scheduled -= 1

    countPerSecond = Double(counted) / Date().timeIntervalSince(started)
  }

  @MainActor
  private func onScheduled() {
    scheduled += 1
  }
}

onTaskCompleted() updates completed, counted, scheduled and countPerSecond, and onScheduled() updates scheduled, so both of these are MainActor methods.

Back in the ScanModel class, you have a convenience method worker(number:) to run a single task:

func worker(number: Int) async -> String {
  await onScheduled()

  let task = ScanTask(input: number)
  let result = await task.run()

  await onTaskCompleted()
  return result
}

• await onScheduled() updates the scheduled counter. Updating the UI should always be a fast operation, so the await here won’t significantly affect the progress of the scanning task.
• The method creates a new ScanTask with the given sector number, waits for the result of the asynchronous call to task.run(), then calls onTaskCompleted() to update the counters and the app’s UI on the main thread.
• And finally, it returns result:

So worker(number:) not only runs a single task, but also tracks the execution in the model’s state.

Because onTaskCompleted() and onScheduled() are MainActor methods, you can safely update the Published properties, even while running multiple copies of this method in parallel.

Serial tasks

Now, add some code to runAllTasks():

var scans: [String] = []
for number in 0..<total {
  scans.append(try await worker(number: number))
}
print(scans)

You create and run a task for every sector. Build and run. Watch the indicators when you tap Engage systems:

This video has been sped up but, in your simulator, there’s always exactly one scheduled task, and the app processes about one task per second. This is because the loop is awaiting the completion of each task before starting the next, so the tasks run serially, not concurrently. When the full scan finishes, the duration is just over 20 seconds.

Concurrent tasks

You know these tasks can run concurrently, so here’s how you make that happen:

Select all the code you added to runAllTasks() and replace it:

await withTaskGroup(of: String.self) { [unowned self] group in

}

You’re creating and running a task group where each task returns a String. Inside the closure, you’ll call worker(number:), so you capture self as an unowned reference.

Inside the task group closure, loop over the sector numbers:

await withTaskGroup(of: String.self) { [unowned self] group in
🟩
  for number in 0..<total {

  }
🟥
}

Then create and run each task:

await withTaskGroup(of: String.self) { [unowned self] group in
  for number in 0..<total {
🟩
    group.addTask {
      await self.worker(number: number)
    }
🟥
  }
}

You add the task for this sector, then move right along to the next loop iteration.

This time, to see the effects of TaskGroup, build and run on a device: It has more CPUs, so more threads, than the simulator. Connect your device and take care of any security alerts. In the target editor, Signing & Capabilities, customize the bundle ID, check Automatically manage signing and set a Team. Select your device, then build and run.

The first time, you might have to “trust your app” after it installs: go to Settings > General > VPN & Device Management > Developer App/Apple Development: … > Trust

Now you can run the app. Tap Engage systems. Scheduled shoots up to 20 and, when the tasks finish, the duration is some fraction of 20. On my iPhone, it’s 4 seconds, which means the app used 5 threads.

But you didn’t see many UI updates until all the scan tasks finished, which isn’t very useful. You’ll fix this now.

Updating the UI

Here’s the problem: By default, a task inherits its parent’s priority value, so the scan tasks and their UI updating subtasks all have the same priority, and the UI updates go into the same queue that already contains all the scan tasks. So no UI updates can appear until all the scan tasks have finished.

One solution is to reduce the priority of the scan tasks.

Get back to ScanTask.swift. In run(), set the priority of the task:

await Task(priority: .medium)

Setting a lower priority for a new ScanTask suggests the scheduler should favor resuming a completed scan over starting a new one. UI updates are fast, so this won’t slow down the total scan much.

So instead of getting stuck behind all the scan tasks, UI update tasks move to the front of the queue.

Build and run on a simulator: This will probably run on only one thread, so updates will appear slowly enough to watch what’s happening.

That’s better! Now, the other two indicators update regularly, while the scan tasks run.

In the next episode, you’ll get and process results from the scan tasks.