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

template

void parallel_for_each(Iterator first, Iterator last, Func f) {

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

if (!length)

return;

unsigned long const min_per_thread = 25;

if (length < (2 * min_per_thread)) {

std::for_each(first, last, f); ← (1)

} else {

Iterator const mid_point = first + length / 2;

std::future first_half = ← (2)

std::async(&parallel_for_each,

first, mid_point, f);

parallel_for_each(mid_point, last, f); ← (3)

first_half.get(); ← (4)

}

}

Как и в случае реализации parallel_accumulateс помощью std::asyncв листинге 8.5, мы разбиваем данные рекурсивно в процессе выполнения, а не заранее, потому что не знаем, сколько потоков задействует библиотека. На каждом шаге данные делятся пополам, пока их не останется слишком мало для дальнейшего деления. При этом одна половина обрабатывается асинхронно (2), а вторая — непосредственно (3). Когда дальнейшее деление становится нецелесообразным, вызывается std::for_each (1). И снова использование std::asyncи функции-члена get()объекта std::future (4)обеспечивает семантику распространения исключения.

Теперь перейдем от алгоритмов, которые выполняют одну и ту же операцию над каждым элементом (к их числу относятся также std::countи std::replace), к чуть более сложному случаю — std::find.

Далее будет полезно рассмотреть алгоритм std::find, потому что это один из нескольких алгоритмов, которые могут завершаться еще до того, как обработаны все элементы. Например, если уже первый элемент в диапазоне отвечает условию поиска, то рассматривать остальные не имеет смысла. Как мы скоро увидим, это свойство существенно для повышения производительности, и от него напрямую зависит структура параллельной реализации. На этом примере мы продемонстрируем, как порядок доступа к данным может оказать влияние на проектирование программы (см. раздел 8.3.2). К той же категории относятся алгоритмы std::equalи std::any_of.

Если вы вместе с женой или другом ищете какую-нибудь старую фотографию в сваленных на чердаке альбомах, то вряд ли захотите, чтобы они продолжали перелистывать страницы, когда вы уже нашли то, что нужно. Наверное, вы просто сообщите им, что искомое найдено (быть может, крикнув «Есть!»), чтобы они могли прекратить поиски и заняться чем-нибудь другим. Но многие алгоритмы по природе своей должны обработать каждый элемент и, стало быть, не имеют эквивалента восклицанию «Есть!». Для алгоритмов типа std::findумение «досрочно» прекращать работу — важное свойство, которое нельзя игнорировать. И, следовательно, его нужно учитывать при проектировании кода — закладывать возможность прерывания других задач, когда ответ уже известен, чтобы программа не ждала, пока прочие рабочие потоки обработают оставшиеся элементы.

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

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

Мы уже знаем о двух способах возврата значений и распространения исключений. Можно использовать массив будущих результатов и объекты std::packaged_taskдля передачи значений и исключений, после чего обработать частичные результаты в главном потоке. Или с помощью std::promiseустанавливать окончательный результат прямо в рабочем потоке. Все зависит от того, как мы хотим обрабатывать исключения, возникающие в рабочих потоках. Если требуется остановиться при первом возникновении исключения (несмотря на то, что обработаны не все элементы), то можно использовать std::promiseдля передачи значения и исключения. С другой стороны, если мы хотим, чтобы рабочие потоки продолжали поиск, то используем std::packaged_task, сохраняем все исключения, а затем повторно возбуждаем одно из них, если искомый элемент не найден.

В данном случае я остановился на std::promise, потому что такое поведение больше походит на поведение std::find. Надо только не забыть о случае, когда искомого элемента в указанном диапазоне нет. Поэтому необходимо дождаться завершения всех потоков перед тем, как получать значение из будущего результата. Если просто блокировать исполнение при обращении к get(), то при условии, что искомого элемента нет, мы будем ждать вечно. Получившийся код приведён ниже.

Листинг 8.9.Параллельная реализация алгоритма find()

template

Iterator parallel_find(Iterator first, Iterator last,

MatchType match) {

struct find_element { ← (1)

void operator()(Iterator begin, Iterator end,

MatchType match,

std::promise* result,

std::atomic* done_flag) {

try {

for(; (begin != end) && !done_flag->load(); ++begin) {← (2)

if (*begin == match) {

result->set_value(begin);← (3)

done_flag->store(true); ← (4)

return;

}

}

} catch (...) { ← (5)

try {

result->set_exception(std::current_exception());← (6)

done_flag->store(true);

} catch (...) ← (7)

{}

}

}

};

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

if (!length)

return last;

unsigned long const min_per_thread = 25;

unsigned long const max_threads =

(length + min_per_thread — 1) / min_per_thread;

unsigned long const hardware_threads =

std::thread::hardware_concurrency();

unsigned long const num_threads =

std::min(

hardware_threads != 0 ? hardware_threads : 2, max_threads);

unsigned long const block_size = length / num_threads;

std::promise result; ← (8)

std::atomic done_flag(false);← (9)

std::vector threads(num_threads — 1); {← (10)

join_threads joiner(threads);

Iterator block_start = first;

for (unsigned long i = 0; i < (num_threads - 1); ++i) {

Iterator block_end = block_start;

std::advance(block_end, block_size);