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

Использование общих переменных добавляет новые трудности в обеспечение синхронизации потоков [13]. В рамках синхронизации потоков вводится понятие критической секции - последовательности инструкций, имеющих доступ к разделяемым данным, которые в случае одновременного исполнения двумя различными потоками приведут к гонке данных (race condition), когда несколько потоков одновременно пытаются обновить и использовать общие данные, что может привести к неопределенному состоянию этих данных. Такое поведение нарушает правильность исполнения программы и требует использования примитивов синхронизации программистом, для защиты критических секций от одновременного исполнения несколькими потоками. Обычно говорят о занятии и освобождении критической секции потоком. Реализуемая программистом синхронизация должна удовлетворять следующим свойствам:

Взаимное исключение (mutual exclusion). В любой момент времени критическая секция исполняется только одним потоком.

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

Отсутствие задержек. Поток немедленно входит в освобожденную критическую секцию.

Возможность входа. Поток, ожидающий входа в критическую секцию, когда-нибудь сможет это осуществить.

При корректной синхронизации обеспечивается потокобезопасность (thread safety) исполняемого кода, что подразумевает возможность одновременного его исполнения несколькими потоками без нарушения свойств, рассмотренных выше. Синхронизация потоков осуществляется с помощью использования программистом примитивов синхронизации [4].

Семафоры. Семафор представляет собой объект-счетчик, инициализируемый числом при его создании, к которому можно применить операции access() и release(). Операция access() проверяет значение счетчика на равенство нулю, после чего уменьшает значение счетчика на единицу, если он не равен нулю, позволяя занять потоку критическую секцию. В ином случае, поток блокируется, ожидая, пока один из потоков, исполняющих критическую секцию, не вызовет функцию release(), которая увеличит значение счетчика на единицу. Критическая секция выступает здесь как набор инструкций, одновременное исполнение которых доступно ограниченному числу потоков. Операции access() и release() являются атомарными, что означает, что до момента их завершения (или блокировки вызывающего их потока в случае вызова access()) никакой другой поток не может получить доступ к семафору. Как видно, семафоры предоставляют довольно низкоуровневый интерфейс синхронизации потоков, операция access() обязана использоваться в паре с операцией release(), о чем может забыть программист при реализации сложной логики программы. Частным случаем семафоров являются мьютексы (mutex), в которых счетчик может принимать только значения 1 и 0.

Мониторы. Мониторы являются высокоуровневым примитивом синхронизации потоков, и оборачивают функционал, предоставляемый семафором, для более безопасного обеспечения взаимоисключения выполнения критических секций. Монитор представляется объектом, доступ к данным которого можно получить с помощью специальных процедур, связанных с монитором. Каждая процедура может выполняться в один момент времени только одним потоком. Другие вызывающие процедуру потоки блокируются до завершения исполнения процедуры первым потоком, а после в порядке очереди выполняют её.

Условные переменные. Отдельно стоит выделить такой примитив синхронизации, как условные переменные, позволяющие реализовать условную блокировку процессов. К условным переменным применимы две процедуры - wait() и signal(). Когда потоку для дальнейшего исполнение требуется выполнение некоторого условия, он вызывает процедуру wait(), которая освобождает связанный с условной переменной мьютекс и блокирует поток. Другой поток выполняет действия, для осуществления условия, необходимого для продолжения исполнения первого потока и вызывает процедуру условной переменной signal(), сигнализируя первому потоку о выполнении условия. Первый поток выходит из состояния ожидания, блокирует связанный с условной переменной мьютекс и продолжает исполнять код до следующего вызова процедуры wait() , если он необходим с точки зрения логики программы. Использование условных переменных решают проблему активного ожидания (спинлоков), когда потоку необходимо постоянно проверять выполнение некоторого условия, считывая общую переменную, изменение значения которой сигнализирует о выполнении условия.

Барьеры. Каждый из потоков в некоторой точке своего исполнения, где необходима синхронизация c другими потоками, обращается к объекту-барьеру, который блокирует его дальнейшее выполнение до момента, пока остальные потоки не обратятся к объекту-барьеру. Далее с каждого из потоков снимается блокировка, и они продолжают выполнение. Такой механизм синхронизации может быть полезен при реализации распараллеливания по данным, когда необходимо в конце работы каждого из потоков выполнить общую операцию над порциями разделенных по потокам данных.

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

