Многопоточность и Kotlin в Яндекс.Картах: как не допустить падения новых фич на iOS

Привет! Меня зовут Женя Васильев, я делаю Яндекс.Карты под Android. А с появлением у нас Kotlin Multiplatform — так уж получилось, ещё и под iOS.

Kotlin Multiplatform позволяет писать код, который будет одинаково работать на iOS и Android. По крайней мере, должен одинаково работать. И в случае с простыми фичами правда работает. Но если вы, как и я, впервые реализовываете в мультиплатформе сложную фичу с многопоточностью, на iOS вас будут ждать креши в рантайме и утечки.

В статье я расскажу и покажу на примерах, с какими проблемами я столкнулся при написании многопоточного кода на Kotlin Multiplatform, как эти проблемы решать, как лучше организовывать потоки данных в многопоточной среде и что ещё нужно делать, чтобы написанное на Kotlin не падало на iOS. Увы, писать код «как обычно» не получится.

Два года назад, чтобы уменьшить дублирование кода между iOS- и Android-приложениями Карт, мы начали выносить в мультиплатформу относительно простую однопоточную бизнес-логику. И это уже было хорошо: меньше дублирования — меньше багов и быстрее разработка, в итоге пользователи быстрее получают более качественный продукт. Потом общей логики стало больше, и мы начали экспериментировать с многопоточностью. Был большой оверхэд на переключение тредов, но для разовых асинхронных операций это было не страшно. Потом появились корутины native-mt с фоновыми потоками — быстрое и официальное решение, позволяющее легко реализовать в мультиплатформе логику любой сложности! Так подумали мы, и я взялся за разработку новой версии поискового слоя.

Поисковый слой отображает результаты поиска в виде пинов на карте. Слой вычисляет, как нужно показать пин каждого объекта, исходя из его рейтинга, плотности поисковой выдачи, уровня зума, расстояния до соседей и так далее. Плавность интерфейса без многопоточности тут недостижима.

В итоге всё получилось. Ошиблись мы только в одном: что будет легко.

Прежде чем начать

Чтобы быть на одной волне, читателю может понадобиться:

• Документация по Kotlin Multiplatform
• Плагин к Android Studio для создания приложения-песочницы с мультиплатформой
• Документация по механизму expect-actual
• Документация по корутинам
• Корутины версии native-mt: для примеров я использовал 1.5.0-native-mt

Корень всех бед

На Android мультиплатформенный код исполняется на Kotlin/JVM — так же, как если бы это был код обычного Android-приложения на Kotlin.

На iOS тот же код компилируется в машинный код, который исполняется без виртуальной машины. Эта технология называется Kotlin/Native. И модель памяти Kotlin/Native отличается от Java Memory Model.

Напишем простой пример:

fun yolo() {
    MainScope().launch {
        var a = 1
        withContext(Dispatchers.Default) {
            a = 2 // InvalidMutabilityException (iOS)
        }
        a = 3
    }
}

На Android всё работает. На iOS при вызове yolo() произойдёт креш. Потому что мы не учли правила Kotlin/Native:

1. Мутабельное состояние, чтобы оставаться таковым, должно принадлежать одному потоку. В большинстве случаев это значит, что мутабельные данные не должны пересекать границы потоков — даже при доступе только на чтение. Единственное исключение: передача данных между потоками с помощью DetachedObjectGraph. О нём поговорим ниже.
2. Состояние может быть доступно из разных потоков, но для этого оно должно быть заморожено.

Заморозка — рантайм-свойство Kotlin/Native. Заморозить любой объект можно с помощью экстеншена .freeze(). Это односторонняя операция: разморозить объект нельзя. Замороженные объекты не должны изменяться, при попытке их мутирования произойдёт InvalidMutabilityException. Заморозка применяется ко всему графу объектов: замораживается всё, к чему можно попасть по ссылкам из исходного объекта.

При переключении потоков с помощью библиотеки корутин заморозка происходит автоматически. Если же возникает необходимость ручной заморозки объектов, в мультиплатформе придётся реализовать expect-actual для вызова .freeze() или подключить библиотеку stately-common.

Как с этим жить

Рассмотрим чуть более реалистичный пример:

MainScope().launch {
    var counter: Int = countSomething()

    withContext(Dispatchers.Default) {
        println(counter)
    }

    counter += countMoar() // InvalidMutabilityException (iOS)
    renderUi(counter)
}

