Паралельна обробка файлів — async + awaitAll()

1. Іноді хочеться обробляти файли паралельно

Коли ви лише починаєте програмувати, здається логічним: «ну, файли ж ідуть один за одним — прочитав, обробив, перейшов далі». І це справді працює… доки файлів мало й вони невеликі. Але в реальності часто буває так: «у мене тека з логами / фото / звітами / вихідним кодом». І треба зробити для кожного файла одне й те саме: порахувати розмір, кількість рядків, перевірити формат, витягти метадані, скопіювати, перейменувати, зібрати звіт.

І ось тут зʼявляється цілком природне бажання: якщо файлів багато, то чому б не обробляти кілька штук одночасно? Корутини якраз і придумали для того, щоб «запускати багато завдань» без потреби вручну плодити потоки, від яких потім хочеться сховатися під стіл. Kotlin прямо описує корутини як легкий механізм конкурентності: він дає змогу призупиняти виконання, не блокуючи системні ресурси.

Але є пастка: «паралельно» не означає «якомога більше й одразу». Файлова система й диск не нескінченні. Тож у цій лекції буде два шари: як розпаралелити роботу — і як не перестаратися.

2. launch і async: схожі зовні, різні за змістом

У цьому місці новачки зазвичай починають підозрювати, що програмісти навмисно вигадують схожі слова, аби ви страждали. І… так, інколи це справді схоже на правду.

І launch, і async — це «білдери» корутин, тобто функції, які запускають корутину. У документації Kotlin їх прямо називають базовими побудовниками корутин. Але призначення в них різне:

• launch — «запусти й керуй цим як Job»: корутина для дії.
• async — «запусти й віддай мені результат згодом»: корутина для результату, вона повертає Deferred<T>.

Щоб було простіше тримати це в голові, ось невелика таблиця:

launch { ... }

Job

async { ... }

Deferred<T>

Міні‑приклад: запускаємо дію через launch

import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking

fun main() = runBlocking {
    val job = launch {
        println("Я просто щось роблю") // Я просто щось роблю
    }
    job.join()
}

Міні‑приклад: запускаємо обчислення й чекаємо результат через async

import kotlinx.coroutines.async
import kotlinx.coroutines.runBlocking

fun main() = runBlocking {
    val d = async { 2 + 3 }
    println(d.await()) // 5
}

У цій лекції нас цікавить саме async, адже під час обробки файлів нам майже завжди потрібен результат: розмір файла, статистика, текст, хеш, звіт — що завгодно.

3. Практичний приклад: утиліта File Reporter

Щоб приклади не виглядали як набір розрізнених заклинань, уявімо, що ми пишемо консольну утиліту File Reporter. Вона отримує список файлів і будує короткий звіт: імʼя, розмір, кількість рядків (для текстових файлів).

Модель даних зробимо простою — без ООП‑«фокусів на виріст». Лише data class, бо нам потрібно зручно зберігати результат.

import java.io.File

data class FileReport(
    val file: File,
    val sizeBytes: Long,
    val lineCount: Int
)

Тепер зробімо suspend‑функцію, яка обробляє один файл. Важливий момент: читання файла — у Dispatchers.IO, а всередині використовуємо .use {}.

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.withContext
import java.io.File

suspend fun buildReport(file: File): FileReport =
    withContext(Dispatchers.IO) {
        val size = file.length()
        val lines = file.bufferedReader().use { it.lineSequence().count() }
        FileReport(file = file, sizeBytes = size, lineCount = lines)
    }

Так, lineSequence().count() усередині виглядає доволі «розумно». Але за змістом усе просто: читаємо рядки й рахуємо.

4. Шаблон: багато файлів → багато async → один awaitAll()

Найприємніша частина сьогоднішньої лекції така: щойно ви один раз побачите цей шаблон, ви почнете помічати його в коді всюди. Він схожий на «конвеєр»: ми створюємо набір завдань, а потім чекаємо, доки всі вони завершаться.

Технічно це виглядає так:

1. у нас є List<File>
2. ми робимо map { async { ... } } і отримуємо List<Deferred<FileReport>>
3. викликаємо awaitAll() і отримуємо List<FileReport>

awaitAll() зручний тим, що «збирає результати пакетом». Це суттєво спрощує код порівняно з ручним for і await().

Приклад: паралельно побудувати звіти за списком файлів

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.runBlocking
import java.io.File

fun main() = runBlocking {
    val files = listOf(File("a.txt"), File("b.txt"), File("c.txt"))

    val reports = files
        .map { f -> async(Dispatchers.IO) { buildReport(f) } }
        .awaitAll()

    println("reports=${reports.size}") // reports=3
}

