Корутины Kotlin в продакшене: уроки и подводные камни

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

Скрытые издержки переключения контекста

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

suspend fun processUserData(userId: String): UserData {
    return withContext(Dispatchers.IO) {
        val basicInfo = userService.fetchBasicInfo(userId)
        withContext(Dispatchers.Default) {
            // process data
            basicInfo.processHeavyComputation()
        }
    }
}

suspend fun processUserData(userId: String): UserData {

    return withContext(Dispatchers.IO) {

        val basicInfo = userService.fetchBasicInfo(userId)

        withContext(Dispatchers.Default) {

            // process data

            basicInfo.processHeavyComputation()

        }

    }

}

Каждый переключатель withContext создает накладные расходы. В сценариях с высокой пропускной способностью мы обнаружили, что этот паттерн приводит к снижению производительности до 30%. Вместо этого структурируйте свой код так, чтобы минимизировать переключения контекста:

suspend fun processUserData(userId: String): UserData {
    // IO operations grouped together
    val basicInfo = withContext(Dispatchers.IO) {
        userService.fetchBasicInfo(userId)
    }

    // CPU-bound operations grouped together
    return withContext(Dispatchers.Default) {
        basicInfo.processHeavyComputation()
    }
}

xxxxxxxxxx

suspend fun processUserData(userId: String): UserData {

    // IO operations grouped together

    val basicInfo = withContext(Dispatchers.IO) {

        userService.fetchBasicInfo(userId)

    }

​

    // CPU-bound operations grouped together

    return withContext(Dispatchers.Default) {

        basicInfo.processHeavyComputation()

    }

}

Перенасыщение пула потоков: тихий убийца производительности

Произошел инцидент, когда мы неосознанно перенасытили наши пулы потоков. Первопричина? Этот обманчивый шаблон:

// DON'T DO THIS
suspend fun processOrders(orders: List<Order>) {
    orders.forEach { order ->
        launch(Dispatchers.IO) {
            processOrder(order)
        }
    }
}

xxxxxxxxxx

// DON'T DO THIS

suspend fun processOrders(orders: List<Order>) {

    orders.forEach { order ->

        launch(Dispatchers.IO) {

            processOrder(order)

        }

    }

}

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

suspend fun processOrders(orders: List<Order>) {
    coroutineScope {
        orders.chunked(100).forEach { chunk ->
            chunk.map { order ->
                async(Dispatchers.IO) {
                    processOrder(order)
                }
            }.awaitAll()
        }
    }
}

xxxxxxxxxx

suspend fun processOrders(orders: List<Order>) {

    coroutineScope {

        orders.chunked(100).forEach { chunk ->

            chunk.map { order ->

                async(Dispatchers.IO) {

                    processOrder(order)

                }

            }.awaitAll()

        }

    }

}

Утечки памяти через отношения «родитель-ребенок

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

class OrderProcessor {
    private val scope = CoroutineScope(Dispatchers.Default)

    fun processOrderAsync(order: Order) {
        scope.launch {
            // long-running operation
            processOrder(order)
        }
    }
}

xxxxxxxxxx

class OrderProcessor {

    private val scope = CoroutineScope(Dispatchers.Default)

​

    fun processOrderAsync(order: Order) {

        scope.launch {

            // long-running operation

            processOrder(order)

        }

    }

}

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

class OrderProcessor : CoroutineScope {
    private val job = SupervisorJob()
    override val coroutineContext = Dispatchers.Default + job

    fun destroy() {
        job.cancel() // cancels all child coroutines
    }

    fun processOrderAsync(order: Order) {
        launch {
            try {
                processOrder(order)
            } catch (e: CancellationException) {
                // handle cancellation gracefully
                order.markCancelled()
            }
        }
    }
}

xxxxxxxxxx

class OrderProcessor : CoroutineScope {

    private val job = SupervisorJob()

    override val coroutineContext = Dispatchers.Default + job

​

    fun destroy() {

        job.cancel() // cancels all child coroutines

    }

​

    fun processOrderAsync(order: Order) {

        launch {

            try {

                processOrder(order)

            } catch (e: CancellationException) {

                // handle cancellation gracefully

                order.markCancelled()

            }

        }

    }

}