2.9 Обзор основных средств реализации многопоточности

В качестве доступного инструментария были рассмотрены две открытые кроссплатформенные библиотеки для разработки C++ приложений wxWidgets и Boost, поскольку выбор инструментальных средств ограничен требованием к решению задачи с учетом минимизации использования сторонних средств. Одной из главных особенностей библиотеки GridMD является обеспечение высокой портируемости использующих ее приложений. Обычно проведение расчета на кластере начинается с копирования исходного текста программы на головную машину кластера с его дальнейшей компиляцией установленным на головной машине компилятором и запуском скомпилированных исполняемых файлов. В описанной ситуации обеспечение портируемости означает что, во-первых, код используемых библиотек должен компилироваться широким классом компиляторов, зачастую устаревших. Во-вторых, компиляция приложения должна быть возможна под широкий класс платформ, которые сторонние библиотеки обязаны поддерживать. В-третьих, должна обеспечиваться быстрая настройка среды компиляции и исполнения расчетного приложения на кластере в виде установки необходимых объектных и заголовочных файлов библиотек на каждую из машин кластера, а значит необходимо использовать минимум сторонних зависимостей необходимых для компиляции и исполнения приложения. Таким образом, выбор инструментальных средств ограничен уже используемыми средствами разработки в GridMD. Несмотря на указанное ограничение, библиотеки Boost и wxWidgets поддерживают широкий функционал для работы с многопоточностью. Ниже будет приведен обзор средств работы с потоками в каждой из библиотек, после чего сделан выбор одной из них в качестве главного средства решения задачи.

wxWidgets. Библиотека wxWidgets является кроссплатформенной библиотекой инструментов для разработки приложений c графическим интерфейсом пользователя на языке C++ с открытым исходным кодом [14]. Библиотека wxWidgets является объектно-ориентированной библиотекой, которая предоставляет набор классов для работы с большим множеством компонентов, составляющих современные приложения, таких как элемент графического интерфейса пользователя, файловыми системами, базами данных, объектами мультимедиа и в частности средствами организации многопоточности. Программист на базе единого API имеет возможность писать и компилировать программы множеством различных компиляторов под множество платформ, таких как Microsoft Windows, Unix-подобные системы, Apple Macintosh и других. Отличительными особенностями является либеральность лицензии wxWidgets license под которой распространятся библиотека, допускающая линковку с несвободными фрагментами кода, что позволяет использовать ее в закрытых коммерческих проектах.

Абстракция потока в wxWidgets представлена в виде класса wxThread. Класс wxThread является абстрактным, поэтому невозможно создать объект этого класса. В wxWidgets принята следующая концепция - для создания объекта потока необходимо унаследовать класс wxThread и переопределить чисто виртуальный метод wxThread::Entry(), наполнив его кодом, который будет исполняться в созданном потоке после его запуска. После создания объекта наследника класса wxThread необходимо последовательно вызвать метод инициализации потока wxThread::Create(), передав при необходимости размер стека, выделяемого потоку, и метод запуска потока wxThread::Run(), который неявно запускает переопределенный в дочернем классе метод wxThread::Entry().

В wxWidgets определено два типа объектов потоков в зависимости от механизма управления системными ресурсами - detached и joinable. Detached объекты потоков самостоятельно освобождают все связанные с ними ресурсы после завершения функции wxThread::Entry(). После выхода из функции wxThread::Entry() автоматически вызывается деструктор объекта, в котором связанный с объектом поток будет остановлен и освобождена выделенная под объект память. Detached объекты потоков должны быть выделены в динамической памяти. Фактически, это означает, что программисту не требуется хранить указатель на detached объект потока, ведь очисткой ресурсов займется сам объект при завершении. Joinable объекты потоков для освобождения ресурсов требуют явного ожидания завершения функции wxThread::Entry() через вызов функции wxThread::Wait(), которая заблокирует вызывающий ее поток до завершения потока, связанного с объектом, для которого был вызван этот метод. Кардинальное отличие двух представленных типов объектов потоков в том, что методы wxThread для joinable объектов вызывать безопасно в течение всего времени жизни объекта, в отличие от detached объектов, чье время жизни не определено, поскольку оно зависит только от продолжительности исполнения функции wxThread::Entry(). Завершение потока joinable объекта можно ожидать в синхронном режиме путем вызова функции wxThread::Wait(), так же, как и проверять состояние его исполнения с помощью методов wxThread::isAlive(), wxThread::isPaused() и wxThread::isRunning(). Явно завершить выполнение потока joinable объекта можно с помощью метода wxThread::Kill().

