Узнайте, как работать с объектом Task для безопасного выполнения асинхронных операций с использованием новых API параллелизма в Swift.

Внедрение структурированного параллелизма в Swift

В предыдущем руководстве автор рассказывал о новой функции async/await в Swift, а также о потокобезопасном параллелизме с использованием акторов, теперь пришло время приступить к другой важной функции параллелизма в Swift, называемой структурированным параллелизмом.

Что такое структурированный параллелизм (structured concurrency)? Если коротко, то это новый механизм, основанный на задачах, который позволяет разработчикам выполнять отдельные элементы задач параллельно. Обычно, когда вы ожидаете какой-то фрагмент кода, вы создаете потенциальную точку приостановки. Если мы возьмем наш пример с вычислением числа из статьи async/await, мы могли бы написать что-то вроде этого:

let x = await calculateFirstNumber()
let y = await calculateSecondNumber()
let z = await calculateThirdNumber()
print(x + y + z)

Я уже упоминал, что каждая строка выполняется после того, как предыдущая строка завершает свою работу. Мы создаем три потенциальные точки приостановки и ждем, пока процессор не получит достаточно мощности для выполнения и завершения каждой задачи. Все это происходит в последовательном порядке, но иногда это не то поведение, которое вы хотите.

Если вычисление зависит от результата предыдущего, этот пример идеален, так как вы можете использовать x для вычисления y, или x & y для вычисления z. А что если мы хотим выполнить эти задачи параллельно, и нам не важны отдельные результаты, но нам нужны все (x,y,z) так быстро, как мы можем?

async let x = calculateFirstNumber()
async let y = calculateSecondNumber()
async let z = calculateThirdNumber()

let res = await x + y + z
print(res)

Я уже показал вам, как это сделать с помощью предложения по привязке async let, которое является своего рода высокоуровневым слоем абстракции поверх структурированного параллелизма. Он до смешного упрощает параллельный запуск асинхронных задач. Итак, большая разница здесь в том, что мы можем выполнять все вычисления одновременно и ожидать "группу" результатов, содержащую x, y и z.

Опять же, в первом примере порядок выполнения следующий:

• Ожидать x, когда он будет готов, двигаемся дальше.

• Ожидать y, когда он будет готов, мы двигаемся вперед.

• Ожидать z, когда он будет готов, двигаемся вперед.

• Суммировать уже вычисленные числа x, y, z и вывести результат.

Второй пример можно описать следующим образом

• Создать асинхронный элемент задачи для вычисления x.

• Создать асинхронный элемент задачи для вычисления y.

• Создать асинхронный элемент задачи для вычисления z.

• Группировать элементы задач x, y, z вместе и ждать суммирования результатов, когда они будут готовы.

• Вывести окончательный результат.

Как видите, на этот раз нам не нужно ждать, пока предыдущий элемент задачи будет готов, но мы можем выполнять их все параллельно, а не в обычном последовательном порядке. У нас все еще есть 3 потенциальные точки приостановки, но порядок выполнения - это то, что действительно имеет значение. Благодаря параллельному выполнению задач вторая версия нашего кода может быть намного быстрее, поскольку CPU может выполнять все задачи одновременно (если у него есть свободный рабочий поток/исполнитель).

На самом базовом уровне это и есть структурированный параллелизм. Конечно, привязка async let скрывает большинство деталей реализации в этом случае, поэтому давайте немного спустимся в кроличью нору и отрефакторим наш код, используя задачи и группы задач.

await withTaskGroup(of: Int.self) { group in
    group.async {
        await calculateFirstNumber()
    }
    group.async {
        await calculateSecondNumber()
    }
    group.async {
        await calculateThirdNumber()
    }

    var sum: Int = 0
    for await res in group {
        sum += res
    }
    print(sum)
}

Мы можем использовать задачи в качестве базовых единиц для выполнения какой-либо работы. Задача может находиться в одном из трех состояний: приостановлена, выполняется или завершена. Задачи также поддерживают отмену и могут иметь связанный с ними приоритет.

Задачи могут образовывать иерархию, определяя дочерние задачи (child tasks). В настоящее время мы можем создавать группы задач и определять дочерние элементы с помощью функции group.async для параллельного выполнения, этот процесс создания дочерних задач может быть упрощен с помощью привязки async let. Дочерние задачи автоматически наследуют атрибуты родительской задачи, такие как приоритет, локальное хранилище задач, сроки выполнения, и они будут автоматически отменены, если родительская задача будет отменена. Поддержка дедлайнов появится в одном из последующих релизов Swift, поэтому я не буду говорить о них подробнее.

Период выполнения задачи называется job, каждое задание выполняется на исполнителе (executor). Исполнитель - это служба, которая может принимать задания и распределять их (по приоритету) для выполнения на доступном потоке. В настоящее время исполнители предоставляются системой, но позже участники смогут определять собственные исполнители.

