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

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

4.1. mtftext.exe: учебный пример

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

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

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

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

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

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

1. Помещаем в очередь сообщений первое сообщение типа FindFiles, содержащее имя корневой директории поиска.
2. С помощью одного или нескольких потоков начинаем обрабатывать очередь, а именно:
3. Пока очередь не пуста и не находится в "прерванном" состоянии, начинаем извлекать сообщения.
1. Если извлечено сообщение типа FindFiles -- обрабатываем директорию, порождая сообщения FindFiles и ScanFile.
2. Если извлечено сообщение типа ScanFile -- обрабатываем файл, выводя найденные строки.
3. В случае возникновения ошибки -- переводим очередь в "прерванное" состояние.
4. Проверяем состояние очереди и завершаем работу с соответствующим кодом возврата.

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

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

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

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

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

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

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

Проверяем заданное пользователем количество потоков на соответствие разумным рамкам.

Этот объемлющий цикл вокруг цикла обработки сообщений встречается только в программе mtftext, т.к. только в ней у пользователя есть возможность не прерывать работу в случае обнаружения ошибок.

Отсутствие прочитанного сообщения может означать как исчерпание всех сообщений, так и is_intr() состояние очереди.

На корректно работающих компиляторах для обработки исключения вполне достаточно единственного блока catch (...) { recatchException() }, что сильно упрощает программирование, но, к сожалению, реальный мир требует жертвовать изящностью кода ради возможности использования важных промышленных компиляторов.

Используем объект text_buf для удобного объединения строки и символа.

Создаем файл для вывода, привязанный к дескриптору stdout.

Записываем в очередь первое сообщение, предписывающее искать файлы в текущей директории.

Создаем thread_pool с указанным пользователем количеством рабочих потоков: если numThr равен одному, то никаких потоков создавать не требуется и для работы используется специальная однопоточная реализация thread_pool интерфейса.

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

Запускаем одновременное выполнение функции MainTask::proc() всеми потоками thread_pool-а + вызвавшим exec() потоком функции main(). Именно поэтому в new_thread_pool() передается значение numThr-1.

Проверяем состояние очереди для определения причины окончания обработки и передаем ОС соответствующий код возврата.

Типичный блок обработки исключения ExitMsgException.

4.2. mtcksrc.exe: проверка исходного кода

Программа предназначена для рекурсивной проверки файлов исходного кода на соответствие следующим условиям:

• все строки файла соответствуют DOS или UNIX формату
• строки не содержат символов табуляции
• длина строки не превышает 80 символов
• отсутствуют пробелы в конце строки

Параметры командной строки: mtcksrc [-nt2] d|u mask[,mask2...]

Примеры использования:
mtcksrc.exe u *.h,*.hpp,*.c,*.cpp
mtcksrc.exe -nt1 d *.?pp

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

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

Ну, а теперь самое время проверить ваши собственные программы -- из-за пресловутого Copy/Paste DOS фрагменты не так уж и редко встречаются в UNIX файлах...

4.3. mtdel.exe: удаление файлов

Программа предназначена для рекурсивного удаления файлов по маскам.

Внимание! Удаление файлов зачастую приводит к безвозвратной потере данных, не запускайте эту программу из праздного любопытства!

Параметры командной строки: mtdel [-nt2] mask[,mask2...]

Примеры использования:
mtdel.exe *.obj,*.exe,*.res,out.???\*
mtdel.exe -nt1 *.bak

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

1. Вывод итоговой статистики, т.е. общего количества удаленных файлов и их совокупного размера.

К несчастью, решение данной простой задачи не вызовет никакого труда даже у едва знакомого с многопоточным программированием кодировщика: всего-то и нужно, что создать один глобальный счетчик, да защитить его mutex-ом!

С другой стороны, толковым программистам уже известно чем чреваты глобальные объекты, защищенные mutex-ами, а некоторые даже знают о том, что грамотный MT дизайн категорически не приветствует подобного рода решений!

Концептуально правильное решение проблемы получения итоговой статистики состоит в том, что каждый из рабочих потоков ведет свою собственную личную статистику, которую он может изменять абсолютно независимо и параллельно. А итоговая статистика получается в конце работы программы путем сложения их всех -- и никаких тебе mutex-ов!

Для создания личных аргументов потока предназначена функция MainTask::proc_arg(). Она вызывается thread_pool-ом для получения указателей, которые им будут переданы как arg в MainTask::proc(). Функция MainTask::proc_arg() сохраняет созданные объекты в списке, который затем обрабатывается MainTask::getStat() для получения итоговой статистики.

Отмечу, что для хранения Stat аргументов потоков специально используется список list<Stat>, а не более привычный vector<Stat>. Дело в том, что в процессе добавления элементов в вектор используемая им память неоднократно перераспределяется, так что указатели на объекты Stat, ранее возвращенные функцией MainTask::proc_arg() инвалидируются. А воспользоваться вызовом vector::reserve() мы не можем в силу того, что MainTask ничего не должна знать о количестве рабочих потоков.

1. Сообщение о возникших ошибках, как последнее сообщение программы.