В качестве примитивов синхронизации библиотека содержит реализации базовых примитивов семафор wxSemaphore, мьютекс wxMutex и условная переменная wxConditional описанных в предыдущей главе. Поддерживаются рекурсивные мьютексы, которые не блокируют вызывающий поток при многократном блокировании мьютекса одним и тем же потоком. Такой тип мьютекса нельзя использовать в качестве ассоциативного мьютекса для условной переменной.

Функциональность мониторов в библиотеке реализуется классами wxMutexLocker (Листинг 3). В конструкторе объекта wxMutexLocker блокируется мьютекс, который передается в конструктор в качестве параметра. В деструкторе объекта происходит разблокировка связанного с ним мьютекса. Таким образом, объект класса wxMutexLocker обеспечивает взаимоисключающий доступ к критической секции с момента своего создания до момента уничтожения. Работа класса wxMutexLocker является примером идиомы объектно-ориентированного программирования получение ресурса есть инициализация (RAII), которая позволяет писать безопасный при исключениях код. Объекты классов, реализующих RAII, должны выделяться в стеке в качестве автоматических переменных, тогда при возникновении исключения произойдет раскрутка стека, представляющая из себя вызов деструкторов всех автоматических переменных, выделенных в функции, породившей исключение. При явном использовании wxMutex в случае порождения исключения может быть не вызван его метод wxMutex::Unlock(), что приведет к перманентной блокировке критической секции, которую этот мьютекс защищает. Документация wxWidgets рекомендует отдавать предпочтение использованию wxMutexLocker для защиты критических секций.

Boost. Проект Boost выступает как набор кроссплатформенных библиотек, позволяющих программисту решать широкий спектр задач, возникающих в его повседневной деятельности [7]. Boost включает в себя инструменты как общего назначения, такие как умные указатели, структуры данных и графы, алгоритмы, так и абстракции для работы со средствами операционной системы, многопоточностью, геометрическими данными, интернационализацией и прочим. Концептуально Boost является расширением стандартной библиотеки STL языка С++ и некоей экспериментальной площадкой, из которой успешные и проверенные программные решения переходят в стандартную библиотеку по решению комитета по стандартизации C++. В реализации Boost сделан упор на обобщенное программирование, широкое использование шаблонов, эффективность и переносимость. Многие библиотеки распространяются в виде заголовочных файлов шаблонных классов, не требующих предварительной сборки и линковки с приложением пользователя, в отличие от отдельно скомпилированных библиотек объектных файлов. В частности, библиотека заголовочных файлов Boost.Graph используется в GridMD для реализации концепции графа исполнения, и не требует отдельной объектной библиотеки для работы скомпилированного пользователем GridMD приложения, обеспечивая высокую переносимость GridMD приложений, о чем было сказано ранее. Большинство библиотек распространяется под лицензией Boost Software License, разрешающей использование Boost как в открытых, так и в коммерческих проектах с закрытым кодом.

В составе Boost имеется библиотека Boost.Thread для работы с многопоточностью [15]. Класс boost::thread представляет абстракцию потока. Для запуска потока необходимо создать объект класса boost::thread, передавая в конструктор в качестве параметра объект вызываемого типа, у которого определен operator(), или указатель на функцию, которая будет выполнена в потоке. После создание объекта поток начинает немедленно выполняться.

