Параллельная обработка данных в режиме конкурирующих потребителей (competing consumers)

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

Схема:

• есть один процесс - мастер и несколько однородных процессов (экземпляров) - воркеров;

• данные нарезаются на блоки, каждый из которых обрабатывается независимо от остальных;

• мастер занимается отбором, подготовкой блоков и вкладкой их на "конвейер" (очередь на обработку), откуда их параллельно разбирают воркеры;

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

Главное условие корректности: результат обработки блока не зависит от других блоков и от порядка их обработки.

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

Что гарантируется, а что нет

Гарантируется:

• каждый успешно поставленный в очередь блок обрабатывается не более одного раза;

• воркеры изолированы отдельными процессами (падение одного воркера не затрагивает остальных)

• мастер прибирает очереди и завершает всех воркеров при любом исходе прогона.

Не гарантируется:

• порядок обработки блоков;

• привязка блока к конкретному воркеру;

• повторная доставка. Подтверждений (ack) нет: если воркер погиб, взятый им блок теряется. Если задача не допускает потерь, нужна собственная идемпотентность или внешний журнал.

Граничные условия

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

• пакетная обработка и преобразование документов;

• разбор и индексация большого числа записей;

• параллельные вычисления по разбиениям;

• любые «embarrassingly parallel» задачи.

Плохо подходит, если:

• важен строгий порядок обработки;

• блоки требуют общего изменяемого состояния;

• недопустима потеря блока при падении воркера;

• блоки слишком мелкие и накладные расходы на очередь перевешивают выгоду.

Как это работает

Основная реализация на IBM i. Ну просто потому что под ней работаю и у нас такая схема обработки востребована в силу нашей специфики.

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

Основные установки на момент начала проектирования

• минимальное количество библиотечных функций;

• код библиотечных функций скрыт от конечного разработчика;

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

Сразу скажу что все получилось как хотелось.

• Весь код скрыт в сервисной программе (*SRVPGM, аналог динамической библиотеки), для разработки конкретного приложения нужно просто подключить *BNDDIR (аналог библиотеки импорта) + один заголовочный файл;

• весь интерфейс уложился в три (!!!) функции: в головном задании startMaster и в воркере initWorker (если требуется построение окружения для задания-обработчика) и startWorker;

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

В качестве транспорта (конвейера) была выбрана очередь UserQueue (*USRQ) - не полный, но достаточно близкий функциональный аналог Posix MQ в Linux. Ранее я уже разрабатывал удобный API для работы с ними. Он стабилен и давно работает в промсреде.

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

Конфигурация хранится в БД в отдельной таблице:

• имя конфигурации (уникальное);

• краткое описание;

• имя программы-воркера;

• имя очереди данных;

• имя контрольной очереди;

• имя конфигурации очереди данных в таблице конфигураций USRQ;

• имя конфигурации очереди данных в таблице конфигураций USRQ;

• размер пакета данных

• количество экземпляров воркеров;

• интервал посылки ALIVE воркером;

• EMPTY - время ожидания воркером при пустой очереди если не был получен пакет "конец данных";

• уровени логирования раздельно для мастера и воркера.

Мастер

При запуске мастер читает заданную ему конфигурацию и валидирует ее. Если валидация прошла успешно, запускает заданное количество воркеров передавая им параметром имя конфигурации. Каждый экземпляр воркера запускается в отдельном задании (на IBM i это делается командой SBMJOB - Submit Job) и далее переходит в режим ожидания - каждый воркер после успешного запуска должен прислать подтверждение - команду READY через контрольную очередь. Время ожидания рассчитывается от параметра ALIVE: 2 * ALIVE + 5сек.

Если в заданное время не было получено подтверждение готовности от кого-то из воркеров, тем, кто успешно стартовал, рассылается команда на завершение работы, для не стартовавших же проводится анализ состояния задания. Если задание завершилось (видимо, аварийно) - перешло в статус OUTQ (на IBM i) - ничего делать не надо. А вот если оно остается в состоянии ACTIVE (например, зациклилось где-то на инициализации), его надо прибить командой ENDJOB. Дальше остается дождаться подтверждения завершения (команда FINISH в контрольной очереди) от успешно стартовавших и выйти с ошибкой неудачного запуска.

