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

С инициализацией объекта block_start (5)всё чисто по тем же причинам, так что перейдём к операциям в цикле запуска потоков (6), (7), (8). Если после создания первого же потока (7)возникнет исключение, и мы его не перехватим, появится проблема; деструкторы объектов std::threadвызывают std::terminate, что приводит к аварийному завершению программы. Нехорошо.

Обращение к accumulate_blockв точке (9)может возбуждать исключения — с точно такими же последствиями: объекты потоков будут уничтожены, а их деструкторы вызовут std::terminate. С другой стороны, исключение, возбуждаемое в последнем обращении к std::accumulate (10), не так опасно, потому что к этому моменту все потоки уже присоединились к вызывающему.

Таким образом, обращения к accumulate_blockиз новых потоков могут возбуждать исключения в точке (1). Так как блоки catchотсутствуют, то исключение останется необработанным и приведёт к вызову std::terminate()и завершению программы.

Если это еще не очевидно, скажем прямо: этот код не безопасен относительно исключений .

Итак, мы выявили все возможные точки возбуждения исключений и поняли, к каким печальным последствиям это приведёт. Что с этим можно сделать? Начнем с вопроса об исключениях, возбуждаемых в созданных нами потоках.

В главе 4 мы уже познакомились с подходящим для решения проблемы средством. Для чего нам вообще нужны новые потоки? Для того чтобы вычислить результат и при этом учесть возможность возникновения исключений. Но это именно то , для чего предназначено сочетание std::packaged_taskи std::future. В листинге 8.3 показан код, переписанный с использованием std::packaged_task.

Листинг 8.3.Параллельная версия std::accumulateс применением std::packaged_task

template

struct accumulate_block {

T operator()(Iterator first, Iterator last) {← (1)

return std::accumulate(first, last, T()); ← (2)

}

};

template

T parallel_accumulate(Iterator first, Iterator last, T init) {

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

if (!length)

return init;

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 i = 0 ? hardware_threads : 2, max_threads);

unsigned long const block_size = length / num_threads;

std::vector > futures(num_threads-1);← (3)

std::vector threads(num_threads — 1);

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( ← (4)

accumulate_block());

futures[i] = task.get_future(); ← (5)

threads[i] =

std::thread(std::move(task), block_start, block_end);← (6)

block_start = block_end;

}

T last_result = accumulate_block()(block_start, last); ← (7)

std::for_each(threads.begin(), threads.end(),

std::mem_fn(&std::thread::join));

T result = init; ← (8)

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

result += futures[i].get(); ← (9)

}

result += last_result; ← (10)

return result;

}

Первое изменение заключается в том, что оператор вызова в accumulate_blockтеперь возвращает результат по значению, а не принимает ссылку на место, где его требуется сохранить (1). Для обеспечения безопасности относительно исключений мы используем std::packaged_taskи std::future, поэтому можем воспользоваться этим и для передачи результата. Правда, для этого требуется при вызове std::accumulate (2)явно передавать сконструированный по умолчанию экземпляр T, а не использовать повторно предоставленное значение result, но это не слишком существенное изменение.

Далее, вместо того заводить вектор результатов, мы создаем вектор futures (3), в котором храним объекты std::futureдля каждого запущенного потока. В цикле запуска потоков мы сначала создаем задачу для accumulate_block (4). В классе std::packaged_taskобъявлена задача, которая принимает два объекта Iteratorи возвращает T, а это именно то, что делает наша функция. Затем мы получаем будущий результат для этой задачи (5)и исполняем ее в новом потоке, передавая начало и конец обрабатываемого блока (6). Результат работы задачи, равно как и исключение, если оно возникнет, запоминается в объекте future.