В случае выхода объекта потока из области видимости, или явного удаления оператором delete при выделении объекта в динамической памяти поток продолжает выполняться вплоть до завершения функции, переданной в конструктор. Для синхронного ожидания завершения потока необходимо вызвать функцию boost::thread::join() или воспользоваться классом boost::scoped_thread, объект которого при создании инициализируется объектом потока и при необходимости объектом определенного пользователем вызываемого типа. При выходе объекта boost::scoped_thread из области видимости по умолчанию гарантируется вызов join() для связанного с ним потока или определенного пользователем operator().

Механизм прерывания потока реализован через вызов функции boost::thread::interrupt(). Как правило, прерывание потока происходит отложено - при достижении определенных точек прерывания поток проверяет, не была ли вызвана функция interrupt(), и если была, то генерирует исключение boost::thread_interrupted, которая может быть обработана в функции, исполняемой потоком.

В месте вызова boost::this_thread::sleep_for будет проверено существование запроса на прерывание и сгенерировано исключение. В общем случае ловить и обрабатывать исключение не обязательно, тогда будет выполнена стандартная схема работы программы при появлении необработанного пользователем исключения в функции - раскрутка стека и выход из потоковой функции. Такое поведение характерно только в контексте прерванного потока, на остальные потоки влияние необработанного исключения оказано не будет. В качестве точек прерывания в библиотеке выделяется места вызова 16 различных функций, которые более подробно описаны в документации, так же поток может быть прерван, если он находится в состоянии ожидания входа в критическую секцию. Если имеется необходимость защитить область потоковой функции от нежелательных проверок на запрос прерывания, то необходимо создать объект класса boost::this_thread::disable_interruption, в область видимости которого будет отменена проверка запросов на прерывания.

Boost.Thread имеет реализации всех традиционных примитивов синхронизации, описанных ранее, и поддерживает идиому RAII в виде реализации класса boost::lock_guard. Кроме того, в Boost поддерживаются концепции исключительных и неисключительных блокировок, в виде классов boost::unique_lock и boost::shared_lock соответственно, позволяющих реализовать схему multiple reader / single write.

В приведенном примере потокам, печатающего содержимое массива и суммирующего его элементы, необходим лишь доступ на чтение к массиву. Так как потоки, производящие чтение разделяемого ресурса, не влияют на работу друг друга, возможен неисключительный одновременный доступ к ресурсу. Это реализуется с помощью неисключительной блокировки boost::shared_lock специального типа мьютекса boost::shared_lock. Для потока, заполняющего массив данных, необходим исключительный доступ, так как он изменяет разделяемый ресурс. С помощью boost::unique_lock поток приобретает исключительный доступ, сначала дожидаясь завершения доступа каждого из читающих ресурс потоков и блокируя их до завершения изменения ресурса в случае их повторного обращения к ресурсу.

2.10 Выбор библиотеки для решения задачи

Библиотека Boost в целом имеет более широкие возможности по организации многопоточности, чем wxWidgets, поддерживая более сложные примитивы синхронизации и абстракции для работы с потоками. Однако библиотека Boost.Thread является отдельно скомпилированной объектной библиотекой, требующей своего распространения вместе с приложением. Более того, необходимо распространять объектные библиотеки Boost.System, Boost.Chrono и Boost.DateTime вместе с приложением, использующим Boost.Thread. В данный момент GridMD не использует объектные библиотеки Boost, но использует объектный модуль wxBase библиотеки wxWidgets для кроссплатформенной работы с файлами, директориями, строками и прочим, в состав который, в частности, входят и классы для работы с потоками. Для сохранения широкой переносимости и легковесности библиотеки GridMD было решено использовать библиотеку wxWidgets в качестве инструмента для распараллеливания выполнения локальных узлов, поскольку она предоставляет весь базовый функционал работы с потоками, необходимый для решения задачи.

3. Создание очереди заданий

Для организации работы потоков был выбран паттерн проектирования пул потоков (Thread pool) [16] . Пул потоков является объектом, которому возможна выдача некоего множества задач на исполнение. Внутри пул потоков представляет собой ограниченный набор инициализированных потоков, которым пул раздает задачи на исполнение из очереди задач пула по мере ее заполнения в соответствии с конкретной схемой планирования назначения потоков на задачи. Поток в пуле может находиться в двух состояниях - ожидания получения задачи, когда поток спит, не занимая процессорное время, и состоянии непосредственного исполнения задачи. Задачи пулом по мере получения собираются в очереди задач и могут быть получены в любой момент существования пула. Пул гарантирует, что в определенный момент времени вновь пришедшая задача будет выполнена, когда до нее дойдет очередь.

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