Зверніть увагу на маленьку деталь: ми передали Dispatchers.IO прямо в async(...). Це означає, що корутина для кожного файла виконуватиметься в I/O‑контексті. Тобто блокувальне читання файла тут допустиме. Це узгоджується з нашою моделлю «файли — в IO», яку ми прийняли раніше.

І так: у документації Kotlin підкреслюють, що launch() і async() — базові побудовники корутин. Вони запускають конкурентні завдання всередині CoroutineScope.

5. Чому «тисяча async» — не завжди свято

У якийсь момент студент робить так: «У мене 20 000 файлів. Я просто зроблю .map { async { ... } }.awaitAll()». І в цей момент компʼютер починає звучати як маленький аеропорт, а ви дізнаєтеся, що таке «деградація продуктивності від надмірного ентузіазму».

Проблема тут не в корутинах — корутини справді легкі. Проблема в іншому:

• кожне читання файла все одно робить блокувальні виклики ОС;
• файлова система й диск не прискоряться від того, що ви попросили «20 тисяч операцій одночасно»;
• ви додатково навантажуєте планувальник завдань, черги I/O, кеші й усе, що між вами та диском.

Тому робимо наступний «інженерний» крок: обмежуємо паралелізм.

Dispatchers.IO.limitedParallelism(n)

Ідея проста: ми беремо I/O‑диспетчер і кажемо: «дорогий диспетчере, будь ласка, не виконуй одночасно більше ніж n завдань». Решта нехай зачекають своєї черги.

import kotlinx.coroutines.Dispatchers

val io4 = Dispatchers.IO.limitedParallelism(4)

Потім запускаємо всі async на io4.

Приклад: обмеження одночасності до 4 файлів

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.runBlocking
import java.io.File

fun main() = runBlocking {
    val io4 = Dispatchers.IO.limitedParallelism(4)
    val files = (1..10).map { File("file-$it.txt") }

    val reports = files
        .map { f -> async(io4) { buildReport(f) } }
        .awaitAll()

    println("done=${reports.size}") // done=10
}

Зауважте: файлів 10, а паралелізм — 4. Це не означає, що ми обробимо лише 4 файли. Це означає, що одночасно «в роботі» буде максимум 4 завдання, а решта чекатимуть, доки звільниться слот.

Як вибрати n

Тут немає «вічно правильного» числа — і це нормально. У навчальному проєкті зазвичай достатньо 2–8. Якщо поставити 64 «бо гарно», ви, найімовірніше, не отримаєте прискорення. А іноді навіть уповільните роботу. У реальних проєктах паралелізм підбирають експериментально та/або задають через конфігурацію.

6. Межа I/O і CPU: чому це впливає на швидкість

Коли ви читаєте файл, ви робите I/O. Але часто після читання починається обробка: парсинг, пошук, підрахунки, нормалізація тексту, обчислення хеша. А це вже CPU‑робота.

Наївне рішення таке: «зроблю все всередині Dispatchers.IO, раз уже файл там». Працювати буде, але з погляду архітектури це нечітко: ви змішуєте різні типи навантаження.

Корисно тримати правило: в I/O‑межі робимо лише те, що повʼязано з файловою системою та потоками. Усе інше — звичайні обчислення, і для них часто краще підходить Dispatchers.Default.

Приклад: прочитати текст в IO, порахувати щось у Default

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.withContext
import java.io.File

suspend fun countWords(file: File): Int {
    val text = withContext(Dispatchers.IO) { file.readText() }
    return withContext(Dispatchers.Default) {
        text.split(Regex("\\s+")).count { it.isNotBlank() }
    }
}

Цей приклад навмисно короткий і неідеальний за продуктивністю (регулярні вирази та split можуть бути дорогими). Але він показує архітектурну межу: читання — окремо, обробка — окремо. Так програма стає передбачуванішою.

7. Збирання звітів: паралельно й безпечно

Тепер оформимо «серце» нашої лекції — функцію, яку можна спокійно викликати з main. Ми зробимо її suspend, щоб вона працювала в корутинному світі без runBlocking усередині. Це важлива дисципліна.

Також додамо параметр parallelism, щоб можна було керувати обмеженням одночасності.

Базовий варіант: buildReportsParallel(...)

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.coroutineScope
import java.io.File

suspend fun buildReportsParallel(
    files: List<File>,
    parallelism: Int
): List<FileReport> = coroutineScope {
    val io = Dispatchers.IO.limitedParallelism(parallelism)

    files
        .map { f -> async(io) { buildReport(f) } }
        .awaitAll()
}

