Использование OmniThread Libray 2.0 (OTL 2.0) для создания многопоточных приложений - 5

Продолжение, начало в http://deeptown12.blogspot.com/2011/01/omnithread-libray-20-otl-20-4.html

Регулировка стадий

В нашем демонстрационном примере (41_Pipeline) большое количество данных (один миллион чисел) передаются через многоступенчатый процесс. Если один поток будет приостановлен в течение некоторого времени или поток будет выполнять вычисление, по времени медленнее чем время вычисления предыдущего потока то может возникнуть ситуация когда входная очередь этого потока может заполниться данными под которые будет выделено слишком много памяти, которую в свою очередь позже нужно будет освобождать. Для таких случаев для более правильной регулировки выделения памяти трубопровод использует новые возможности блокирующей коллекции, появившиеся в версии 2.0.

Регулировка варьирует максимальный размер (в модуле TOmniValue) коллекции блокирования. Когда количества данных в блокирующей коллекции будет сохранено и дойдет до предела заданного регулировкой, то последующее добавление в коллекцию данных будет остановлено. То есть последующее добавление станет невозможным.

Функция добавления в коллекцию  (Add) просто заблокируется, пока коллекция не будет достаточно пуста снова, или не будет вызван метод CompleteAdding.

Коллекция считается достаточно пуста, когда данные считываются (удаляются из коллекции) ниже некоторого предела, которое можно передать в параметре unblockAtCount метода Throttle. Если параметр не используется то по умолчанию OTL вычислит этот предел как 3/4 максимального размера.

Количество параллельных задач для стадии многоступенчатого процесса  (трубопровода)

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

Как вы видите из представленной выше диаграммы, всегда есть только одна очередь, расположенная между стадиями, даже если для стадии еcть несколько фоновых задач. Такая поддержка многократного чтения и многократной записи в потокозащищенной (threadesafe) манере легко достигнуто в IOmniBlockingCollection

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