Использование пула потоков является оптимальным решением для распараллеливания выполнения локальных узлов в GridMD. Помимо преимуществ, указанных выше, использование пула в качестве отдельного модульного элемента позволяет инкапсулировать функционал по организации работы с задачами, генерируемыми менеджером сценариев, таким как ожидание завершения выполнения задачи, опрос статуса исполнения и получение результата задачи, в рамки одного компонента. Такой подход в реализации библиотеки виден в выделении отдельных компонентов, как менеджер сценариев и менеджер заданий. Для координирования исполнения задач в рамках потоков необходимо реализовывать отдельный менеджер, хорошим кандидатом на место которого выступает пул потоков (Рис. 8).

Рис. 8 Пул потоков в контексте компонентов GridMD

Взаимодействие с пулом происходит через его интерфейсные функции и состоит из отправки задачи на исполнение в очередь пула, опроса статуса исполнения задачи, ожидания завершения выполнения задачи с получением ее результата и отмены исполнения в случае необходимости. Логически, пул потоков является вычислительным ресурсом для локальных задач, генерируемых менеджером сценариев. С точки зрения реализации пул не оформлен в виде ресурса, как это сделано с командным интерпретатором локальной операционной системы или удаленной системой очередей и нет необходимости указывать его в списке ресурсов в файле resources.xml. Вместо этого, пользователь в секции настройки численного эксперимента при необходимости указывает, что локальные узлы могут быть выполнены с помощью пула, передавая значение перечислимого типа gmEXE_THREADS в метод gmExperiment::set_execution().

Пул потоков реализован в GridMD в виде класса gmThreadPool и в своей реализации имеет два основных компонента - класс рабочего потока gmThread и класс исполняемой задачи gmTask. При создании пулу потоков передается необходимое число рабочих потоков, которое по умолчанию равно числу ядер процессора, умноженному на два.

Рассмотрим более подробно интерфейс класса gmThreadPool:

· Конструктор gmThreadPool (size_t threads=wxThread::GetCPUCount()-1) инициализирует threads рабочих потоков на исполнение в режим ожидания поступления новых задач в очередь задач. Параметром по умолчанию является оптимальное число потоков для данной аппаратной конфигурации, минус один, поскольку работа самого пула потоков выполняется в отдельном потоке, необходимость чего будет рассмотрена ниже.

· ~gmThreadPool (). Деструктор пула потоков. Вызывается при выходе объекта пула потоков из области видимости, или явного удаления оператором delete в случае выделения пула в динамической памяти. Немедленно останавливает все рабочие потоки, прерывая выполнение их текущих задач, освобождает связанные с потоками системные ресурсы и удаляет все оставшиеся задачи из очереди задач.

· gmTaskID CreateGMMainTask (int (*gridmd_main)(int, char*[]), int argc, char* argv[]). Метод отправки пулу задачи по исполнению функции int ((*gridmd_main)(int, char*[]), int argc). Пул потоков работает с абстракциям задач реализованных в виде класса gmTask и его наследников. Метод создает задачу в виде объекта класса gmMainTask, наследника класса gmTask, кладет задачу в очередь задач и устанавливает ей статус «принято на исполнение». Возвращает уникальный идентификатор принятой задачи gmTaskID, под которым задача регистрируется в пуле в реестре идентификаторов, и который возможно передать в другие интерфейсные функции пула для работы с конкретной задачей.

· gmTASK_STATUS TaskStatus (gmTaskID taskID) const. Метод возвращает статус отправленной пулу задачи в виде перечислимого типа gmTASK_STATUS по уникальному идентификатору задачи taskID. Поддерживаются следующие состояния задачи:

? gmTASK_POOLED - задача принята на исполнение, в текущий момент ожидает исполнения в очереди задач.

? gmTASK_PROCESSED - задача исполняется рабочим потоком.

? gmTASK_FINISED - задача успешно завершена, возможно получение ее возвращаемого значения с помощью метода int TaskResult (gmTaskID taskID).