Тут є важливий момент: ми використовуємо coroutineScope { ... }. Це означає, що всі створені async — «діти» поточної корутини. А сама функція не завершиться, доки всі «діти» не завершаться. Це і є structured concurrency в дії: завдання повʼязані життєвим циклом, а не «полетіли в космос і забулися». Kotlin описує контекст та ієрархію завдань як основу структурованої конкурентності.

Використання з main

import kotlinx.coroutines.runBlocking
import java.io.File

fun main() = runBlocking {
    val files = listOf(File("a.txt"), File("b.txt"), File("c.txt"))

    val reports = buildReportsParallel(files, parallelism = 2)

    for (r in reports) {
        println("${r.file.name}: ${r.lineCount} рядків") // наприклад: a.txt: 10 рядків
    }
}

Якщо один файл «поганий»: як не завалити все одразу

Зараз буде важлива (і трохи філософська) частина: конкурентність — це не лише про «швидше». Це ще й про «а що, якщо один із них упав?».

Поведінка за замовчуванням у такого коду доволі сувора: якщо всередині одного async стався виняток, він «спливе» під час await()/awaitAll(). Тобто помилка стане видимою там, де ви чекаєте результат. І це, чесно кажучи, хороша модель: ви або отримали всі звіти, або дізналися, що щось не вийшло.

У межах цієї лекції ми не будуватимемо складну стратегію часткових результатів (це окрема велика тема). Але покажемо простий прийом: якщо ви хочете, щоб обробка інших файлів тривала, ви можете всередині async перехоплювати помилку й повертати null.

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.coroutineScope
import java.io.File

suspend fun buildReportsOrSkip(
    files: List<File>,
    parallelism: Int
): List<FileReport> = coroutineScope {
    val io = Dispatchers.IO.limitedParallelism(parallelism)

    files
        .map { f ->
            async(io) {
                try { buildReport(f) } catch (e: Exception) { null }
            }
        }
        .awaitAll()
        .filterNotNull()
}

Це не «ідеальна обробка помилок», але як навчальний місток — дуже корисно. Ви бачите, як помилки впливають на збирання результатів і чому awaitAll() — це точка, у якій результат стає «обовʼязковим».

8. Типові помилки під час паралельної обробки файлів

Помилка № 1: запускати файлові операції на Dispatchers.Default.
З вигляду це нешкідливо: «ну, воно ж працює». Але так ви починаєте блокувати потоки, призначені для обчислень. У підсумку інша корутина, яка чесно рахує щось CPU‑важке, раптом змушена чекати, бо ви зайняли її потік читанням файла. Ліки прості й нудні, як техніка безпеки: файлові операції завжди відправляємо в Dispatchers.IO.

Помилка № 2: створювати async усередині map, але забувати про await/awaitAll.
Це класичне «я все запустив, але нічого не сталося». Насправді сталося: ви створили набір Deferred, але так і не дочекалися результатів. У найкращому разі програма завершиться зарано. У найгіршому — ви отримаєте дивні ефекти: робота ще триває, а ви вже друкуєте «готово». Лікується дисципліною: якщо використали async, десь має бути await() або awaitAll().

Помилка № 3: запускати надто багато завдань одночасно без обмеження.
Коли файлів багато, спокуса велика: «корутини ж легкі». Так, але диск не гумовий. Якщо ви запускаєте тисячі завдань читання одночасно, ви створюєте конкуренцію за ресурс, який і так один (ну гаразд, два, якщо у вас два диски). Часто це уповільнює виконання й робить систему менш чуйною. Хороша звичка — використовувати limitedParallelism(n) і починати з малого n, наприклад 4.

Помилка № 4: робити async(Dispatchers.IO) { withContext(Dispatchers.IO) { ... } }.
Це виглядає як «подвійний захист від дурня», але на практиці це просто зайва вкладеність і шум. Ви або запускаєте корутину на IO‑диспетчері, або перемикаєтеся на IO всередині. У нашому сьогоднішньому стилі зручніше так: async(ioLimited) { ... }, а всередині — просто викликаємо код роботи з файлами (або функції, які самі перемикаються; але тоді вдруге перемикатися не треба).

Помилка № 5: змішувати читання файла й важку обробку в одному IO‑контексті.
Такий код часто зʼявляється сам собою: «я вже в IO, отже зроблю тут усе». Але IO‑контекст призначений для блокувальних операцій, а не для «довго рахуємо регулярними виразами й парсимо гігабайт тексту». У результаті ви забиваєте I/O‑пул обчисленнями й втрачаєте передбачуваність. Набагато чистіше: I/O окремо, CPU окремо — хоча б у два кроки.