Если все воркеры стартовали успешно, мастер начинает отбор данных для обработки, формирование пакетов и выкладку их на конвейер (в очередь данных)

Элементы для обработки как правило объединяются в пакеты. Размер (количество элементов в пакете) выбирается таким, чтобы обработка его не занимала слишком много времени с одной стороны (это в целом выравнивает среднюю утилизацию воркеров за прогон), а, с другой, слишком мелкий пакет увеличивает общее их количество и накладные расходы на транспорт.

В процессе выкладки мастер контролирует заполнение очереди. Преимущество использования очередей перед, например, пайпами или сокетами в том, что у очереди есть операция "материализации" позволяющее видеть, в частности, текущее количество сообщений в ней. Это позволяет контролировать заполнение очереди и оценивать со сторорны мастера скорость разбора пакетов воркерами.

В данном случае был выбран жестко закрепленный вариант "короткой очереди" - емкость ее привязана к количеству воркеров и составляет 8 пакетов на воркера с гистерезисом 6/4 - как только мастер видит что заполнение очереди составляет более чем 6 пакетов на воркера, он приостанавливает раздачу и ждет пока уровень заполнения не упадет до 4-х пакетов на воркера.

В цикле выдачи мастер разом выкладывает в очередь "пачку" пактов по числу воркеров.

В перерывах между пачками мастер проверяет контрольную очередь - каждый воркер должен туда что-то посылать не реже чем раз в ALIVE секунд (это или команда ALIVE, или, если случилась какая-то ошибка, сообщение об ошибке). Если от воркера ничего не было в течении 2 * ALIVE, он переводится в разряд "подозреваемых" - проверяется статус его задания. Тут возможны два варианта - задание завершилось (скорее всего, падение по системному исключению) и перешло в состояние OUTQ, или оно продолжает жить в состоянии ACTIVE. В этом случае дополнительно проверяется его статус - оно может выдать ошибку с ожиданием ответа оператора - статус MSGQ. Тогда задание принудительно завершается командой ENDJOB.

Иначе, если ACTIVE и не MSGQ, воркер остается "подозреваемым" и проверяется еще через один период ALIVE. Если от него до сих пор (т.е. в течении 3 * ALIVE) не было никаких известий, предполагается что он зациклился и его задание принудительно завершается (ENDJOB).

После того, как данные закончились, мастер выкладывает пачку "терминаторов" - пустых пакетов с командой STOP после чего переходит в режим ожидания (продолжая контролировать воркеров и состояние очереди). Как только очередь опустела - значит воркеры обработали все данные и все получили команду STOP. Далее начинается цикл ожидания - каждый воркер должен прислать пакет о том, что он завершил свою работу - FINISH. Здесь действуют те же правила что и при контроле ALIVE.

Дальше контрольная проверка что все задания воркеров перешли в OUTQ (завершились) и выход.

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

Воркер

Тут все намного проще. Воркер в цикле берет очередной пакет из очередной пакет из очереди и обрабатывает его. И так по кругу пока не получит пустой пакет с командой STOP на который отвечает командой FINISH в контрольную очередь и завершает работу (задание).

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

Единственное, что может контролировать воркер - пустая очередь. Если он видит что очередь пуста и при этом не было получено команды STOP в течении 3 * EMPTY секунд, воркер считает что мастер перестал нормально работать и завершается.

Логирование

Лог традиционно для наших реализаций ведется в виде таблицы в БД. Уровни логирования задаются раздельно для мастера и воркера. Всего возможно 4 уровня

• N - лог отключен

• E - логируются только ошибки

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

• T - трассировка в процессе работы мастера и воркера

Режимы S и T - служебные, используются в процессе отладки и тестов. По результата сбора статистики, в частности, можно оценивать время обработки пакета, уровень утилизации воркоров (и подбирать оптимальное их количество).

Состав статистики:

Как это выглядит для разработчика конкретного приложения

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

Оптимально для этого подходит механизм callback-функций. Это как раз то, что, собственно, пишет разработчик. На стороне мастера таких функций две - подготовка очередного пакета данных и отбработка ошибки, полученной мастером от воркера через контрольную очередь.

