Параллельное программирование на С++ в действии

Этот класс является простой оберткой вокруг std::deque (1), которая защищает все операции доступа к очереди с помощью мьютекса. Функции push() (2) и try_ pop() (3) работают с началом очереди, а функция try_steal() — с концом (4).

Получается, что эта «очередь» для потока-владельца на самом деле является стеком, обслуживаемым согласно дисциплине «последним пришёл, первым обслужен», — задача, которая была помещена последней, извлекается первой. С точки зрения кэш-памяти это даже может повысить производительность, так как относящиеся к последней задаче данные с большей вероятностью окажутся в кэше, чем данные, относящиеся к предыдущей задаче. К тому же, такая дисциплина прекрасно подходит для алгоритмов типа Quicksort. В предшествующих реализациях каждое обращение к do_sort() помещает элемент в очередь, а затем ждет его. Обрабатывая последний помещенный в очередь элемент первым, мы гарантируем, что блок, необходимый текущему вызову для завершения работы, будет обработан раньше блоков, нужных другим ветвям, а, значит, уменьшается как количество активных задач, так и занятый размер стека. Функция try_steal() извлекает элементы из противоположного по сравнению с try_pop() конца очереди, чтобы минимизировать конкуренцию; в принципе, можно было бы применить технику, обсуждавшуюся в главах 6 и 7, чтобы поддержать одновременные обращения к try_pop() и try_steal().

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

Листинг 9.8. Пул потоков с использованием занимания работ

class thread_pool {

 typedef function_wrapper task_type;

 std::atomic_bool done;

 thread_safe_queue pool_work_queue;

 std::vector > queues;←(1)

 std::vector threads;

 join_threads joiner;

 static thread_local work_stealing_queue* local_work_queue;←(2)

 static thread_local unsigned my_index;

 void worker_thread(unsigned my_index_) {

  my_index = my_index_;

  local_work_queue = queues[my_index].get(); ←(3)

  while (!done) {

   run_pending_task();

  }

 }

 bool pop_task_from_local_queue(task_type& task) {

  return local_work_queue && local_work_queue->try_pop(task);

 }

 bool pop_task_from_pool_queue(task_type& task) {

  return pool_work_queue.try_pop(task);

 }

 bool pop_task_from_other_thread_queue(task_type& task) { ←(4)

  for (unsigned i = 0; i < queues.size(); ++i) {

   unsigned const index = (my_index + i + 1) % queues.size();←(5)

   if (queues[index]->try_steal(task)) {

    return true;

   }

  }

  return false;

 }

public:

 thread_pool() :

  done(false), joiner(threads) {

  unsigned const thread_count =

   std::thread::hardware_concurrency();

  try {

   for (unsigned i = 0; i < thread_count; ++i) {

    queues.push_back(std::unique_ptr (←(6)

     new work_stealing_queue));

    threads.push_back(

     std::thread(&thread_pool::worker_thread, this, i));

   }

  } catch (...) {

   done = true;

   throw;

  }

 }

 ~thread_pool() {

  done = true;

 }

 template

 std::future::type> submit(

  FunctionType f) {

  typedef typename std::result_of::type

   result_type;

  std::packaged_task task(f);

  std::future res(task.get_future());

  if (local_work_queue) {

   local_work_queue->push(std::move(task));

  } else {

   pool_work_queue.push(std::move(task));

  }

  return res;

 }

 void run_pending_task() {

  task_type task;

  if (pop_task_from_local_queue(task) || ←(7)

      pop_task_from_pool_queue (task) || ←(8)

      pop_task_from_other_thread_queue(task)) { ←(9)