Кошмар отладки и способы решения

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

Когда мы только начали отлаживать проблемы с корутинами, мы обратили внимание на Threads Monitor. Для тех, кто работает с традиционными Java-приложениями, монитор может показаться пугающим. Здесь много потоков с таймером ожидания (DefaultDispatcher-worker-x).

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

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

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

Корутины Kotlinx сами по себе не предоставляют MBean, однако мы можем реализовать простой MBean, который предоставит нам несколько метрик:

• Количество активных корутинов
• Количество приостановленных корутин
• Какие корутины существуют и какой диспетчер они используют

Имейте в виду, что этот MBean следует использовать только для отладки, код, показанный ниже, также сопровождается снижением производительности, о чем можно прочитать в документации.

Сначала нам нужен интерфейс, который сообщит JMX, какие функции будут открыты:

import javax.management.MXBean

@MXBean
interface CoroutineMonitorMBean {
    fun getActiveCoroutines(): Int
    fun getSuspendedCoroutines(): Int
    fun getCoroutineStackTraces(): List<String>
}

xxxxxxxxxx

import javax.management.MXBean

​

@MXBean

interface CoroutineMonitorMBean {

    fun getActiveCoroutines(): Int

    fun getSuspendedCoroutines(): Int

    fun getCoroutineStackTraces(): List<String>

}

Реализация MBean требует от нас добавления дополнительного пакета: org.jetbrains.kotlinx:kotlinx-coroutines-debug. После его добавления мы получим доступ к классу DebugProbes, который позволит нам дампить информацию о существующих корутинах.

import kotlinx.coroutines.*
import kotlinx.coroutines.debug.DebugProbes
import kotlinx.coroutines.debug.State

@OptIn(ExperimentalCoroutinesApi::class)
class CoroutineMonitor : CoroutineMonitorMBean {
    init {
        // enables the debug probes and enables additional logging of 
        // coroutine contexts
        DebugProbes.install()
    }

    override fun getActiveCoroutines(): Int {
        return DebugProbes.dumpCoroutinesInfo().count { it.state == State.RUNNING }
    }

    override fun getSuspendedCoroutines(): Int {
        return DebugProbes.dumpCoroutinesInfo().count { it.state == State.SUSPENDED }
    }

    override fun getCoroutineStackTraces(): List<String> {
        return DebugProbes.dumpCoroutinesInfo().map { it.toString() }
    }
}

xxxxxxxxxx

import kotlinx.coroutines.*

import kotlinx.coroutines.debug.DebugProbes

import kotlinx.coroutines.debug.State

​

@OptIn(ExperimentalCoroutinesApi::class)

class CoroutineMonitor : CoroutineMonitorMBean {

    init {

        // enables the debug probes and enables additional logging of 

        // coroutine contexts

        DebugProbes.install()

    }

​

    override fun getActiveCoroutines(): Int {

        return DebugProbes.dumpCoroutinesInfo().count { it.state == State.RUNNING }

    }

​

    override fun getSuspendedCoroutines(): Int {

        return DebugProbes.dumpCoroutinesInfo().count { it.state == State.SUSPENDED }

    }

​

    override fun getCoroutineStackTraces(): List<String> {

        return DebugProbes.dumpCoroutinesInfo().map { it.toString() }

    }

}

После этого вы можете зарегистрировать MBean в своем приложении следующим образом:

fun main() {
    val mBeanServer = ManagementFactory.getPlatformMBeanServer()
    val monitorMBean = CoroutineMonitor()
    val beanName = ObjectName("com.enapi:type=CoroutineMonitor")
    mBeanServer.registerMBean(monitorMBean, beanName)
}

xxxxxxxxxx

fun main() {

    val mBeanServer = ManagementFactory.getPlatformMBeanServer()

    val monitorMBean = CoroutineMonitor()

    val beanName = ObjectName("com.enapi:type=CoroutineMonitor")

    mBeanServer.registerMBean(monitorMBean, beanName)

}