InvalidMutabilityException будет брошен при попытке мутирования замороженной переменной counter. А заморозится переменная автоматически при переключении потоков (диспатчеров), поскольку будет захвачена лямбдой. Нужно любым способом избавиться от мутирования замороженных данных. Например, так:

MainScope().launch {
    val frozenCounter: Int = countSomething()

    withContext(Dispatchers.Default) {
        println(frozenCounter)
    }

    val counter: Int = frozenCounter + countMoar()
    renderUi(counter)
}

Даже в таком простом случае мы вынуждены пересмотреть поток данных и учесть в коде, что переменные, которые пересекли границы тредов, станут иммутабельными, даже если они объявлены как var.

Как отморозить целый класс

Итак, нам нужен кеш и фоновые вычисления. Кажется, мы к этому готовы:

class PinStateProcessor {
    private val cache: MutableMap<PinId, PinState> = mutableMapOf()
    private val defaultPinState = PinState.DefaultState

    suspend fun calculateState(pinId: PinId, ctx: SearchContext): PinState {
        cache[pinId]?.let { cachedState ->
            return cachedState
        }

        val pinState = withContext(Dispatchers.Default) {
            calculateStateInternal(pinId, ctx, defaultPinState)
        }

        cache[pinId] = pinState
        return pinState
    }

    private fun calculateStateInternal(pinId: PinId, ctx: SearchContext, defaultState: PinState): PinState {
        // ... Some math ...
        return PinState.SomeState
    }
}

Создаём и работаем с PinStateProcessor на главном потоке, calculateState() зовём из Main-диспатчера. Внутри в случае необходимости корутина на главном потоке приостанавливается, на фоновом потоке происходят вычисления, затем на главном потоке обновляется кеш (для простоты опустим синхронизацию). Таким образом кеш не пересекает границы тредов и остаётся мутабельным. А заморозка константы defaultPinState, попадающей на Default-диспатчер, нас не волнует. Всё в порядке. Или нет?

Здесь есть две похожие ошибки, каждая из которых приводит к InvalidMutabilityException при вызове calculateState() на iOS. Чтобы стало очевидно, перепишу небольшой кусочек:

val pinState = withContext(Dispatchers.Default) {
    this@PinStateProcessor.calculateStateInternal(
        pinId,
        ctx,
        this@PinStateProcessor.defaultPinState
    )
}

Поля и функции-члены класса доступны только в контексте этого класса. Поэтому при работе с defaultPinState или при вызове calculateStateInternal() на фоновом потоке лямбдой будет захвачен весь инстанс класса. А заморозка класса приведёт к заморозке всех его полей, включая кеш, который мы впоследствии пытаемся изменить.

Константу здесь можно заинлайнить, вынести в статику или переложить в локальную переменную перед переключением потоков. А вот чистую функцию отвязать от класса можно только одним способом: сделать её статической.

Опять ноге досталось

С детскими болезнями справились, теперь аккуратно добавим логирование:

suspend fun calculateState(pinId: PinId, ctx: SearchContext): PinState {
    cache[pinId]?.let { return it }

    val pinState = withContext(Dispatchers.Default) {
        calculateStateInternalStatic(pinId, ctx, PinState.DefaultState)
    }

    cache[pinId] = pinState

    logPinStates(cache.values)

    return pinState
}

Реализация где-то в отдельном модуле:

fun logPinStates(states: Collection<PinState>) {
    println(states.toString())
}

После этого уйдём в отпуск.

Вернувшись из отпуска, заглянем в модуль с логированием. Коллекция тут read-only, так что вынесение логирования в фон, кажется, не помешает:

suspend fun logPinStates(states: Collection<PinState>) {
    withContext(Dispatchers.Default) {
        println(states.toString())
    }
}

При первом вызове calculateState() на iOS всё хорошо. А при втором происходит креш. Происходит он потому, что cache.values возвращает вьюху к исходной мутабельной мапе. А read-only интерфейс, как и в «обычном» мире, не утверждает, что за ним скрываются иммутабельные данные. В итоге на фоновый поток передаётся весь кеш, и при следующей попытке его мутирования всё закономерно падает.

Получаем два грустных правила:

• Не уверен — не замораживай. Если вы получаете read-only-объект извне, замораживать его можно, только если он на самом деле иммутабелен. При этом хорошо бы в названии функции отразить факт переключения потоков или заморозки.
• Не уверен — не мутируй. Если интерфейс объекта допускает его изменение, это не значит, что объект не заморожен. Заморожен может быть любой объект, но синтаксис этого не подскажет.

Местная анестезия

InvalidMutabilityException выкидывается при попытке мутирования замороженного объекта. В последнем примере это выглядит так: InvalidMutabilityException: mutation attempt of frozen kotlin.collections.HashMap. Место изменения кеша найти несложно, но его мутирование — не ошибка. Ошибка в том, что кеш оказался заморожен. А вот найти место заморозки в большом количестве разветвлённого кода может быть сложно.

Будет легче, если использовать экстеншен Any.ensureNeverFrozen(). Он доступен в Kotlin/Native; в мультиплатформе придётся реализовать expect-actual или воспользоваться библиотекой stately-common.

Глобальные правила игры не изменятся: по-прежнему нельзя допускать, чтобы объект, который должен оставаться мутабельным, замораживался. Если это произойдёт, по-прежнему будет креш в рантайме. Но это будет другой креш, в другом месте и в другое время: FreezingException. Брошен он будет при заморозке объекта, на котором был вызван .ensureNeverFrozen().

private val cache = mutableMapOf<PinId, PinState>().apply { ensureNeverFrozen() }

Теперь при первом же вызове calculateState() со «сломанным» логированием мы получим FreezingException с ведущим к logPinStates() стектрейсом.

Имеет смысл вызывать .ensureNeverFrozen() на всех объектах, которые, согласно архитектуре вашего приложения, действительно не должны замораживаться. Но, помня про то, что граф объектов замораживается целиком, .ensureNeverFrozen() достаточно вызывать лишь на ключевых объектах-листьях графа.

Также в Kotlin/Native есть экстеншен-проперти .isFrozen. Он тоже может помочь при отладке. Ещё с его помощью можно строить сложные условные потоки данных или производить оптимизации:

var list = listOf(MutableData(1), MutableData(2), MutableData(3))
blackBox(list)
if (list.isFrozen) {
    list = list.deepCopy()
}
list.forEach { it.counter++ }

Нарушители правил

Предположим, нам удобнее (или даже необходимо), чтобы с нашим классом и кешем можно было работать из любого потока.

К счастью, в Kotlin/Native есть костыли. Имя им — shareable-сущности. Вот некоторые из них: AtomicReference, AtomicInt, Deferred, Channel, Mutex, Flow. Они заморожены и ссылаются на замороженные данные. Но их содержимое может меняться!

С их помощью можно реализовать по сути мутабельный кеш в полностью замороженной среде:

class PinStateProcessor {
    private val cache = AtomicReference<Map<PinId, PinState>>(emptyMap())
    private val mutex = Mutex()

    init {
        freeze()
    }

    suspend fun calculateState(pinId: PinId, ctx: SearchContext): PinState {
        mutex.withLock {
            cache.value[pinId]?.let { return it }

            return withContext(Dispatchers.Default) {
                val pinState = calculateStateInternal(pinId, ctx, DefaultState)
                cache.value += pinId to pinState
                pinState
            }
        }
    }
}

Чтобы атомики проросли в мультиплатформу, можно использовать библиотеку stately-concurrency.

На всякий случай расшифрую: cache.value += pinId to pinState — эквивалент cache.set(cache.get().toMutableMap().apply { put(pinId, pinState) }). Вызывать руками freeze() необязательно: при первом же переключении диспатчеров всё заморозится автоматически. Но, как и в случае с добавлением ensureNeverFrozen(), так мы быстрее обнаружим ошибку, если случайно заморозим что-то лишнее.

Забавный факт: если поменять местами init-блок и объявление полей класса, мы получим InvalidMutabilityException при инстанциировании PinStateProcessor. Потому что вначале всё заморозится, и только после этого полям будут присваиваться указанные в коде значения. В Kotlin так принято.

Кроме того, у перечислений, глобальных переменных и объектов в Kotlin/Native есть особенности в плане заморозки и доступа с разных потоков. Не буду пересказывать документацию, почитайте.

Утекай