А вот еще одна серьезная задача, когда пресловутого mutex-а, казалось бы, уж точно не избежать! Суть проблемы в том, что если сразу же выводить на экран возникающие в процессе работы ошибки, то они вполне могут затеряться в выводе других потоков, продолжающих обработку своих сообщений. Решением задачи является запись сообщения об ошибке в специальный глобальный буфер, чтобы после окончания работы MainTask::proc() функция main() смогла его вывести последним.

В этом случае действительно имеет смысл завести общий для всех потоков буфер MainTask::gerr, распределенный с помощью глобального пула MainTask::gmp, и разграничить к нему доступ посредством привычной блокировки. Мы, конечно, могли бы завести по буферу в каждом аргументе потока и объединить их в конце работы точно так же, как мы объединяем статистику, но в данном случае усложнение кода программы неоправданно, т.к. потокам не нужно постоянно обращаться к этим данным и никаких дополнительных накладных расходов из-за данной блокировки не возникает.

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

К счастью, решение есть и оно заключается в использовании интерфейса task_opers, передаваемого в виде аргумента to в функцию MainTask::proc() thread_pool-ом. Он предоставляет функцию invoke(), которая позволяет вызывать переданный указатель на функцию с применением необходимой блокировки, т.е. необходимый (или нежелательный!) mutex автоматически обеспечивается самим thread_pool-ом:

приводим аргумент потока к его настоящему типу Stat

заполняем ErrData и вызываем функцию addError() для добавления сообщения об ошибке

4.4. mtcnvsrc.exe: конвертация исходного кода

Программа предназначена для рекурсивной конвертации файлов исходного кода в DOS/UNIX формат и/или удаления пробелов в конце строки.

Параметры командной строки: mtcnvsrc [-nt2] [d|u][e] mask[,mask2...]

Примеры использования:
mtcnvsrc.exe ue *.h,*.hpp,*.c,*.cpp
mtcnvsrc.exe -nt4 e *.?pp

Как можно видеть, прямым назначением программы является исправление ошибок, найденных mtcksrc. При этом, автоматически исправляется только то, что не требует человеческого вмешательства.

Ну а если говорить о сложности, то данная программа скорее является небольшим подготовительным этапом перед следующей, чем усложнением предыдущей. Единственный момент, на который можно обратить внимание -- это аргумент потока в виде более общей структуры Arg. На этот раз аргументы хранятся в векторе vector<sh_ptr<Arg> >, но это не приводит к описанным выше проблемам, т.к. в процессе их создания копируются и перераспределяются элементы sh_ptr<Arg>, а не сами тяжеловесные Arg структуры.

4.5. mtdirdiff.exe: сравнение директорий

Программа предназначена для рекурсивного сравнения директорий.

Параметры командной строки: mtdirdiff [-nt2] old_dir new_dir diff_dir

Примеры использования:
mtdirdiff.exe src\mtprog.sav src\mtprog src\mtprog.diff

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

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

И, как следствие, устройство функции proc_arg() немного усложнено:

Необходимость разбиения на этапы возникает из желания эффективным и естественным образом организовать сканирование и последующее сравнение old_dir и new_dir директорий:

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

Кстати сказать, проверка данной гипотезы является неплохим упражнением для самостоятельного изучения MT программирования -- возьметесь?! Вот и мне тоже лень...

1. Затем стартует сравнение просканированных директорий, причем потокам не нужно синхронизировать свой доступ к hash_vec-рам с именам, т.к. их содержимое уже никем не изменяется.

Как можно видеть, без разбиения на этапы потокам пришлось бы синхронизировать свой доступ к hash_vec-рам, что неприемлемо с точки зрения правильного MT дизайна.

1. Структура DirCont использует свой собственный пул для имен директорий и файлов:

Здесь мы имеем тот редкий случай, когда в одной точке приложения одновременно сходятся несколько объектов mem_pool, принадлежащих разным потокам:

1. объект ownmp, принадлежащий соответствующей структуре mt.oldDir или mt.newDir
2. привычный объект mp, являющийся временным личным объектом рабочего потока, создаваемым thread_pool-ом на время работы функции MainTask::proc()

Как следствие, doFindFiles() создает сохраняемые имена директорий и файлов с помощью ownmp:

А doCompare() и doCopyFile() запоминают найденные имена (факт. ключи hash_vec<sh_text,...>-ра) по ссылке, а не значению:

Т.к. сохранение по значению приведет к копированию объекта sh_ptr<text> и, как следствие, одновременному использованию ownmp несколькими рабочими потоками. Тем самым в программе появится ошибка синхронизации, т.е. та самая "великая и ужасная" race condition, непонятно когда и как проявляющаяся.

И если желаете, то в качестве неприятного, но весьма поучительного упражнения можно и в самом деле заменить ссылки на значения и попытаться ее отследить...

1. Изменившийся файл может находиться достаточно глубоко в структуре сравниваемых поддиректорий, поэтому его копирование в директорию diff_dir/mod/... может потребовать предварительного создания недостающих промежуточных поддиректорий. Задача осложняется еще и тем, что подлежащие созданию поддиректории могут быть созданы параллельно работающими потоками в промежутке между временем обнаружения их отсутствия и попыткой их создать. Для эффективного решения данной проблемы приложение пробует сразу же создать несуществующий файл и только в случае ошибки выполняется попытка создания промежуточных директорий и повторная попытка создания файла:

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