Многопоточное программирование на C++

1. Многопоточное программирование на C++

  Работа с памятью

Для начала поговорим  о подводных камнях C++ применительно  к MT программированию. На первый взгляд может показаться, что написанные на C++ потоки, не изменяющие одни и те же данные одновременно, могут работать абсолютно независимо и параллельно. К сожалению, это не совсем так: основной проблемой является стандартный  распределитель памяти, т.е. операторы new/delete. Суть проблемы в том, что приложение получает свободные блоки памяти от системы и выдает их части потокам  по требованию, т.е. в процессе своей  работы потоки вынуждены изменять одни и те же структуры данных (глобальные цепочки свободных блоков) и, следовательно, синхронизация их работы неизбежна.

Прямым решением данной проблемы является повсеместное явное использование собственного распределителя памяти, кэширующего  полученные от глобального распределителя блоки. Т.е. своего объекта mem_pool для каждого отдельного потока (как минимум). Конечно, с точки зрения удобства кодирования повсеместное мелькание ссылок mem_pool& трудно назвать приемлемым -- стоит ли овчинка выделки? Давайте разберемся с помощью следующего примера:

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

  Работа с файлами

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

Или все же покажет?!

Как известно, стандартные потоки ввода/вывода C++ работают существенно медленнее потоков ввода/вывода C. Но, к сожалению, даже C-шные потоки FILE не подходят для MT приложений, т.к. все операции над ними обязаны быть thread-safe по умолчанию, а быстрые варианты функций getc() и putc(), не использующие пресловутые mutex-ы, имеют другие имена: getc_unlocked() и putc_unlocked() соответственно. Тем самым, написание кода, одинаково хорошо работающего как в ST, так и в MT окружении становится невозможным.

Главным образом, для решения этой проблемы и был  создан класс file, не использующий блокировок:

Пример запускает  заданное количество потоков, в каждом из которых создается копия указанного в командной строке файла посредством  выбранной функции: start_std() или start_ders(). Потоками используется посимвольное копирование через буфер размера BUF_SIZE.  

  Многопоточные программы 

Коль  скоро с примерами уже покончено -- самое время перейти к программам. Удивительно, но факт: все из них (кроме mtftext, естественно) я ежедневно использую на практике! Т.е. даже этот функционально неполный инструментарий derslib, специально написанный для данной статьи, вполне подходит для создания реально полезных программ!

Пример

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

Более того, согласно замыслу прикладной код ничего не должен знать о количестве одновременно работающих потоков, да и вообще каким  бы то ни было образом ссылаться  на объекты типа mutex и thread.

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

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

Просматриваем все имена заданной директории.

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

Если найденное  имя является файлом и удовлетворяет  маске, то нужно просмотреть содержимое данного файла для поиска подходящих строк.

Операции 2 и 3, очевидно, можно выполнять параллельно  операции 1, а сам алгоритм, при  этом, будет выглядеть следующим  образом:

Помещаем в  очередь сообщений первое сообщение  типа FindFiles, содержащее имя корневой директории поиска.

С помощью одного или нескольких потоков начинаем обрабатывать очередь, а именно:

Пока очередь  не пуста и не находится в "прерванном" состоянии, начинаем извлекать сообщения.

Если извлечено  сообщение типа FindFiles -- обрабатываем директорию, порождая сообщения FindFiles и ScanFile.

Если извлечено  сообщение типа ScanFile -- обрабатываем файл, выводя найденные строки.

В случае возникновения  ошибки -- переводим очередь в "прерванное" состояние.

Проверяем состояние  очереди и завершаем работу с  соответствующим кодом возврата.

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

/** @file

* Main file of mtftext program.

*/ 

#include <vector>

#include <stdlib.h>

#include <ders/dir.hpp>

#include <ders/file.hpp>

#include <ders/text_buf.hpp>

#include <ders/thread_pool.hpp>

#include <ders/wldcrd_mtchr.hpp>

#include "msg.hpp" 

namespace mtftext {  // ::mtftext

Весь исходный код, кроме функции main(), естественно, заключен в namespace, совпадающий с именем программы.

using namespace ders; 

struct CmdLineParser {

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

       bool isS;

       int numThr;

       sh_text word;

       sh_text mask; 

       CmdLineParser(mem_pool& mp) : isS(false), numThr(0), word(nt(mp)), mask(

         nt(mp)) {} 

       void parse(int argc, char** argv);

}; 

struct MainTask : public task {

Главная структура  программы, параллельно выполняющая  свою функцию proc() с помощью класса ders::thread_pool.

       bool exitOnErr;

       sh_text srchPatt;

       wldcrd_mtchr fileMchr; 

       MainTask(bool eoe, sh_text sp, sh_text mk) : exitOnErr(eoe),

         srchPatt(sp), fileMchr(sp.pool(), mk) {} 

       virtual void destroy(mem_pool& mp2) { destroy_this(this, mp2); }

Типичная реализация чистой виртуальной функции destroy, косвенно унаследованной от интерфейса ders::destroyable.

       virtual void proc(mem_pool& mp, const dq_vec& dqv, void* arg, task_opers&

         to); 

       void doFindFiles(mem_pool& mp, data_queue& dq, const FindFilesMsg& msg);

       void doScanFile(mem_pool& mp, data_queue& dq, const ScanFileMsg& msg);

}; 

void CmdLineParser::parse(int, char** argv)

{

 const char* usage="mtftext [-s] num_threads word mask";

Командная строка имеет необязательный параметр -s (stop), предписывающий сразу же завершать работу при обнаружении ошибок. По умолчанию же в stderr записывается сообщение об ошибке и работа продолжается.

 mem_pool& mp=word.pool(); 

 char** it=argv;

 if (!*++it) throw newExitMsgException(mp, _FLINE_, usage, 1);

Возбуждаем исключение ExitMsgException в случае ошибки, которое используется для выхода из программы с заданным кодом возврата и, возможно, текстом сообщения. Похожего результата можно добиться и с помощью пары fprintf()/exit(), но в этом случае не будут вызваны деструкторы локальных объектов, что, вообще говоря, неприемлемо.