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

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

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

Я ставил себе задачей продемонстрировать определенные приёмы, а не написать самый оптимальный код. Реализации, в которых лучше используется имеющееся оборудование, можно найти в академической литературе по параллельным алгоритмам или в специализированных многопоточных библиотеках типа Intel Threading Building Blocks [20] http://threadingbuildingblocks.org/ .

Концептуально простейшим параллельным алгоритмом является параллельная версия std::for_each, с которой я и начну.

Идея std::for_eachпроста — этот алгоритм вызывает предоставленную пользователем функцию для каждого элемента диапазона. Различие между параллельной и последовательной реализацией std::for_eachзаключается, прежде всего, в порядке вызовов функции. Стандартная версия std::for_eachвызывает функцию сначала для первого элемента диапазона, затем для второго и так далее, тогда как параллельная версия не дает гарантий относительно порядка обработки элементов, они даже могут (и хочется надеяться, будут ) обрабатываться параллельно.

Для реализации параллельной версии нужно всего лишь разбить диапазон на участки, которые будут обрабатываться каждым потоком. Количество элементов известно заранее, поэтому такое разбиение можно произвести до начала работы (см. раздел 8.1.1). Мы будем предполагать, что это единственная исполняемая параллельная задача, поэтому вычислить количество требуемых потоков можно с помощью функции std::thread::hardware_concurrency(). Мы также знаем, что элементы можно обрабатывать абсолютно независимо, поэтому для предотвращения ложного разделения (см. раздел 8.2.3) имеет смысл использовать соседние блоки.

По своей структуре этот алгоритм похож на параллельную версию std::accumulate, описанную в разделе 8.4.1, только вместо вычисления суммы элементов он применяет к ним заданную функцию. На первый взгляд, это должно бы существенно упростить код, потому что не нужно возвращать никакой результат. Но если мы собираемся передавать исключения вызывающей программе, то все равно придется воспользоваться механизмами std::packaged_taskи std::future, чтобы передавать исключения из одного потока в другой. Ниже приведен пример реализации.

Листинг 8.7.Параллельная реализация std::for_each

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;

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::vector > futures(num_threads - 1); ← (1)

std::vector threads(num_threads – 1);

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);

std::packaged_task task( ← (2)

[=]() {

std::for_each(block_start, block_end, f);

});

futures[i] = task.get_future();

threads[i] = std::thread(std::move(task)); ← (3)

block_start = block_end;

}

std::for_each(block_start, last, f);

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

futures[i].get(); ← (4)

}

}

Структурно код ничем не отличается от приведенного в листинге 8.4, что и неудивительно. Основное различие состоит в том, что в векторе futures хранятся объекты std::future (1), потому что рабочие потоки не возвращают значение, а в качестве задачи мы используем простую лямбда-функцию, которая вызывает функцию fдля элементов из диапазона от block_startдо block_end (2). Это позволяет не передавать конструктору потока (3)диапазон. Поскольку рабочие потоки ничего не возвращают, обращения к futures[i].get() (4)служат только для получения исключений, возникших в рабочих потоках; если мы не хотим передавать исключения, то эти обращения можно вообще опустить.