Нельзя написать хорошую фичу, особенно на новой технологии, не посидев с профайлером. Вот и я перед написанием реального кода занялся исследованиями. С производительностью в Kotlin/Native всё хорошо, накладные расходы на переключение потоков минимальны. Но проблема всё же нашлась.

suspend fun leak() {
    val leakedList = withContext(Dispatchers.Default) {
        val list = mutableListOf<Int>()
        repeat(1_000_000) {
            list += it
        }
        list
    }
    println(leakedList.size)
}

Вызываем эту функцию несколько раз и смотрим график потребления памяти на Android:

И на iOS:

Оказывается, объекты, пересекающие границы потоков Kotlin/Native, могут утекать. Оказывается, в Kotlin/Native есть сборщик мусора, но его нужно вызывать руками. Оказывается, и этого не было в документации, GC Kotlin/Native — тред-локальный: на каком потоке утекло, на том и надо его вызывать. Делать это нужно после выхода из Kotlin-скоупа, объекты которого хочется освободить, чтобы на стеке и в Continuation не оставалось мусора.

В мультиплатформе GC доступен через expect-actual:

// commonMain
expect object GC {
    fun collect()
}

// iosMain
actual typealias GC = kotlin.native.internal.GC

// androidMain
actual object GC {
    actual fun collect() = Unit
}

В нашем случае поможет, например, функция, вызываемая следом за функцией с утечками:

suspend fun gc() {
    GC.collect()
    withContext(Dispatchers.Default) {
        GC.collect()
    }
}

Проверяем:

Отлично.

Но когда в реальном мире нужно вызывать GC и нужно ли? Действуйте по ситуации и на свой вкус. Вот возможные варианты:

• Утечек может и не быть (да, вам тоже понадобится профайлер). В этом случае, разумеется, вызывать GC не нужно.
• Предположим, у вас утекает незначительный объём данных или из-за своей специфики ваше приложение в принципе долго не живёт. В этом случае можно оставить всё как есть и тоже не вызывать GC руками.
• В долгоживущем алгоритме имеет смысл вызывать сборщик по мере появления крупных кусков мусора, чтобы было где появляться новым.
• Можно звать GC при завершении работы соответствующей фичи. Например, я это делаю при закрытии экрана поиска.
• Если приложение живёт долго и в нём постоянно работает множество маленьких генераторов мусора, можно централизованно собирать всё по таймеру, раз в n минут.

Он вам не друг

DetachedObjectGraph — единственный способ передачи объектов между потоками без заморозки. И сейчас я постараюсь вас убедить, что он вам не нужен.

Для работы с DetachedObjectGraph в мультиплатформе придётся написать expect-actual на сам DetachedObjectGraph, TransferMode и экстеншен DetachedObjectGraph<T>.attach().

Вот простейший пример использования:

val dog = DetachedObjectGraph(TransferMode.SAFE) { mutableListOf<Any>() }
withContext(Dispatchers.Default) {
    val list = dog.attach()
    list += Unit
    println(list.size) // 1
}

Правила такие: лямбда в конструкторе DetachedObjectGraph должна возвращать передаваемый между потоками объект. Больше никаких ссылок на этот объект быть не должно. После переключения потоков метод attach() «прикрепляет» к текущему потоку передаваемый объект и возвращает его. Далее всё работает как обычно.

Пример не содержит ошибок, но бесполезен, потому что такой новый и пустой мутабельный список можно создать сразу на нужном потоке. Попробуем передать использованный объект:

var outerList: MutableList<Any>? = mutableListOf()
outerList?.add(Unit)

val dog = DetachedObjectGraph(TransferMode.SAFE) {
    outerList.also {
        outerList = null
    }
}

withContext(Dispatchers.Default) {
    val innerList = dog.attach()
    innerList?.add(Unit)
    println(innerList?.size) // 2
}

Переменную outerList нужно занулять, чтобы ссылка на список была только в конструкторе. В противном случае при создании DetachedObjectGraph вы получите IllegalStateException: Illegal transfer state. Но если запустить этот код в цикле достаточно много раз, вы всё равно получите IllegalStateException.

Вспоминаем про GC и дописываем:

val dog = DetachedObjectGraph(TransferMode.SAFE) {
    {
        outerList.also {
            outerList = null
        }
    }().also {
        GC.collect()
    }
}

Теперь всё правда работает.

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

