PATTERNS OF PARALLEL PROGRAMMING / ПАТТЕРНЫ ПАРАЛЛЕЛЬНОГО ПРОГРАММИРОВАНИЯ: ПОСТАВЩИК/ ПОТРЕБИТЕЛЬ

В реальном мире все вращается вокруг паттерна "поставщик/потребитель". Отдельные сущности отвечают за определенные функции: одни производят материал, который в конце концов потребляют другие. Случается, что потребители являются поставщиками для дальнейших потребителей. Иногда присутствует несколько поставщиков для одного потребителя, иногда - наоборот - несколько потребителей у одного поставщика, а бывает и вовсе много поставщиков для многих потребителей. Мы живем в мире поставщиков/потребителей и в компьютерном мире этот паттерн также играет важную роль.

vzhik 21.07.11 в 18:41

Зачастую связка поставщик-потребитель применяется в параллелизме, когда невозможно распараллелить отдельную операцию, но несколько операций могут быть выполнены параллельно и каждая следующая операция зависит от предыдущей. Например, если необходимо сжать и зашифровать некоторый файл. Это можно сделать последовательно: читая, сжимая и шифруя порции данных, пока файл не будет полностью обработан. В зависимости от используемых алгоритмов не всегда можно распараллелить само сжатие или шифрование и нельзя сжимать и шифровать данные параллельно, потому что алгоритму шифрования нужна уже сжатая, а не исходная последовательность. Тогда организовывается некоторое подобие конвейера из нескольких потоков. Первый поток читает данные. Он же передает считанные данные во второй поток, который сжимает их и в свою очередь отправляет результат сжатия третьему потоку. Третий поток их шифрует и посылает четвертому потоку, который записывает зашифрованные данные в файл. Каждый участвующий в обработке "агент" (или "актер"), в такой схеме сам по себе последователен и только преобразовывает входные данные в выходные. А поскольку передачи между агентами не используют никаких переупорядочивающих операций, то на выходе всей цепочки данные появляются в том порядке, в каком передаются в нее.

vzhik 25.07.11 в 14:25

Такой конвейер можно организовать с помощью нового типа BlockingCollection {T}, который обеспечивает поддержку этого паттерна в .NET Framework 4.

vzhik 03.08.11 в 06:42

КОНВЕЙЕРЫ

vzhik 03.08.11 в 06:42

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

vzhik 03.08.11 в 06:58

Мы можем построить упрощенную версию такого механизма передачи с использованием Queue{T} и SemaphoreSlim:

vzhik 03.08.11 в 07:02

Это весьма простая "блокирующая очередь". Поставщики вызывают метод Enqueue чтобы вставить данные в очередь, который добавляет их во внутренний объект Queue{T} и с помощью семафора уведомляет потребителей о том, что доступен новый элемент данных. Аналогично потребители вызывают Dequeue, который ожидает появления элемента данных и после извлечения удаляет его из внутренней очереди. Обратите внимание, что поскольку к структуре данных может выполняться одновременный доступ из разных потоков, используется блокировка для защиты непотокобезопасного экземпляра Queue{T}.

vzhik 03.08.11 в 07:55

Другая реализация использует возможности уведомления объекта Monitor вместо семафора:

vzhik 03.08.11 в 09:30

Эти реализации предоставляют базовую поддержку обмена данными между потоками, но вместе с тем имеют ряд важных недостатков. Как поставщики сообщат о том, что новых элементов больше не будет? При такой блокирующей природе, что, если потребитель хочет ожидать некоторое ограниченное время, а затем продолжать работу? Как приостановить поставщика и запретить добавлять новые элементы, когда внутренняя очередь переполнена? Что если Вы хотите получать данные из нескольких блокирующих очередей, а не из одной? Что если для внутреннего хранилища требуется семантика, отличная от "первый-вошел-первый-вышел"(FIFO)? Как отменять поставщиков и потребителей? И так далее.

vzhik 03.08.11 в 11:29