? gmINVALID_STATUS - Задача с идентификатором taskID не зарегистрирована в пуле. Такое возможно, если идентификатор taskID был получен не с помощью одного из методов пула по созданию задач, или задача была явно удалена из пула после вызова других методов, рассмотренных ниже.

· int TaskResult (gmTaskID taskID). Метод получения возвращаемого значения задачи. При вызове ожидает завершение исполнения задачи, а именно до момента, когда ее статус станет равен gmTASK_FINISHED. Ожидание завершения задачи происходит в синхронном режиме, блокируя вызывающий метод поток. Возвращает число, сигнализирующее об успешности выполнения задачи, в соответствии с логикой, заложенной пользователем (например, при выполнении int (*gridmd_main)(int, char*[], int argc) ее возвращаемое значение). Удаляет задачу из пула потоков вместе с ее идентификатором из реестра идентификаторов, делая идентификатор taskID невалидным для остальных интерфейсных функций.

· void RemoveTask (gmTaskID taskID). Метод удаления задачи с идентификатором taskID. В зависимости от статуса задачи удаляет ее из очереди задач или немедленно останавливает ее исполнение, удаляя связанный с ней объект типа gmTask и ее идентификатор из реестра идентификаторов задач пула, делая идентификатор taskID невалидным для остальных интерфейсных функций.

· bool IsValidIndex(gmTaskID taskID) const. Метод проверки, зарегистрирована ли задача с идентификатором taskID в реестре идентификаторов.

· void RegisterRedirector(gmRedirectorBase* redirector). В пуле потоков для получения уникальной копии конкретного объекта для каждой из исполняемых задач была реализована концепция редиректоров. Редиректор представляется в виде шаблонного класса gmRedirector, который параметризируется типом хранимого объекта и хранит уникальную копию объекта для каждой задачи. Копию объекта можно получить с помощью метода gmRedirector<T>::GetObject(). Для использования редиректора его необходимо зарегистрировать с помощью рассматриваемого метода пула, передав указатель на базовой класс редиректора gmRedirectorBase*. Использование базового класса позволяет хранить уникальные объекты разного типа для каждой из задач.

Рассмотрим отдельные компоненты и некоторые детали реализации пула потоков.

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

Рис. 9 Диаграмма классов реализации пула потоков

В данный момент реализован класс gmMainTask для исполнения задачи, генерируемой менеджером сценариев, при неявном определении действий, а именно для исполнения функции gridmd_main(). В будущем планируется реализовать поддержку исполнения задач в пуле при явном определении действий с помощью класса gmNodeAction и при определении действий в виде скриптов.

Рис. 10 Диаграмма состояний объекта класса gmTask

На Рис. 10 представлена диаграмма состояний класса gmMainTask. Жизненный цикл задачи, исполняемой пулом, начинается с создания объекта gmTask и отправки задачи в очередь заданий пула с помощью метода gmThreadPool::GreateGMMainTask(). Задача переходит в состояние gmTASK_POOLED, при котором она ожидает своей очереди на исполнение. Когда задача становится в начале очереди, ее забирает первый освободившийся поток и вызовом метода gmThread::StartTask() начинает исполнение задачи, вызывая метод gmMainTask::Run() и устанавливая ее в состояние gmTASK_PROCESSED. После завершения метода gmMainTask::Run() исполнение задачи завершено, поток устанавливает состояние задачи в gmTASK_FINISHED и забирает новую задачу в случае, если очередь задач не пуста. В состоянии gmTASK_FINISED вызов метода gmThreadPool::TaskResult() немедленно возвращает число, сигнализирующее об успешности выполнения задачи, в соответствии с логикой, заложенной пользователем, и удаляет объект gmMainTask вместе с ее уникальным идентификатором из реестра идентификаторов пула. Пул не хранит информацию о выполненной и опрошенной о результатах задаче, и метод опроса статуса gmThreadPool::TaskStatus() по такому идентификатору вернет gmINVALID_STATUS.