Но у всего есть цена. И здесь эта цена — O(N). Платить вы будете при создании DetachedObjectGraph, потому что каждый раз происходит проверка, что никто кроме DetachedObjectGraph не держит ссылок ни на один из объектов передаваемого графа. Святослав Щербина из JetBrains в переписке со мной это прокомментировал, публикую комментарий с его согласия:

Там всё немного хуже. Ещё обрабатывается текущий буфер уменьшений счётчиков ссылок в GC, который тоже может быть большим. И он после этого не очищается, так что эта цена не амортизируется в O(N).

За проверку наличия лишних ссылок отвечает TransferMode. При TransferMode.SAFE проверка включена; повторюсь, чем больше объектов передаётся между потоками, тем она дороже. TransferMode.UNSAFE проверку отключает — не требование про количество ссылок, а только проверку при создании. Если, дочитав досюда, вы совсем преисполнились, и слово unsafe в контексте Kotlin/Native не вызывает у вас никаких эмоций, обратите внимание на KDoc:

/**
 * Skip reachibility check, can lead to mysterious crashes in an application.
 * USE UNSAFE MODE ONLY IF ABSOLUTELY SURE WHAT YOU'RE DOING!!!
 */

Более того, Святослав в переписке рассказал о важном нюансе: на самом деле и обработка буфера уменьшений счётчиков ссылок, и обход подграфа в UNSAFE-режиме тоже будут происходить, хоть и меньшее количество раз. Так что накладные расходы на передачу большого числа объектов останутся, поэтому прирост производительности надо замерять: он может оказаться незначительным.

Итак, я во всё это погрузился, и решил не использовать DetachedObjectGraph в проде. И вам не советую.

Выбираем меньшее зло

Теперь можно подытожить, когда и как лучше передавать данные между потоками в мультиплатформе.

• В простейшем случае локальные данные «туда» передаются через захват лямбдой withContext, async и так далее, а «обратно» — через возвращаемое значение withContext или Deferred.
• Для хранения долгоживущих объектов в полях класса подойдут атомики, с которыми можно работать из любого потока.
• Если мутабельные состояния или коллекции не слишком велики, их можно тоже хранить в атомиках в замороженном виде, а «мутирование» осуществлять через копирование с изменением.
• Если коллекции могут сильно разрастаться, или если это просто хорошо согласуется с потоком данных в вашей архитектуре, работу с коллекциями можно ограничить одним тредом, тем самым сохранив их мутабельность, а для фоновых вычислений использовать первый способ и необходимые DTO'шки.
• Если вы зачем-то хотите сохранить мутабельность передаваемых между потоками данных, можно использовать DetachedObjectGraph на свой страх и риск.

Бонус: уменьшаем лаги без многопоточности

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

Такой код вешает главный поток на несколько секунд:

var counter = 0L
repeat(1_000_000_000) {
    counter++
}

А такой выполняет столько же работы и не тормозит:

var counter = 0L
repeat(1_000) {
    repeat(1_000_000) {
        counter++
    }
    yield()
}

Эпилог

Я люблю сидеть с напильником и микроскопом. И полученный опыт, и результат мне понравились. Новый поисковый слой сейчас постепенно раскатывается, скоро он будет доступен у всех пользователей. И на приборах мы не видим ни одного InvalidMutabilityException, хотя получить его чертовски просто.

Разработчики Kotlin/Native из JetBrains давно работают над созданием новой модели памяти. Совсем недавно было опубликовано превью для разработчиков, уже можно попробовать жизнь без заморозки. Когда-нибудь новая модель памяти зарелизится, надеюсь, без критичных багов. И тогда можно будет забыть про InvalidMutabilityException. Но этот день ещё не настал.

Если вы уже используете Kotlin Multiplatform, надеюсь, мой опыт будет вам полезен. Если же вы только поглядываете на мультиплатформу, рекомендую её попробовать хотя бы для простых фич. А как распробуете, глядишь, и новая модель памяти подъедет.

В качестве заключения, три совета на разных уровнях абстракции:

• Используйте ensureNeverFrozen(): это очень упрощает отладку.
• Осознайте и реорганизуйте свои потоки данных с учётом заморозки: писать всё как обычно не получится.
• Чаще запускайте код на iOS: именно там проявляются все проблемы из-за специфики Kotlin/Native.