Ответ на все эти вопросы дает новый тип System.Collections.Concurrent.BlockingCollection{T} из сборки System.dll в .NET Framework 4. Он предоставляет ту же базовые возможности, что и приведенные выше простоватые реализации: методы для добавления и извлечения элементов коллекции. А еще позволяет притормаживать поставщиков и потребителей, задавать периоды ожиданий, указывать произвольные внутренние структуры данных и много другое. Также предоставляет встроенные реализации типовых шаблонов кодирования, относящихся к поставщику/потребителю для облегчения использования этих шаблонов.

vzhik 03.08.11 в 11:46

В качестве обычного примера шаблона поставщик/потребитель рассмотрим задачу чтения файла, преобразования каждой строки файла с помощью регулярного выражения и записи измененной строки в файл. Мы можем решить ее, используя Task для асинхронного запуска каждого уровня конвейера и BlockingCollection{string} в качестве точки передачи между всеми уровнями.

vzhik 03.08.11 в 13:11

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

vzhik 04.08.11 в 07:49

Указав параметр boundedCapacity (для ясности здесь используется именованный параметр, поддерживаемый языками C# и Visual Basic в Visual Studio 2010) мы блокируем попытки добавления новых элементов поставщиком, пока количество элементов в коллекции не станет меньше 20, тем самым замедляя чтение файла. С другой стороны, мы можем наращивать параллелизм решения. Предположим, что в результате тестирования обнаружилась проблема: вычисление processLines приводит к высокой нагрузке. Тогда вы можете распараллелить этот процесс, используя PLINQ для подключения большего количества ядер:

vzhik 04.08.11 в 08:55

Заметьте, что указав “.AsOrdered()” после “.AsParallel()” мы гарантируем, что PLINQ сохранит тот же порядок, что и при последовательном выполнении.

vzhik 03.08.11 в 13:55

ДЕКОРАТОР К КОНВЕЙЕРУ

vzhik 04.08.11 в 09:51

Декоратор является одним из архитектурных шаблонов Банды Четырех. Декоратор выставляет тот же интерфейс, что и содержащийся в нем объект. В терминах ООП это объект имеющий связи "является" ("is-a") и "содержит" ("has-a") с конкретным типом. Рассмотрим класс CryptoStream из пространства имен System.Security.Cryptography. CryptoStream наследуется от типа Stream ("является" Stream), а заодно принимает Stream в конструктор и хранит Stream внутри ("содержит" Stream), куда и записываются шифрованные данные. Поэтому CryptoStream является декоратором.

vzhik 04.08.11 в 13:34

Мы обычно выстраиваем цепочку из декораторов. Например, как упоминается в введении к этой главе о поставщиках/потребителях, в програмном обеспечении распространена необходимость сжимать и шифровать данные. В .NET Framework есть два декоратора для типа Stream, позволяющих решить эту задачу: уже упомянутый CryptoStream и класс GZipStream. Можно сжать и зашифровать входной файл следующим образом:

vzhik 04.08.11 в 13:45

Поток входного файла копируется в GZipStream, который оборачивает CryptoStream, который оборачивает выходной поток. Данные переходят из одного потока в другой, содержимое которых изменяется по пути.

vzhik 04.08.11 в 14:01

И сжатие и шифрование являются операциями, связанными с интенсивными вычислениями, поэтому их распараллеливание может принести пользу. Однако, учитывая специфику проблемы, мы не можем просто запустить параллельно сжатие и шифрование входных данных, поскольку шифрование работает с результатами сжатия. Зато мы можем построить подобие конвейера, где результат сжатия "скармливается" алгоритму шифрования, и пока шифруется N-й блок, подпрограмма сжатия переходит уже к обработке блока N+1, N+2 и так далее. Для упрощения мы реализуем такой конвейер с помощью еще одного декоратора - TransferStream. Идея этого декоратора состоит в том, что для записи выделен отдельный поток, который сохраняет во внутреннем Stream все, что пишется в TransferStream. Таким образом, когда код вызывает Write у TransferStream, последний не блокируется на время работы всей цепочки декораторов. Write только добавляет работу в очередь и немедленно завершается, и вызывающий код может продолжить работу. Простая реализация TransferStream (опирается на наследника базового абстрактного типа Stream, который просто реализует все абстрактные методы по умолчанию) приведена ниже с использованием Task и BlockingCollection:

vzhik 09.08.11 в 13:34

В конструкторе запоминается поток для записи данных. Затем настраиваются необходимые компоненты параллельного конвейера. Сначала создается BlockingCollection для хранения всех порций данных, предназначенных к записи. Затем запускается длительная задача Task, которая поледовательно выбирает из коллекции порции и записывает их в поток. Метод Write копирует предоставленные входные данные в новый массив, который добавляется в конец BlockingCollection. По умолчанию в BlockingCollection используется очередь с поддержкой семантики FIFO (первый вошел, первый вышел), то есть данные будут записываться во внутренний поток в том же порядке, что и добавляться в коллекцию. Это важная особенность при работе с потоками, применяющими скрытое упорядочивание. В конце, при закрытии потока у BlockingCollection выставляется признак о завершении добавления, что приводит к остановке запущенного в конструкторе Task цикла потребления, как только коллекция опустошается, и ожидается завершение Task. Таким образом гарантируется, что все данные будут записаны в внутренний поток до того, как он будет закрыт, и любое исключение, возникшее при обработке, будет корректно распространено далее.

vzhik 11.08.11 в 11:00

При наличии такого TransferStream, мы можем распараллелить предыдущий фрагмент кода, выполняющий сжатие/шифрование:

vzhik 15.08.11 в 06:40

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

vzhik 15.08.11 в 07:33

Преимущества такого подхода могут проявиться, когда I/O операции являются узким местом в задаче.

vzhik 15.08.11 в 10:12

IPRODUCERCONSUMERCOLLECTION{T}

vzhik 15.08.11 в 10:44

мы уже упоминали, что BlockingCollection{T} по умолчанию хранит данные в виде очереди, хотя и позволяет задавать другие структуры данных для хранения. Это достигается за счет нового интерфейса в .NET Framework 4, экземпляр реализации которого передается в конструктор BlockingCollection:

vzhik 15.08.11 в 10:48

Умело названный в честь паттерна, тип IProducerConsumerCollection представляет коллекцию для использования в реализациях поставщик/потребитель, когда данные добавляются в коллекцию поставщиком, а извлекаются потребителем. Поэтому двумя ключевыми методами интерфейса являются TryAdd и TryTake, которые оба должны быть реализованы потокобезопсным и атомарным образом.

vzhik 15.08.11 в 11:12

В .NET Framework 4 есть три конкретные реализации интерфейса: ConcurrentQueue{T}, ConcurrentStack{T} и ConcurrentBag{T}. ConcurrentQueue{T} - реализация интерфейса, которая по умолчанию используется в BlockingCollection{T}, подразумевая семантику "первый-пришел-первый-ушел" (FIFO). ConcurrentStack{T} предоставляет поведение "последний пришел-первый-ушел" (LIFO) , а ConcurrentBag{T} не дает никаких гарантий очередности, а ориентирован на повышение производительности в некоторых сценариях использования, например, когда поток работает и как поставщик и как потребитель.

vzhik 15.08.11 в 11:17

На основании IProducerConsumerCollection{T}, помимо BlockingCollection{T} могут строиться и другие структуры данных. Например, пул объектов - это простая структура данных, предназначенная для повторного использования объектов. Мы можем создать параллельный пул объектов с помощью привязки его к хранилищу конкретного типа или же реализовать его в рамках IProducerConsumerCollection{T}.

vzhik 15.08.11 в 11:51

Параметризуя хранилище таким способом, мы можем адаптировать наш ObjectPool{T} под конкретные сценарии использования с учетом сильных сторон подходящих для них реализаций коллекций. Например, для разработки UI приложения с динамичной графикой, мы возможно захотим заполнить графические буферы в фоновых потоках, а отразить их на пользовательском интерфейсе уже в UI-потоке. Учитывая возможный размер этих буферов, нам лучше организовать из них пул, чем постоянно выделять объекты огромных размеров, тем самым нагружая сборщик мусора. В таком сценарии ConcurrentQueue{T} подходящий выбор в качестве внутреннего хранилища. С другой стороны, если пул будет использоваться при параллельном выделении памяти для кеширования объектов различных размеров, мне не нужна FIFO-шность ConcurrentQueue{T}, а скорее структура данных, в которой реализована минимальная синхронизации между потоками (идеально подходит ConcurrentBag{T}). Внутри ConcurrentBag{T} использует по списку экземпляров T на каждый поток. Каждый обращающийся к мешку поток может добавлять и удалять данные относительно независимо от остальных потоков, которые также работают с мешком. Только когда поток пытается выбрать элемент и его собственный список пуст, он ищет в списках других потоков (реализация делает локальные списки потоков видимыми для других потоков только для этой цели). Это кажется весьма знакомым: ConcurrentBag{T} реализует шаблон аналогично алгоритму пула потоков .NET Framework 4, который "ворует" работу. Если доступ к локальному списку относительно дешевый, то захват элементов из списка другого потока достаточно дорог. Поэтому ConcurrentBag{T} лучше всего подходит для сценариев, где каждому потоку большую часть времени нужно работать с собственным списком. Для того, чтобы добиться этого в примере с объектным пулом, можно попробовать каждому потоку при инициализации заполнить пул несколькими объектами, так, чтобы впоследствии, при взятии и добавлении объектов, поток работал преимущественно со своей очередью.

vzhik 16.08.11 в 10:14

ПОСТАВЩИК / ПОТРЕБИТЕЛЬ ВЕЗДЕ

vzhik 23.08.11 в 11:34

Если Вы писали приложения под Windows, высока вероятность того, что Вы уже использовали паттерн поставщик/потребитель, возможно, даже не понимая этого. У шаблона поставщик/потребитель есть множество известных реализаций.

vzhik 23.08.11 в 11:39

ПУЛ ПОТОКОВ

vzhik 23.08.11 в 11:40

Если Вы использовали пул потоков, значит использовали главнейшую реализацию паттерна поставщик/потребитель. Пул потоков обычно проектируется вокруг структур данных, содержащих работу, которую необходимо выполнить. Каждый поток из пула следит за этой структурой данных, ожидая поступления новой работы. Когда это происходит, поток извлекает и выполняет ее, после чего возвращается в состояние ожидания. С такой точки зрения поставляемая работа потребляется и выполняется потоками из пула. Используя уже известный нам тип BlockingCollection{T} нетрудно построить простой, без всяких излишеств, пул потоков:

vzhik 23.08.11 в 12:23

В сущности, это очень похоже на работу пула потоков из .NET Framework 3.5 и более ранних версий. В .NET Framework 4 структура данных для хранения выполняемых рабочих элементов стала более распределенной. Вместо поддержки одной глобальной очереди, как в нашем примере выше, ThreadPool в .NET Framework 4 поддерживает не только глобальную очередь, но и по очереди на каждый поток. Работа, поступающая в пул извне, добавляется в глобальную очередь, как и раньше, но потоки в пуле могут помещать создаваемую ими работу в локальные очереди потоков, а не в глобальную очередь. Когда поток начинает искать работу для выполнения, он сначала проверяет свою локальную очередь, а при отсутствии элементов в ней ищет в глобальной. Если и глобальная очередь пуста, поток проверяет очереди своих коллег, "воруя" работу у других, чтобы не простаивать самому. Подход с кражей работы обладает преимуществами, среди которых можно упомянуть минимальную конкуренцию и синхронизацию между потоками (при идеальной нагрузке, потоки большую часть времени будут работать с собственными очередями), а также эффективное использование кеша. (Вы можете добиться похожего поведения в SimpleThreadPool, передав экземпляр ConcurrentBag{Action} в качестве внутреннего хранилища в BlockingCollection{Action}, вместо использования стандартного ConcurrentQueue{Action}.)

vzhik 23.08.11 в 13:56

В предыдущем абзаце мы сказали, что "потоки могут помещать создаваемую ими работу в свои локальные очереди", но это совершенно не обязательно. Вообще-то, метод ThreadPool.QueueUserWorkItem не использует возможность отбора работы. Эта функциональность доступна и включена по умолчанию только при работе с Task. Такое поведение можно отключить в самом Task, используя TaskCreationOptions.PreferFairness.

vzhik 24.08.11 в 08:03

По умолчанию при выполнении Task в пуле потоков используются эти самые внутренние очереди, поддерживающие "кражу работы". Впрочем, такое поведение не привязано жестко к Task, а вынесено в планировщик задач TaskScheduler. Task выполняется в планировщике задач, версия которого для .NET Framework 4 направляет задачу в ThreadPool. Именно такой экземпляр Вы получите при обращении к свойству TaskScheduler.Default, и, как подразумевается из названия, это планировщик, который используется по умолчанию объектом Task. Как и с другими вещами, к которым мы применяем термин "по умолчанию", обычно существует способ для подмены умолчания. В случае с выполнением задач такой способ есть. Вы можете написать собственную реализацию TaskScheduler для выполнения задач таким образом, какого требует Ваше приложение.

vzhik 06.10.11 в 11:27

Планировщик задач сам по себе воплощает концепцию поставщик-потребитель. Являясь абстрактным классом он поставляет несколько абстрактных методов, которые можно переопределить и несколько методов, которые нужно переопределить. Главный метод называется QueueTask и используется инфраструктурой NET Framework повсеместно, которая выступает в роли поставщика, для добавления задач в планировщик. Реализация планировщика (потребитель) выполняет эти задачи любым удобным ей способом. Мы можем создать очень простой TaskScheduler, на основе ранее написанного SimpleThreadPool, который просто передает вызов из QueueTask в QueueWorkItem с делегатом, выполняющщим задачу:

vzhik 06.10.11 в 13:07

Теперь мы можем создавать задачи и выполнять их в нашем планировщике:

vzhik 06.10.11 в 13:12

В класс TaskFactory, экземпляр которого по умолчанию возвращается при обращении к свойству Task.Factory, также можно параметризовать экземпляром TaskScheduler. В дальнейшем это позволит использовать методы фабрики, которые будут работать с нашим планировщиком:

vzhik 06.10.11 в 14:29

Если Вам приходилось писать быстро реагирующие приложения для Windows, значит Вы уже пользовались шаблоном поставщик/потребитель. Как в Windows Forms, так и в Windows Presentation Foundation (WPF) доступ к элементам управления пользовательского интерфейса должен выполняться только из потока, в котором они были созданы (разновидность родства потоков). Это проблематично по нескольким причинам, очевиднейшей из которых является необходимость отклика пользовательского интерфейса. При написании реагирующего приложения обычно требуется выносить работу из UI-потока в фоновый поток, чтобы UI-поток мог продолжить обработку сообщений операционной системы, которые заставляют пользовательский интерфейс перерисовываться, реагировать на мышку и тому подобное. Такая обработка выполняется в коде, относящемуся к циклу сообщений Windows. Хотя работа выполняется в фоне, зачастую необходимо обновлять пользовательский интерфейс, что обычно приводит к манипуляциям с элементами управления, которые были созданы в UI-потоке. А следовательно фоновый поток должен размещать обращение к таким элементам управления в UI-потоке.

vzhik 07.10.11 в 10:19

И Windows Forms, и WPF предоставляют поддержку такой возможности. Windows Forms содержит экземплярный метод Invoke класса Control. Этот метод принимает делегат и направляет его выполнение в поток, с которым связан данный экземпляр Control. Следующее приложение Windows каждую секунду обновляет текстовую метку в UI-потоке:

vzhik 07.10.11 в 12:21

Вызов Invoke является синхронным, то есть он не завершится, пока не выполнится делегат. Также существует асинхронный способ выполнения делегата с помощью метода BeginInvoke.

vzhik 07.10.11 в 12:24

Такой механизм сам по себе является реализацией паттерна поставщик/потребитель. Windows Forms обслуживает очередь делегатов, которые должны быть обработаны UI-потоком. При вызове Invoke или BeginInvoke новый делегат добавляется в очередь, и посылается сообщение Windows в UI-поток. Цикл сообщений UI-потока рано или поздно обработает это сообщение? тем самым сообщая, о том, что делеат нужно извлеть из очереди и выполнить. Согласно этой логике вызов Invoke или BeginInvoke - это поставщик, UI-поток - потребитель, а в роли данных выступает делегат.

vzhik 07.10.11 в 12:29

Разновидность шаблона поставщик/потребитель, применяемая в Invoke, имеет специальное название - "рандеву", которое обычно используется для обозначения ситуации, когда много потоков встречаются для обоюдного обмена информацией. Тот, кто вызывает Invoke, предоставляет делегат и потенциально ожидает результата его выполнения. UI-поток получает делегат и, если необходимо, передает результат его выполнения. Ни один из потоков не продолжится, пока рандеву не достигнет точки, когда обмен данными полностью завершен.

vzhik 07.10.11 в 13:37

Аналогичный механизм поставщика/потребителя доступен и в WPF через класс Dispatcher, который выставляет cхожие методы Invoke и BeginInvoke. Чтобы отделить эту возможность и облегчить написание компонентов, которым необходимо направлять вызовы в UI и работать в разных окружениях UI, .NET Framework предоставляет класс SynchronizationContext. У есть SynchronizationContext есть методы Send и Post, которые транслируются в Invoke и BeginInvoke соответственно. Для Windows Forms есть внутренний наследник SynchronizationContext по имени WindowsFormsSynchronizationContext, который переопределяет Send для того, чтобы позвать Invoke и Post - для BeginInvoke. В WPF есть аналогичный тип. Таким образом можно писать библиотеки в терминах SynchronizationContext, а потом предоставлять правильный SynchronziationContext во время исполнения, чтобы обеспечить возможность обращения к UI-потоку в запущенном окружении.

vzhik 10.10.11 в 11:38

Тип SynchronizationContext можно использовать и для других целей, фактически в .NET Framework есть и другие его реализации, не заточенные для использования в UI. Однако в этой книге при упоминании SynchronizationContext мы подразумеваем обращение к UI.

vzhik 12.10.11 в 07:30

Для облегчения жизни программистов существует статическое свойство SynchronizationContext.Current, с помощью которого можно захватить ссылку на контекст синхронизации и в дальнейшем использовать ее для направления вызовов в текущий поток. И Windows Forms, и WPF инициализируют это свойство в UI-потоке соответствующим экземпляром SynchronizationContext. А код может получить значение этого свойства и использовать для направления задач в UI. В качестве тренировки можно переписать предыдущий пример, заменив явный вызов Control.Invoke на SynchronizationContext.Send:

vzhik 13.10.11 в 10:58

Уже упоминалось, что можно создать собственный TaskScheduler, чтобы реализовать расширенную логику обработки поставляемых объектов Task. Вместе со стандартным планировщиком задач, который работает с внутренними "отбирающими работу" очередями из пула потоков ThreadPool, .NET Framework 4 также содержит метод TaskScheduler.FromCurrentSynchronizationContext, который генерирует планировщик задач, работающий с текущим контекстом синхронизации. Воспользуемся этим для дальнейшего улучшения нашего примера:

vzhik 13.10.11 в 11:08

Возможность выполнять задачи в различных контекстах синхронизации отлично совмещается с продолжениями и потоками данных. Например:

vzhik 13.10.11 в 11:18