В состояниях gmTASK_POOLED, gmTASK_PROCESSED, gmTASK_FINISHED возможно явно отменить выполнение задачи и уничтожить связанный с ней объект методом gmThreadPool::RemoveTask(), который установит состояние задачи в gmINVALID_STATUS. Реализация механизма удаления состоит либо в удалении задачи из очереди в случае, если она находится в статусе gmTASK_POOLED, или удалении связанного с ней потока при нахождении задачи в состоянии gmTASK_PROCESSED. Не существует другого способа отцепить задачу от рабочего потока пула, кроме явного удаления потока с освобождением связанных с ним ресурсов и его переинициализации. Но для восстановления исходного количества рабочих потоков после удаления задачи в классе пула потоков необходим механизм, который будет производить мониторинг текущего состояния потоков пула, получать сигнал об удалении задачи и инициализировать новый поток вместо удаленного. Именно поэтому класс gmThreadPool, как и его рабочие потоки gmThread, является наследником класса wxThread и переопределяет метод wxThread::Entry(), который работает в отдельном потоке, ожидая сообщения об удалении рабочего потока и инициализируя новый вместо удаленного.

В приведена реализация восстановления количества рабочих потоков при удалении одного из них. Метод gmThreadPool:: Entry() ожидает сигнала об удалении потока с помощью условной переменной mThreadsNotifier и ее метода Wait(). В деструкторе gmThread происходит вызов mThreadNotifier.Signal(), который будит ожидающий служебный поток пула, и проверяет условие равентва текущего количества потоков переданному пользователем при создании пула. Если они не равны, то служебный поток блокирует связанный с массивом потоков мьютекс, чтобы предотвравить конкуретный доступ к ней (при удалении еще одной задачи деструктор другого потока обратиться к масссиву для удаления себя из него), инициализирует необходимое количество потоков и добавляет их в массив.

При отсутствии в очереди задач каждый из рабочих потоков gmThread в переопределенном методе gmThread::Entry() ожидает сигнала о прибытии в очередь новой задачи с помощью условной переменной mQueueNotifier, входящей в состав объекта gmThreadPool, и ее метода Wait(). В методе gmThreadPool::SubmitTask(gmTask *task) задача кладется в очередь, о чем сигнализируется одному из ожидающих потоков методом mQueueNotifier.Wait(). Кто именно из ожидающих потоков получит сообщение о получении новой задачи не определено и зависит от планировщика операционной системы, и в данной реализации это не имеет важности. После получения сообщения поток просыпается, блокирует связанный с очередью задач мьютекс mQueueMutex для получения эксклюзивного доступа к очереди задач и начинает выполнение задачи вызовом метода gmThread::StartTask(). После завершения выполнения задачи поток пытается взять новую задачу из очереди. В этом случае, если очередь не пуста, распределение задач по потокам определяется случайным образом и зависит от последовательности получения каждым из них исключительного доступа к критической секции путем блокировки мьютекса mQueueMutex. Поток, занявший критическую секцию, получает первую задачу из очереди.

Важной особенностью представленной реализации пула потоков является его модульность и высокая степень сокрытия реализации. Пул потоков самостоятельно управляет ресурсами, такими как рабочие потоки и задачи, выделяя их по внешнему запросу и освобождая по мере использования. Внешние компоненты, взаимодействующие с пулом (например, менеджер сценариев), не имеют доступа к объектам рабочих потоков gmThread и задач gmTask, им лишь выдаются идентификаторы для работы с ресурсами через интерфейс пула. Поддерживаемый набор типов задач так же является расширяемым через наследование от класса gmTask, интерфейс которого не зависит от других компонентов библиотеки GridMD. Таким образом, реализованный пул потоков имеет слабую связность с остальными компонентами библиотеки, не зависит от их реализации и может быть использован обособленно в других проектах.

4. Тестирование эффективности многопоточной реализации исполнения локальных узлов в библиотеке GridMD

Тестирование эффективности многопоточной реализации исполнения локальных узлов производилось на примере расчета определенного интеграла функции. Расчет был реализован средствами GridMD (см Приложение), где он был представлен в виде графа, продемонстрированном на Рис. 11.

Рис. 11 Граф исполнения GridMD для расчета определенного интеграла