В VisualVM подключитесь к своему приложению, перейдите на вкладку MBeans и найдите com.enapi.CoroutineMonitor, который должен выглядеть следующим образом:

Есть и третий метод, который менее удобен для работы, но может помочь вам найти проблемы в вашем приложении. Это Java-агент, который можно загрузить и использовать следующим образом: java -jar your-application.jar -javaagent:kotlinx-coroutines-debug.jar. Это позволит включить отладочные зонды, как мы это делали в нашем MBean. Включение отладочных зондов позволяет вести дополнительное логирование. Кроме того, в Linux и MacOS вы можете kill -5 $pid, чтобы заставить ваше приложение вывести все активные корутины. Однако это не то, что вы можете сделать в производственной среде, и мы сами этим не пользовались.

Когда корутины — неправильный выбор

На собственном горьком опыте мы выявили два основных сценария, в которых корутины больше вредят, чем помогают:

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

val computationPool = Executors.newFixedThreadPool(
    Runtime.getRuntime().availableProcessors()
)

// better for CPU-intensive work
fun processImage(image: Image): ProcessedImage =
    computationPool.submit(Callable {
        image.applyHeavyFilters()
    }).get()

xxxxxxxxxx

val computationPool = Executors.newFixedThreadPool(

    Runtime.getRuntime().availableProcessors()

)

​

// better for CPU-intensive work

fun processImage(image: Image): ProcessedImage =

    computationPool.submit(Callable {

        image.applyHeavyFilters()

    }).get()

Простые последовательные операции: не все должно быть асинхронным. Мы обнаружили, что чрезмерно усложняем простые операции:

// unnecessarily complex
suspend fun getUserPreferences(userId: String): Preferences {
    return withContext(Dispatchers.IO) {
        userPreferencesDao.get(userId) // already suspending
    }
}

// better - just use the suspending DAO directly
suspend fun getUserPreferences(userId: String): Preferences =
    userPreferencesDao.get(userId)

xxxxxxxxxx

// unnecessarily complex

suspend fun getUserPreferences(userId: String): Preferences {

    return withContext(Dispatchers.IO) {

        userPreferencesDao.get(userId) // already suspending

    }

}

​

// better - just use the suspending DAO directly

suspend fun getUserPreferences(userId: String): Preferences =

    userPreferencesDao.get(userId)

Успешное решение: параллельная обработка данных

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

suspend fun processLargeDataSet(
    items: List<DataItem>,
    concurrency: Int = 10
): ProcessingResult = coroutineScope {
    val results = Channel<ProcessingResult>(Channel.BUFFERED)
    val jobs = Channel<DataItem>(concurrency)

    // launch processor coroutines
    repeat(concurrency) {
        launch {
            for (item in jobs) {
                try {
                    val result = processItem(item)
                    results.send(result)
                } catch (e: Exception) {
                    results.send(ProcessingResult.Error(e))
                }
            }
        }
    }

    // feed items to processors
    launch {
        items.forEach { jobs.send(it) }
        jobs.close()
    }

    // collect results
    val processedItems = mutableListOf<ProcessingResult>()
    repeat(items.size) {
        processedItems.add(results.receive())
    }

    ProcessingResult.Batch(processedItems)
}

xxxxxxxxxx

suspend fun processLargeDataSet(

    items: List<DataItem>,

    concurrency: Int = 10

): ProcessingResult = coroutineScope {

    val results = Channel<ProcessingResult>(Channel.BUFFERED)

    val jobs = Channel<DataItem>(concurrency)

​

    // launch processor coroutines

    repeat(concurrency) {

        launch {

            for (item in jobs) {

                try {

                    val result = processItem(item)

                    results.send(result)

                } catch (e: Exception) {

                    results.send(ProcessingResult.Error(e))

                }

            }

        }

    }

​

    // feed items to processors

    launch {

        items.forEach { jobs.send(it) }

        jobs.close()

    }

​

    // collect results

    val processedItems = mutableListOf<ProcessingResult>()

    repeat(items.size) {

        processedItems.add(results.receive())

    }

​

    ProcessingResult.Batch(processedItems)

}

Заключение

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

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

Источник