Достаточно теории, как вы видите, можно определить группу заданий с помощью методов withTaskGroup или withThrowingTaskGroup. Единственное различие в том, что последний является вариантом c исключением, поэтому вы можете попытаться ожидать завершения работы async-функций.

Группа задач требует тип ChildTaskResult в качестве первого параметра, который должен быть типом Sendable. В нашем случае тип Int является идеальным кандидатом, так как мы собираемся собирать результаты с помощью группы. Вы можете добавить в группу элементы асинхронных задач, которые возвращаются с соответствующим типом результата.

Мы можем собирать отдельные результаты из группы, ожидая следующего элемента (await group.next()), но поскольку группа соответствует протоколу AsyncSequence, мы можем перебирать результаты, ожидая их с помощью стандартного цикла for.

Вот как работает структурированный параллелизм в двух словах. Самое лучшее во всей этой модели то, что благодаря использованию иерархий задач ни одна дочерняя задача не сможет случайно просочиться и продолжать работать в фоновом режиме. Это основная причина для таких API, что они всегда должны ожидать до окончания области видимости. (Спасибо за предложения @ktosopl).

Позвольте мне показать вам еще несколько примеров...

Ожидание зависимостей

Если у вас есть асинхронная зависимость для элементов задачи, вы можете либо вычислить результат заранее, до определения группы задач, либо внутри групповой операции вызвать несколько вещей.

import Foundation

func calculateFirstNumber() async -> Int {
    await withCheckedContinuation { c in
        DispatchQueue.main.asyncAfter(deadline: .now() + 2) {
            c.resume(with: .success(42))
        }
    }
}

func calculateSecondNumber() async -> Int {
    await withCheckedContinuation { c in
        DispatchQueue.main.asyncAfter(deadline: .now() + 1) {
            c.resume(with: .success(6))
        }
    }
}

func calculateThirdNumber(_ input: Int) async -> Int {
    await withCheckedContinuation { c in
        DispatchQueue.main.asyncAfter(deadline: .now() + 4) {
            c.resume(with: .success(9 + input))
        }
    }
}

func calculateFourthNumber(_ input: Int) async -> Int {
    await withCheckedContinuation { c in
        DispatchQueue.main.asyncAfter(deadline: .now() + 3) {
            c.resume(with: .success(69 + input))
        }
    }
}

@main
struct MyProgram {
    
    static func main() async {

        let x = await calculateFirstNumber()
        await withTaskGroup(of: Int.self) { group in
            group.async {
                await calculateThirdNumber(x)
            }
            group.async {
                let y = await calculateSecondNumber()
                return await calculateFourthNumber(y)
            }
            

            var result: Int = 0
            for await res in group {
                result += res
            }
            print(result)
        }
    }
}

Стоит отметить, что если вы хотите поддерживать правильную логику отмены, вам следует быть осторожным с точками приостановки. В этот раз я не буду вдаваться в подробности отмены, но когда-нибудь я напишу отдельную статью на эту тему (я все еще учусь этому...).

Задачи с разными типами результатов

Если элементы вашей задачи имеют различные типы возврата, вы можете легко создать новое перечисление со связанными значениями и использовать его в качестве общего типа при определении группы задач. Вы можете использовать перечисление и вставлять базовые значения при возврате с помощью функций асинхронных элементов задач.

import Foundation

func calculateNumber() async -> Int {
    await withCheckedContinuation { c in
        DispatchQueue.main.asyncAfter(deadline: .now() + 4) {
            c.resume(with: .success(42))
        }
    }
}

func calculateString() async -> String {
    await withCheckedContinuation { c in
        DispatchQueue.main.asyncAfter(deadline: .now() + 2) {
            c.resume(with: .success("The meaning of life is: "))
        }
    }
}

@main
struct MyProgram {
    
    static func main() async {
        
        enum TaskSteps {
            case first(Int)
            case second(String)
        }

        await withTaskGroup(of: TaskSteps.self) { group in
            group.async {
                .first(await calculateNumber())
            }
            group.async {
                .second(await calculateString())
            }

            var result: String = ""
            for await res in group {
                switch res {
                case .first(let value):
                    result = result + String(value)
                case .second(let value):
                    result = value + result
                }
            }
            print(result)
        }
    }
}

После завершения задач вы можете поменять местами элементы последовательности и выполнить окончательную операцию над результатом на основе обернутого значения перечисления. Этот маленький трюк позволит вам запускать все виды задач с различными типами возврата для параллельной работы с использованием нового API Tasks.

Неструктурированные и отделенные (detached) задачи

Как вы уже могли заметить, невозможно вызвать async API из функции синхронизации. Именно здесь могут помочь неструктурированные задачи. Самое важное, что нужно отметить, это то, что время жизни неструктурированной задачи не привязано к создающей задаче. Они могут пережить родителя, и они наследуют приоритеты, локальные значения задачи, сроки от родителя. Неструктурированные задачи представляются хэндлом задачи, который можно использовать для отмены задачи.

