Энтони Уильямс - Параллельное программирование на С++ в действии. Практика разработки многопоточных программ

threads[i] = std::thread(find_element(), ← (11)

block_start, block_end, match,

&result, &done_flag);

block_start = block_end;

}

find_element()(

block_start, last, match, &result, &done_flag);← (12)

if (!done_flag.load()) { ← (13)

return last;

}

return result.get_future().get() ← (14)

}

В основе своей код в листинге 8.9 похож на предыдущие примеры. На этот раз вся работа производится в операторе вызова, определенном в локальном классе find_element (1). Здесь мы в цикле обходим элементы из назначенного потоку блока, проверяя флаг на каждой итерации (2). Если искомый элемент найден, то мы записываем окончательный результат в объект-обещание (3)и перед возвратом устанавливаем флаг done_flag (4).

Если было возбуждено исключение, то его перехватит универсальный обработчик (5)и попытается сохранить исключение в обещании (6)перед установкой done_flag. Но установка значения объекта-обещания может возбудить исключение, если значение уже установлено, поэтому мы перехватываем и игнорируем любые возникающие здесь исключения (7).

Это означает, что если поток, вызвавший find_element, найдет искомый элемент или возбудит исключение, то все остальные потоки увидят поднятый флаг done_flagи прекратят работу. Если несколько потоков одновременно найдут искомое или возбудят исключение, то возникнет гонка за установку результата в обещании. Но это безобидная гонка: победивший поток считается «первым», и установленный им результат приемлем.

В самой функции parallel_findмы определяем обещание (8)и флаг прекращения поиска (9), которые передаем новым потокам вместе с диапазоном для просмотра (11). Кроме того, главный поток пользуется классом find_elementдля поиска среди оставшихся элементов (12). Как уже отмечалось, мы должны дождаться завершения всех потоков перед тем, как проверять результат, потому что искомого элемента может вообще не оказаться. Для этого мы заключаем код запуска и присоединения потоков в блок (10), так что к моменту проверки флага (13)все потоки гарантировано присоединены. Если элемент был найден, то, обратившись к функции get()объекта std::future, мы либо получим результат из обещания, либо возбудим сохраненное исключение.

Как и раньше, в этой реализации предполагается, что мы собираемся использовать все доступные аппаратные потоки или располагаем каким-то механизмом, который позволит заранее определить количество потоков для предварительного разделения между ними работы. И снова мы можем упростить код, воспользовавшись функцией std::asyncи рекурсивным разбиением данных, если готовы принять автоматический механизм масштабирования, скрытый в стандартной библиотеке С++. Реализация parallel_findс применением std::asyncприведена в листинге ниже.

Листинг 8.10.Параллельная реализация алгоритма find()с применением std::async

template ← (1)

Iterator parallel_find_impl(Iterator first, Iterator last,

MatchType match,

std::atomic& done) {

try {

unsigned long const length = std::distance(first, last);

unsigned long const min_per_thread = 25; ← (2)

if (length < (2 * min_per_thread)) { ← (3)

for(; (first != last) && !done.load(); ++first) {← (4)

if (*first == match) {

done = true; ← (5)

return first;

}

}

return last; ← (6)

} else {

Iterator const mid_point = first + (length / 2); ← (7)

std::future async_result =

std::async(&parallel_find_impl,← (8)

mid_point, last, match, std::ref(done));

Iterator const direct_result =

parallel_find_impl(first, mid_point, match, done); ← (9)

return (direct_result == mid_point) ?

async_result.get() : direct_result; ← (10)

}

} catch (...) {

done = true; ← (11)

throw;

}

}

template

Iterator parallel_find(

Iterator first, Iterator last, MatchType match) {

std::atomic done(false);

return parallel_find_impl(first, last, match, done); ← (12)

}

Желание закончить поиск досрочно при обнаружении совпадения заставило нас ввести флаг, разделяемый между всеми потоками. Этот флаг, следовательно, нужно передавать во все рекурсивные вызовы. Проще всего сделать это, делегировав работу отдельной функции (1), которая принимает дополнительный параметр — ссылку на флаг done, передаваемый из главной точки входа (12).

Основная же ветвь кода не таит никаких неожиданностей. Как и во многих предыдущих реализациях, мы задаем минимальное количество элементов, обрабатываемых в одном потоке (2); если размер обеих половин диапазона меньше этой величины, то весь диапазон обрабатывается в текущем потоке (3). Собственно алгоритм сводится к простому циклу — он продолжается, пока не будет достигнут конец заданного диапазона или не установлен флаг done (4). При обнаружении совпадения мы устанавливаем флаг doneи выходим из функции (5). Если мы дошли до конца списка или вышли из цикла, потому что другой поток установил флаг done, то возвращаем значение last, означающее, что совпадение не найдено (6).

Если диапазон можно разбивать, то мы сначала находим среднюю точку (7), а потом через std::asyncзапускаем поиск во второй половине диапазона (8), не забыв передать ссылку на флаг doneс помощью std::ref. Одновременно мы просматриваем первую половину диапазона, рекурсивно вызывая себя же (9). И асинхронный, и рекурсивный вызов могут разбивать диапазон и дальше, если он достаточно велик.

Если прямой рекурсивный вызов вернул mid_point, значит, он не нашел совпадения, поэтому нужно получить результат асинхронного поиска. Если и в той половине ничего не было найдено, то мы получим last (10). Если «асинхронный» вызов на самом деле был не асинхронным, а отложенным, то выполняться он начнет именно при обращении к get(); в таком случае поиск во второй половине списке вообще не будет производиться, если поиск в первой оказался успешным. Если же асинхронный поиск действительно выполнялся в другом потоке, то деструктор переменной async_resultбудет ждать завершения этого потока, поэтому утечки потоков не произойдет.

Как и раньше, применение std::asyncгарантирует безопасность относительно исключений и распространения исключений вверх по стеку вызовов. Если прямой рекурсивный вызов возбудит исключение, то деструктор будущего результата позаботится о том, чтобы поток, в котором работал асинхронный поиск, завершился до возврата из функции. Если исключение возбудит асинхронный вызов, то оно распространится вверх при вызове get() (10). Внешний блок try/catchнужен только для того, чтобы установить флаг doneи обеспечить тем самым быстрое завершение всех потоков в случае исключения (11). Программа правильно работала бы и без этого, по продолжала бы сравнивать элементы до естественного завершения всех потоков.