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

  std::unique_lock lk(mut);
  data_cond.wait(lk, [this]{return !data_queue.empty();});
  value = data_queue.front();
  data_queue.pop();
 }
};

threadsafe_queue data_queue; ←(1)

void data_preparation_thread() {
 while (more_data_to_prepare()) {
  data_chunk const data = prepare_data();

  data_queue.push(data); ←(2)

 }
}

void data_processing_thread() {
 while (true) {
  data_chunk data;

  data_queue.wait_and_pop(data); ←(3)

  process(data);
  if (is_last_chunk(data))
   break;
 }
}

Теперь мьютекс и условная переменная находятся в экземпляре threadsafe_queue, поэтому не нужно ни отдельных переменных (1), ни внешней синхронизации при обращении к функции push()(2). Кроме того, wait_and_pop() берет на себя заботу об ожидании условной переменной (3).

Второй перегруженный вариант wait_and_pop() тривиален, а остальные функции можно почти без изменений скопировать из кода стека в листинге 3.5. Ниже приведена окончательная реализация.

Листинг 4.5. Полное определение класса потокобезопасной очереди на базе условных переменных

#include
#include
#include
#include

template
class threadsafe_queue {
private:

 mutable std::mutex mut;←(1) Мьютекс должен быть изменяемым

 std::queue data_queue;
 std::condition_variable data_cond;

public:
 threadsafe_queue() {}
 threadsafe_queue(threadsafe_queue const& other) {
  std::lock_guard lk(other.mut);
  data_queue = other.data_queue;
 }

 void push(T new_value) {
  std::lock_guard lk(mut);
  data_queue.push(new_value);
  data_cond.notify_one();
 }

 void wait_and_pop(T& value) {
  std::unique_lock lk(mut);
  data_cond.wait(lk, [this]{ return !data_queue.empty(); });
  value = data_queue.front();
  data_queue.pop();
 }

 std::shared_ptr wait_and_pop() {
  std::unique_lock lk(mut);
  data_cond.wait(lk, [this]{ return !data_queue.empty(); });
  std::shared_ptr
   res(std::make_shared(data_queue.front()));
  data_queue.pop();
  return res;
 }

 bool try_pop(T& value) {
  std::lock_guard lk(mut);
  if (data_queue.empty())
   return false;
  value = data_queue.front();
  data_queue.pop();
  return true;
 }

 std::shared_ptr try_pop() {
  std::lock_guard lk(mut);
  if (data_queue.empty())
   return std::shared_ptr();
  std::shared_ptr
   res(std::make_shared(data_queue.front()));
  data_queue.pop();
  return res;
 }

 bool empty() const {
  std::lock_guard lk(mut);
  return data_queue.empty();
 }
};

Хотя empty() — константная функция-член, а параметр копирующего конструктора — const-ссылка, другие потоки могут хранить неконстантные ссылки на объект и вызывать изменяющие функции-члены, которые захватывают мьютекс. Поэтому захват мьютекса — это изменяющая операция, следовательно, член mut необходимо пометить как mutable(1), чтобы его можно было захватить в функции empty() и в копирующем конструкторе.

Условные переменные полезны и тогда, когда есть несколько потоков, ожидающих одного события. Если потоки используются для разделения работы и, следовательно, на извещение должен реагировать только один поток, то применима точно такая же структура программы, как в листинге 4.1; нужно только запустить несколько потоков обработки данных. При поступлении новых данных функция notify_one() разбудит только один поток, который проверяет условие внутри wait(), и этот единственный поток вернет управление из wait() (в ответ на помещение нового элемента в очередь data_queue). Заранее нельзя сказать, какой поток получит извещение и есть ли вообще ожидающие потоки (не исключено, что все они заняты обработкой ранее поступивших данных).

Альтернативный сценарий — когда несколько потоков ожидают одного события, и отреагировать должны все. Так бывает, например, когда инициализируются разделяемые данные, и все работающие с ними потоки должны ждать, пока инициализация завершится (хотя для этого случая существуют более подходящие механизмы, см. раздел 3.3.1 главы 3), или когда потоки должны ждать обновления разделяемых данных, например, в случае периодической повторной инициализации. В таких ситуациях поток, отвечающий за подготовку данных, может вызвать функцию-член notify_all() условной переменной вместо notify_one(). Эта функция извещает все потоки, ожидающие внутри функции wait(), о том, что они должны проверить ожидаемое условие.

Если ожидающий поток собирается ждать условия только один раз, то есть после того как оно станет истинным, он не вернется к ожиданию той же условной переменной, то лучше применить другой механизм синхронизации. В особенности это относится к случаю, когда ожидаемое условие — доступность каких-то данных. Для такого сценария больше подходят так называемые будущие результаты (future).

4.2. Ожидание одноразовых событий с помощью механизма будущих результатов

Предположим, вы летите самолетом в отпуск за границу. Вы приехали в аэропорт, прошли регистрацию и прочие процедуры, но должны ждать объявления о посадке — быть может, несколько часов. Можно, конечно, найти себе занятие — например, почитать книжку, побродить в Интернете, поесть в кафе за бешеные деньги, но суть от этого не меняется: вы ждете сигнала о том, что началась посадка в самолет. И есть еще одна особенность — данный рейс вылетает всего один раз; в следующий отпуск вы будете ждать посадки на другой рейс.