Morsel-driven parallelism

В прошлом уроке мы разрезали запрос на pipelines и выяснили: внутри одного pipeline батчи независимы, и потоки могут прокачивать их параллельно. Но остался главный вопрос — как именно входные данные pipeline раздаются потокам? Кто решает, какому потоку какую порцию обрабатывать? Ответ — morsel-driven parallelism, подход, на котором построен весь параллелизм DuckDB. Этот урок про то, что такое morsel, как работает диспетчер morsel-ов и почему этот способ раздачи работы бьёт привычное статическое разбиение.

Наивный способ и его провал

Допустим, нужно просканировать таблицу из 8 миллионов строк на 8 ядрах. Самое очевидное решение — статическое разбиение: поделить таблицу на 8 равных кусков по миллиону строк, отдать каждому потоку по куску, дождаться всех. Просто, очевидно, и на практике плохо работает.

Проблема в слове «равных». Куски равны по числу строк, но не по времени обработки. Причины разные. Сжатие: один кусок сжат RLE и распаковывается мгновенно, другой — bit packing с широкой шириной, и распаковка дороже. Фильтрация: после WHERE один кусок отдал много строк, другой почти все отбросил — дальнейшим операторам достаётся неравная нагрузка. Сами ядра: на реальной машине ядра не идентичны и не свободны одинаково — одно занято фоновым процессом ОС, другое тротлится по температуре, у третьего хуже доступ к памяти.

Итог: восемь «равных» кусков обрабатываются за разное время. Семь потоков закончили за секунду, восьмой возится три. И тут всплывает железное правило параллелизма: запрос завершён не тогда, когда закончил самый быстрый поток, а тогда, когда закончил самый медленный. Семь ядер простаивают, ожидая отстающего. Эта ситуация называется skew — перекос нагрузки, и статическое разбиение от неё принципиально не защищено: куски нарезаны заранее, в момент нарезки невозможно предсказать, какой окажется тяжёлым.

Morsel: маленький кусок работы

Morsel-driven parallelism решает skew сменой гранулярности. Слово «morsel» переводится как «кусочек», «крошка», и в этом вся идея: вместо того чтобы делить вход на крупные куски «по одному на поток», его делят на множество мелких — morsel-ов.

Morsel — это небольшая порция входных данных pipeline, единица работы, которую поток берёт и обрабатывает за раз. Маленькая по сравнению с целой таблицей: morsel — это кратное вектору 2048 количество строк. У этого размера есть характерное значение в коде DuckDB — PARTIAL_CHUNK_COUNT равно 50 умножить на 2048, около ста тысяч строк на morsel. Крупная таблица распадается не на 8 кусков, а на сотни и тысячи morsel-ов.

И раздаются morsel-ы не заранее. Никто не говорит «поток 1, твои morsel-ы с такого-то по такой-то». Morsel-ы лежат общим пулом, и потоки разбирают их на ходу.

Диспетчер: раздача по требованию

Раздачей morsel-ов заведует диспетчер. Его модель работы предельно проста и в этом её сила.

Все morsel-ы pipeline составляют общий пул необработанных кусков. Каждый рабочий поток крутит один и тот же цикл: пришёл к диспетчеру, попросил morsel, получил — обработал — вернулся за следующим. Освободился — снова просит. И так пока пул не опустеет.

Сравните это со статическим разбиением. Там нагрузку раздали заранее, один раз, вслепую. Здесь раздача идёт по требованию (on-demand): поток получает следующий кусок ровно тогда, когда освободился от предыдущего. Это и есть переход от «толкаем работу потокам» к «потоки сами тянут работу».

Почему это автоматически выравнивает нагрузку. Допустим, потоку достался тяжёлый morsel — он возится с ним дольше. Пока он занят, остальные потоки продолжают разбирать morsel-ы из пула — за то же время они успеют обработать больше штук. Быстрый поток сделает 12 morsel-ов, медленный — 7, но оба всё это время заняты полезной работой. Никто не простаивает в ожидании: пока в пуле есть необработанные morsel-ы, освободившийся поток немедленно берёт следующий. Нагрузка распределяется сама собой, пропорционально реальной скорости каждого потока, — а не по слепому предсказанию.

Почему мелкая гранулярность побеждает skew

Стоит разобрать, почему именно мелкий размер morsel-а снимает перекос — и почему перебарщивать с мелкостью тоже нельзя.

