Распараллеливание пакетной обработки с ограничением количества используемых потоков

[gl00mie]

На многих проектах мне лично встречался такой типовой сценарий:

• Есть десятки-сотни-тысячи объектов, которые надо обработать/обсчитать в пакетном режиме. Это могут быть накладные для разноски, импортируемые из очереди сообщения, сеть розничных магазинов для обсчета каких-нибудь скидок/ассортиментов, etc.
• Реализуется "двухуровневая" логика:
  1. задача-планировщик разбирает очередь сообщений/документов, перебирает магазины и т.п.;
  2. задача-исполнитель обрабатывает один переданный ей объект.
  Пакетная инфраструктура позволяет удобно насоздавать из задачи-планировщика отдельную runtime-задачу на каждый обрабатываемый объект и распараллелить, таким образом, обработку, в том числе задействовав несколько пакетных серверов. Задача-планировщик может настраивать зависимости одних runtime-задач от других, скажем, финальная задача формирования и рассылки отчета может зависеть от завершения всех вычислительных runtime-задач.
• Есть при этом необходимость ограничить нагрузку такой распараллеливаемой обработки некоторым количеством потоков, чтобы умерить нагрузку на СУБД, и чтоб для других периодических операций остались свободные потоки пакетной обработки.
• Также может быть желание снизить накладные расходы пакетной инфраструктуры, связанные с планированием и выполнением большого количества пакетных задач.

В стандартном приложении, начиная с AX2009, есть пример пакетирования отдельных задач для уменьшения накладных расходов пакетной инфраструктуры - см. обсуждение классов BatchTaskBundle и BatchTaskBundleAssmebler. Там реализована попытка "взвешивания" отдельных пакетных задач с тем, чтобы пакеты получались примерно одинакового "веса" - в общем случае, чтобы каждый пакет обрабатывал примерно одинаковое количество строк документов. Правда, количество самих пакетов (и, таким образом, потоков пакетной обработки) при этом никак не ограничивается.
Во вложении представлена реализация классов BatchTaskBundleSimple и BatchTaskBundleMaxThreadsAssembler, позволяющих пакетировать отдельные runtime-задачи так, чтобы в итоге ограничить сверху используемое количество потоков пакетной обработки.

BatchTaskBundleMaxThreadsAssembler  bundleAssembler;
BatchHeader                         batchHeader;
List                                listOfBundles;
SysRunable                          myTask;
RetailStoreTable                    rtlStoreTable;
;
// предположительно мы - задача-планировщик, запущенная в пакетной обработке
batchHeader = BatchHeader::getCurrentBatchHeader();
bundleAssembler = BatchTaskBundleMaxThreadsAssembler::construct(
    BatchHeader::getCurrentBatchTask(),
    RetailParameters::find().MaxBatchThreads); // некое ограничение числа потоков
// нарезаем runtime-задачи на определенное число потоков пакетной обработки
while select StoreNumber from rtlStoreTable
{
    myTask = MyBatchTask::construct();
    myTask.parmStoreNumber(rtlStoreTable.StoreNumber);
    bundleAssembler.addTask(myTask, rtlStoreTable.StoreNumber);
}
// получаем список сформированных пакетов задач - экземпляров BatchTaskBundleSimple
listOfBundles = bundleAssembler.scheduleBundledTasks(batchHeader);
// добавляем еще runtime-задачи, настраиваем зависимости от задач в listOfBundles
// ...
// сохраняем все созданные runtime-задачи и зависимости в текущем пакетном задании
batchHeader.save();

Отличия от стандартных классов следующие:

• Можно "пакетировать" любые задачи (экземпляры классов) вперемешку, потому что не нужно уметь их взвешивать - учитывается только количество добавляемых задач, равномерно распределяемых по пакетам. В стандартной реализации класс-bundle должен заранее "знать", экземпляры каких классов будут в него добавляться.
• Можно добавлять экземпляры классов, не являющиеся наследниками RunBaseBatch, достаточно, чтобы они реализовывали SysRunable и SysPackable (да, наследники RunBase тоже можно будет поставить в пакетную обработку ). В стандартной реализации поддерживаются только наследники RunBaseBatch (SysOperationServiceController? неа, не слышали).
• Можно использовать классы, которые не реализуют статический ::create(container _packedClass), необходимый для распаковки коллекций классов. Все добавляемые runtime-задачи помещаются в List, который затем запаковывается. В стандарте при запуске в пакетной обработке List распаковывается штатными средствами - через List::create(). Если ваш класс не реализует статический ::create(container _packedClass), распаковка List с вашим классом внутри вызовет ошибку времени выполнения. В предложенной реализации List распаковывается "вручную", поэтому добавленным runtime-задачам достаточно реализовывать SysPackable.
• Можно передать свой BatchHeader для добавления runtime-задач на основе сформированных пакетов и получить на выходе список экземпляров классов этих пакетов без сохранения изменений BatchHeader в пакетном задании. Это позволит, к примеру, добавить другие runtime-задачи в пакетное задание и настроить зависимости их запуска от сформированных задач пакетов (условно, та самая финальная рассылка отчета). Стандартный BatchTaskBundleAssembler предполагает, что кроме формирования пакетов задач в пакетном задании больше ничего делать не нужно, и никакие зависимости других задач от формируемых им пакетов вы уже не настроите.
• Можно при добавлении runtime-задачи в пакет указать её короткий идентификатор (код заказа/сообщения/магазина/склада/еще что) - потом из таких идентификаторов будут автоматически сформированы заголовки пакетов. Правда, нужно помнить, что BatchCaption штатно имеет длину всего 100 символов.
• По мере выполнения задач пакета обновляется штатный "градусник", т.е. при выполнении пакетного задания на форме "Просмотр задач" для каждого пакета будет виден прогресс выполнения, пропорциональный количеству завершившихся внутри него задач. Это может быть особенно полезно для службы поддержки - понимать, а не зависло ли пакетное задание.