На стороне воркера одна функция - обработка очередного пакета данных.

Итого, код мастера выглядит примерно так (мы пишем на RPG)

**free
    CTL-OPT  ACTGRP(*CALLER) DFTACTGRP(*NO)
             OPTION(*SRCSTMT : *NODEBUGIO)
             BNDDIR('MPQBNDDIR')
             MAIN(MYMASTER);

      /INCLUDE MPQCPYSRC,MPQRPGP

dcl-c CFG_NAME const('MYCFG');

dcl-proc MYMASTER;
  dcl-pi *n;
  end-pi;

  dcl-s ok ind inz(*off);
  dcl-ds err likeds(t_dsMPError);

  clear err;
  ok = mpqStartMaster(CFG_NAME : %paddr(myNextData)
                              : %paddr(myOnError) : err);

  if not ok;
    // err содержит причину; при MPQ0019 в eP1 - имя неверного поля.
    // Здесь: сообщить оператору / записать в журнал приложения.
  endif;

  return;

  begsr *pssr;
    dump;
  endsr;
end-proc;

// Подготовка очередного блока. Вызывается многократно.
dcl-proc myNextData;
  dcl-pi *n ind;
    oData char(MP_BODY_MAX) options(*varsize); // тело пакета
    oLen  int(10);                             // вход: ёмкость, выход: длина
    oErr  likeds(t_dsMPError);
  end-pi;

  dcl-s cap int(10) inz(0);

  clear oErr;
  cap = oLen;               // MP_BODY_MAX

  // ... прочитать/сформировать очередной блок в oData (учесть cap) ...
  // oLen = <длина блока>;
  // return *on;            - блок будет отправлен
  // oLen = 0; return *on;  - блока пока нет, мастер продолжит опрос
  // return *off;           - данные закончились

  return *on;

  begsr *pssr;
    dump;
  endsr;
end-proc;

// Ошибки прогона: ошибки воркеров, ошибки мастера, принудительные завершения.
dcl-proc myOnError;
  dcl-pi *n ind;
    oErr likeds(t_dsMPError);
  end-pi;

  // ... запись в журнал приложения, счётчики, пороги ...
  // return *off;  - приложение считает ошибку критической: прогон свернётся

  return *on;

  begsr *pssr;
    dump;
  endsr;
end-proc;

Главная процедура мастер-приложения - MYMASTER. Из нее просто вызывается mpqStartMaster куда передается имя конфигурации и адреса пользовательских callback функций формирования очередного пакета (myNextData) и реакции на ошибку (myOnError). И та и другая функции возвращают логическое значение *on - продолжаем работу или *off - завершаем (т.е., например, в случае возникновения "шторма ошибок" от воркера пользователь внутри функции обработки ошибки может принять решение "ну все, хватит" и завершить прогон с ошибкой).

Аналогично код воркера

**free
    CTL-OPT  ACTGRP(*CALLER) DFTACTGRP(*NO)
             OPTION(*SRCSTMT : *NODEBUGIO)
             BNDDIR('MPQBNDDIR')
             MAIN(MYWRKPGM);

      /INCLUDE MPQCPYSRC,MPQRPGP     // публичный API MPQ, t_dsMPError

// Точка входа задания-воркера: имя конфигурации приходит от мастера.
dcl-proc MYWRKPGM;
  dcl-pi *n;
    cfgName char(10) const;
  end-pi;

  dcl-s ok ind inz(*off);
  dcl-ds err likeds(t_dsMPError);

  // Окружение задания по мнемонике FOP (если MPQFOP заполнено).
  mpqInitWorker(cfgName);
  // Цикл: очередь -> колбэк на каждый блок -> Finish по команде мастера.
  ok = mpqStartWorker(cfgName : %paddr(myProcess) : err);

  if not ok and err.eId <> *blanks;
    // Ошибка уже записана в лог воркера; при необходимости - свой журнал.
    // Здесь: сообщить оператору / вернуть код приложения.
  endif;

  return;

  begsr *pssr;
    dump;
  endsr;
end-proc;