import Foundation

func calculateFirstNumber() async -> Int {
    await withCheckedContinuation { c in
        DispatchQueue.main.asyncAfter(deadline: .now() + 3) {
            c.resume(with: .success(42))
        }
    }
}

@main
struct MyProgram {
    
    static func main() {
        Task(priority: .background) {
            let handle = Task { () -> Int in
                print(Task.currentPriority == .background)
                return await calculateFirstNumber()
            }
            
            let x = await handle.get()
            print("The meaning of life is:", x)
            exit(EXIT_SUCCESS)
        }
        dispatchMain()
    }
}

Вы можете получить текущий приоритет задачи с помощью статического свойства currentPriority и проверить, совпадает ли он с приоритетом родительской задачи (конечно, он должен совпадать).

Так в чем же разница между неструктурированными задачами и отделенными задачами? Ну, ответ довольно прост: неструктурированная задача наследует родительский контекст, с другой стороны, отделенные задачи не наследуют ничего от своего родительского контекста (приоритеты, задания-локалы, сроки).

@main
struct MyProgram {
    
    static func main() {
        Task(priority: .background) {
            Task.detached {
                /// false -> Task.currentPriority is unspecified
                print(Task.currentPriority == .background)
                let x = await calculateFirstNumber()
                print("The meaning of life is:", x)
                exit(EXIT_SUCCESS)
            }
        }
        dispatchMain()
    }
}

Вы можете создать отделенную задачу с помощью метода detached, как вы можете видеть, приоритет текущей задачи внутри отделенной задачи не определен, что определенно не равно приоритету родительской задачи. Кстати, получить текущую задачу можно также с помощью функции withUnsafeCurrentTask. Вы можете использовать этот метод также для получения приоритета или проверки, не отменена ли задача.

@main
struct MyProgram {
    
    static func main() {
        Task(priority: .background) {
            Task.detached {
                withUnsafeCurrentTask { task in
                    print(task?.isCancelled ?? false)
                    print(task?.priority == .unspecified)
                }
                let x = await calculateFirstNumber()
                print("The meaning of life is:", x)
                exit(EXIT_SUCCESS)
            }
        }
        dispatchMain()
    }
}

Существует еще одно большое различие между отделенными (detached) и неструктурированными (unstructured) задачами. Если вы создадите неструктурированную задачу на основе актора, задача будет выполняться непосредственно на этом акторе и НЕ параллельно, а отсоединенная задача будет выполняться сразу же параллельно. Это означает, что неструктурированная задача может изменять внутреннее состояние агента, а отсоединенная задача не может изменять внутреннее состояние агента.

Вы также можете использовать преимущества неструктурированных задач в группах задач для создания более сложных структур задач, если структурированная иерархия не соответствует вашим потребностям.

Локальные значения задачи (task local values)

Есть еще одна вещь, которую я хотел бы вам показать. Мы уже много раз упоминали локальные значения задач (task local values), поэтому вот краткий раздел о них. Эта функция, по сути, является улучшенной версией потокового локального хранилища, созданного для того, чтобы хорошо сочетаться со структурированным параллелизмом в Swift.

Иногда вы хотите передавать пользовательские контекстные данные вместе с задачами, и здесь на помощь приходят локальные значения задач. Например, вы можете добавить отладочную информацию в объекты задачи и использовать ее для более легкого поиска проблем. Донни Уолс написал подробную статью о локальных значениях задач, если вам интересно узнать больше об этой функции, вам обязательно нужно прочитать его статью. ????

Итак, на практике вы можете аннотировать статическое свойство с помощью обертки свойства @TaskLocal, а затем вы можете читать эти метаданные внутри другой задачи. С этого момента вы можете изменять это свойство только с помощью функции withValue на самой обертке.

import Foundation

enum TaskStorage {
    @TaskLocal static var name: String?
}

@main
struct MyProgram {
    
    static func main() async {
        await TaskStorage.$name.withValue("my-task") {
            let t1 = Task {
                print("unstructured:", TaskStorage.name ?? "n/a")
            }
            
            let t2 = Task.detached {
                print("detached:", TaskStorage.name ?? "n/a")
            }
            /// runs in parallel
            _ = await [t1.value, t2.value]
        }
    }
}

Задачи наследуют эти локальные значения (кроме detached), и вы можете изменять значения локальных значений задач внутри данной задачи, но эти изменения будут видны только для текущей задачи и дочерних задач. Подводя итог, можно сказать, что локальные значения задач всегда привязаны к данной области видимости задачи.

Как вы видите, структурированный параллелизм в Swift - это довольно сложная задача, но как только вы поймете основы, все встанет на свои места, а новые функции async/await и Tasks позволят вам легко создавать задания для последовательного или параллельного выполнения. В любом случае, я надеюсь, что вам понравилась эта статья.

Если вам понравилась статья, то подписывайтесь, тут я делюсь историями о разработке.