Распараллеливание интегрирования функции возможно путем разбиения области интегрирования на отдельные интервалы, параллельного интегрирования функции на каждом из интервалов и итогового суммирования результатов интегрирования на интервалах разбиения. Именно такую схему изображает представленный граф исполнения, в узле start которого разбивается исходная область интегрирования на интервалы, в узлах split происходит интегрирование функции на каждом из интервалов и в узле end суммируется результат исполнения всех узлов split. Рассмотренная схема концептуально является шаблоном ветвления, поддерживаемым в GridMD, и реализуется с помощью класса gmFork. Каждый из узлов графа был объявлен как локальный.

В текущей реализации менеджер сценариев по результатам разбора графа спланирует параллельное исполнение узлов split и отправит их на исполнение в пул потоков.

При тестировании рассчитывался определенный интеграл функции f(x) = sin(x) на отрезке [-1000,1000] c переменным шагом интегрирования. Тестирование производилось на ОС Ubuntu 14.04 и четырехядерном процессоре Intel Core i7 2670QM. Число рабочих потоков в пуле устанавливалось равным числу виртуальных ядер, а именно восьми. Сравнение времени расчета интеграла с помощью GridMD без поддержки многопоточности и при использовании пула потоков представлено на Рис. 12.

Рис. 12 Сравнение времени расчета с помощью многопоточной реализации GridMD и без поддержки потоков

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

Далее было проведено сравнение расчета интеграла с помощью многопоточной версии GridMD и пула потоков, реализованного с помощью средств библиотеки Boost в виде отдельного компонента с аналогичным классу gmThreadPool интерфейсом. (Рис. 13).

Рис. 13 Оценка накладных расходов, вносимых служебными действиями GridMD при расчете

Тaкое сравнение помогает оценить степень влияния накладных расходов GridMD по разбору графа менеджером сценариев, планированию и генерации заданий, разрешение зависимостей узлов по данным через файловую систему. На Рис. 13 видно, что кривые лежат близко друг к другу, и в среднем GridMD своими служебными действиями дает увеличение времени расчета в 6 секунд.

Заключение

В работе произведена успешная оптимизация выполнения локальных узлов графа исполнения приложений, основанных на библиотеки GridMD. В качестве метода для повышения производительности работы библиотеки было выбрано внедрение многопоточности в механизм исполнения локальных узлов. Анализ существующих методик в области разработки многопоточных приложений показал, что наиболее эффективным решением по внедрению многопоточности является разработка и интегрирование компонента пула потоков в контекст исполнения GridMD приложений. Из существующих вариантов инструментов разработки многопоточных компонентов была выбрана библиотека wxWidgets, успешно реализован пул потоков и проведены тесты, продемонстрировавшие увеличение быстродействия исполнения GridMD приложений, имеющих в составе графа исполнения локальные узлы.

Список используемых источников

1. Косяков М.С. Введение в распределенные вычисления. - СПб: НИУ ИТМО, 2014. - 155 с.

2. Распределенные системы. Принципы и парадигмы / Э. Таненбаум, М. ван Стеен. -- СПб.: Питер, 2003. -- 877 с: ил. -- (Серия «Классика computer science»).

3. Гергель В. П. Теория и практика параллельных вычислений //М.: БИНОМ. - 2007.

4. Nunn R. Distributed software architectures using middleware //3C05 Coursework. - 2002. - Т. 2. - С. 1.

5. Лазарев И. В., Сухорослов О. В. Использование workflow-методологии для описания процесса распределенных вычислений //Проблемы вычислений в распределенной среде: Модели обработки и представления данных. Динамические системы. Труды ИСА РАН. - 2005. - Т. 14. - С. 254-255.

6. Valuev I. A., Morozov I. V. Managing Dynamical Distributed Applications with GridMD Library //Computational Science and Its Applications--ICCSA 2015. - Springer International Publishing, 2015. - С. 272-289.

7. Morozov I. V., Valuev I. A. Automatic distributed workflow generation with GridMD library //Computer Physics Communications. - 2011. - Т. 182. - №. 9. - С. 2052-2058.

8. Валуев И. А., Морозов И. В. GridMD: компактная переносимая библиотека С++ для управления распределенными вычислениями. - 2014.

Размещено на Allbest.ru