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

 if (input.empty()) {
  return input;
 }

 std::list result;
 result.splice(result.begin(), input, input.begin());
 T const& pivot = *result.begin();

 auto divide_point = std::partition(input.begin(), input.end(),
  [&](T const& t) {return t
 std::list lower_part;
 lower_part.splice(
  lower_part.end(), input, input.begin(), divide_point);

 std::future > new_lower( ←(1)

  std::async(¶llel_quick_sort, std::move(lower_part)));

 auto new_higher(

  parallel_quick_sort(std::move(input))); ←(2)

 result.splice(result.end(), new_higher); ←(3)

 result.splice(result.begin(), new_lower.get()); ←(4)

 return result;
}

Существенное изменение здесь заключается в том, что сортировка нижней части списка производится не в текущем, а в отдельном потоке — с помощью std::async()(1). Верхняя часть списка сортируется путем прямой рекурсии, как и раньше (2). Рекурсивно вызывая parallel_quick_sort(), мы можем задействовать доступный аппаратный параллелизм. Если std::async() создает новый поток при каждом обращении, то после трех уровней рекурсии мы получим восемь работающих потоков, а после 10 уровней (когда в списке примерно 1000 элементов) будет работать 1024 потока, если оборудование позволяет. Если библиотека решит, что запущено слишком много задач (быть может, потому что количество задач превысило уровень аппаратного параллелизма), то может перейти в режим синхронного запуска новых задач. Тогда новая задача будет работать в том же потоке, который обратился к get(), а не в новом, так что мы не будем нести издержки на передачу задачи новому потоку, если это не увеличивает производительность. Стоит отметить, что в соответствии со стандартом реализация std::async вправе как создавать новый поток для каждой задачи (даже при значительном превышении лимита), если явно не задан флаг std::launch::deferred, так и запускать все задачи синхронно, если явно не задан флаг std::launch::async. Рассчитывая, что библиотека сама позаботится об автоматическом масштабировании, изучите, что говорится на эту тему в документации, поставляемой вместе с библиотекой.

Можно не использовать std::async(), а написать свою функцию spawn_task(), которая будет служить оберткой вокруг std::packaged_task и std::thread, как показано в листинге 4.14; нужно создать объект std::packaged_task для хранения результата вызова функции, получить из него будущий результат, запустить задачу в отдельном потоке и вернуть будущий результат. Само по себе это не дает большого преимущества (и, скорее всего, приведёт к значительному превышению лимита), но пролагает дорогу к переходу на более хитроумную реализацию, которая помещает задачу в очередь, обслуживаемую пулом потоков. Рассмотрение пулов потоков мы отложим до главы 9. Но идти по такому пути вместо использования std::async имеет смысл только в том случае, когда вы точно знаете, что делаете, и хотите полностью контролировать, как пул потоков строится и выполняет задачи.

Но вернемся к функции parallel_quick_sort. Поскольку для получения new_higher мы применяли прямую рекурсию, то и срастить (splice) его можно на месте, как и раньше (3). Но new_lower теперь представляет собой не список, а объект std::future>, поэтому сначала нужно извлечь значение с помощью get(), а только потом вызывать splice()(4). Таким образом, мы дождемся завершения фоновой задачи, а затем переместим результат в параметр splice(); функция get() возвращает ссылку на r-значение — хранимый результат, следовательно, его можно переместить (подробнее о ссылках на r-значения и семантике перемещения см. в разделе А.1.1 приложения А).

Даже в предположении, что std::async() оптимально использует доступный аппаратный параллелизм, приведённая реализация Quicksort все равно не идеальна. Основная проблема в том, что std::partition делает много работы и остается последовательной операцией, но пока остановимся на этом. Если вас интересует максимально быстрая параллельная реализация, обратитесь к научной литературе.

Листинг 4.14. Простая реализация функции spawn_task

template
std::future::type>
spawn_task(F&& f, A&& a) {
 typedef std::result_of::type result_type;
 std::packaged_task
 task(std::move(f)));
 std::future res(task.get_future());
 std::thread t(std::move(task), std::move(a));
 t.detach();
 return res;
}

Функциональное программирование — не единственная парадигма параллельного программирования, позволяющая избежать модификации разделяемых данных. Альтернативой является парадигма CSP (Communicating Sequential Processes — взаимодействующие последовательные процессы)[10], в которой потоки концептуально рассматриваются как полностью независимые сущности, без каких бы то ни было разделяемых данных, но соединенные коммуникационными каналами, по которым передаются сообщения. Эта парадигма положена в основу языка программирования Erlang (http://www.erlang.org/) и среды MPI (Message Passing Interface) (http://www.mpi-forum.org/), широко используемой для высокопроизводительных вычислений на С и С++. Уверен, что теперь вы не удивитесь, узнав, что и эту парадигму можно поддержать на С++, если соблюдать определенную дисциплину; в следующем разделе показано, как это можно сделать.