Чем мельче morsel, тем точнее выравнивание. Представьте крайность: один morsel на весь оставшийся объём — это и есть статическое разбиение, со всеми его бедами. Теперь другая крайность: morsel-ов очень много, каждый крошечный. Перекос становится почти невозможен. Даже если какой-то morsel случайно тяжёлый, он тяжёлый лишь относительно крошечного — его «лишнее» время мало в масштабе всего запроса. А в самом конце, когда работа подходит к концу, в пуле ещё остаются мелкие morsel-ы, которыми можно догрузить освободившиеся потоки. С крупными кусками такой «дозагрузки» нет: последний крупный кусок не на что разбить.

Но и бесконечно мельчить нельзя — есть встречное давление. Каждый поход к диспетчеру за morsel-ом — это пусть крошечная, но накладная работа: синхронизация доступа к общему пулу. Если morsel-ы сделать совсем микроскопическими, потоки будут чаще ходить к диспетчеру, чем считать, и накладные расходы съедят выигрыш.

Размер morsel-а в DuckDB — PARTIAL_CHUNK_COUNT около 100 тысяч строк — это и есть найденный баланс. Достаточно мелко, чтобы перекос практически исчез и оставался запас morsel-ов для дозагрузки в конце. И достаточно крупно, чтобы поток успевал сделать осмысленную порцию работы между двумя визитами к диспетчеру, и синхронизация на пуле не стала узким местом. Та же логика «найти золотую середину гранулярности», что и при выборе размера вектора 2048 и размера row group.

Что это даёт на практике

Из morsel-driven подхода вытекают свойства, которые ощущаются в реальной работе с DuckDB.

Масштабирование по ядрам близко к линейному. Раз нагрузка раздаётся динамически и перекос мал, добавление ядер реально ускоряет запрос: каждое новое ядро просто включается в разбор общего пула morsel-ов. Удвоили число ядер — запрос на крупной таблице ускоряется почти вдвое, без ручной настройки.

Устойчивость к неоднородности. Реальные данные неоднородны: где-то сжатие плотнее, где-то фильтр избирательнее. Статическое разбиение спотыкается об эту неоднородность, morsel-driven к ней безразлично — какой morsel ни окажись тяжёлым, его «лишнее» время растворяется среди сотен других.

Параллелизм не требует настройки от пользователя. Вы не делите таблицу на куски, не назначаете потоки, не балансируете нагрузку руками. Диспетчер morsel-ов делает это сам для любого запроса. Параллелизм в DuckDB — поведение по умолчанию.

При этом morsel-driven описывает раздачу работы внутри одного pipeline. На границах pipelines, на pipeline breakers, потоки по-прежнему синхронизируются — это мы видели в прошлом уроке. А как morsel-driven раскладывается на конкретные операторы — параллельное сканирование, параллельный hash join, параллельную агрегацию — разберут следующие уроки модуля. Сейчас главное: morsel — мелкий кусок работы, диспетчер раздаёт morsel-ы по требованию, и именно это делает параллелизм DuckDB равномерным и масштабируемым.

Попробуй сам

Понаблюдайте за параллелизмом и его масштабированием.

1. Создайте крупную таблицу: CREATE TABLE big AS SELECT range AS id, (random() * 1000)::INTEGER AS v FROM range(50000000); и CHECKPOINT.
2. Узнайте число потоков по умолчанию: SELECT current_setting('threads');. Обычно это число логических ядер машины.
3. Выполните тяжёлый запрос с замером времени: .timer on, затем SELECT v, count(*) FROM big GROUP BY v;. Запишите время.
4. Ограничьте DuckDB одним потоком: SET threads = 1;. Повторите тот же запрос, запишите время. Во сколько раз медленнее?
5. Верните половину ядер: SET threads = 4; (или половину вашего числа), повторите. Затем все ядра. Постройте мысленный график время-от-числа-потоков. Близко ли ускорение к линейному?
6. Поразмышляйте: при threads = 1 morsel-ы всё равно нарезаются, но разбирает их один поток. Что именно даёт прирост при увеличении threads — и почему этот прирост не требовал от вас ручного разбиения таблицы?

Этот эксперимент показывает morsel-driven параллелизм в действии: вы меняете только число потоков, а диспетчер сам раздаёт им работу, и время масштабируется почти линейно.