// Обработка одного блока. Блоки независимы: результат не зависит
// от других блоков и от порядка обработки.
dcl-proc myProcess;
  dcl-pi *n ind;
    pBlock char(MP_BODY_MAX) options(*varsize);  // тело блока
    nLen   int(10);                              // длина блока
    oErr   likeds(t_dsMPError);
  end-pi;

  clear oErr;

  // Читаются только nLen байт: %subst(pBlock : 1 : nLen).
  // ... обработка блока ...
  // Неблокирующая ошибка: заполнить oErr и вернуть *on - цикл продолжится.
  // Досрочное завершение воркера: вернуть *off.

  return *on;

  begsr *pssr;
    dump;
  endsr;
end-proc;

Опять - MYWRKPGM - главная процедура где сначала вызывается построение окружения mpqInitWorker и затем mpqStartWorker с передачей ему адреса функции обработки очередного пакета данных myProcess которую уже пишет разработчик конкретного приложения.

mpqInitWorker и mpqStartWorker разнесены т.к. (теоретически) могут потребоваться какие-то действия после того как окружение задания построено, но перед запуском собственно воркера.

Здесь тоже разработчик может принять решение о досрочном завершении воркера по ошибке, которую он сочтет критической, просто вернув *off из myProcess. Естественно, мастеру об этом будент направлено уведомление-ошибка через контрольную очередь которая в конечном итоге попадет в обработчик ошибок myOnError уже на стороне мастера.

Имя конфигурации передается в главную процедуру параметром. А момент запуска воркера эта конфигурация уже провалидирована мастером и признана корректной и согласованной.

И еще раз о конфигурации

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

Число воркеров (MPWRKCNT)

Закон Литтла: чтобы обрабатывать r блоков в секунду при среднем времени обработки блока proc, нужно r × proc занятых воркеров, плюс один запас на разброс и паузы:

MPWRKCNT = ceil(r_target × proc_avg) + 1

proc_avg берётся из статистики прогона либо из замера на опытном прогоне.

Ориентиры: не более числа физических ядер + 1. Слишком много воркеров не ускоряет работу, а добавляет переключения и трафик ALIVE. Но это скорее касается слабых серверов.

Проверка по утилизации: Σ(blk × proc) / (period × MPWRKCNT).

Размер блока (MPDATASZ)

• Крупный блок выгоднее мелкого: каждый блок — это обмен через очередь и работа мастера с воркером.

• Число блоков должно заметно превышать MPWRKCNT, иначе последние воркеры простаивают.

Период ALIVE (MPALVSEC)

Задаётся с запасом 2–3× к максимальному времени обработки блока. Таймауты фаз старта и завершения считаются из него автоматически.

Практическое правило: занижение всегда даёт ложную аварию, завышение — только задержку обнаружения настоящей. На пропускную способность нормального прогона тайминги не влияют (проверено: при MPALVSEC 5 и 1 скорость подачи одинаковая).

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

Таймауты же в подавляющем большинстве случаев могут выбираться в 3-5 секунд и это будет работать.

Что можно сделать еще, но пока не реализовано

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

Общая идея такова - в конфигурации задаются границы (минимальное/максимальное) по количеству воркеров и размер временного окна для оценки статистики.

Запуск идет с некоего среднего значения воркеров. Далее в течении временного окна снимается статистика с оценкой уровня утилизации воркеров и работы очереди

Это по состоянию очереди. По утилизации воркеров оценка приводилась выше.

На основании этого можно делать вывод о том в какую сторону двигаться - увеличивать или уменьшать количество воркеров.

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

В целом как это сделать понятно, но есть моменты которые потребуют проработки в процессе тестов.

• порядок размера тестового окна сбора статистики - секунды? минуты?

• время стабилизации после того, как количество воркеров изменилось и до начала сбора новой статистики - одного окна хватит? или нужно больше - два?

• где и как задавать гистерезисы изменения количества воркеров чтобы избежать гонок (косвенно связано с предыдущим пунктом)?

• надо ли учитывать стабильный разброс в утилизации воркеров если он есть - например, когда два воркера берут на себя 80% работы, а остальные три всего 20% - явно избыток воркеров?

• и главный вопрос - оно того стоит? дело в том, что запуск/остановка задания не бесплатное с точки зрения нагрузки на систему - окупится ли выгода динамической балансировки?