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

Ну и как же воспользоваться классом std::condition_variable в примере, упомянутом во введении, — как сделать, чтобы поток, ожидающий работу, спал, пока не поступят данные? В следующем листинге приведён пример реализации с использованием условной переменной.

Листинг 4.1. Ожидание данных с помощью std::condition_variable

std::mutex mut;

std::queue data_queue; ←(1)

std::condition_variable data_cond;

void data_preparation_thread() {

 while (more_data_to_prepare()) {

  data_chunk const data = prepare_data();

  std::lock_guard lk(mut);

  data_queue.push(data);  ←(2)

  data_cond.notify_one(); ←(3)

 }

}

void data_processing_thread() {

 while(true) {

  std::unique_lock lk(mut); ←(4)

  data_cond.wait(

   lk, []{ return !data_queue.empty(); }); ←(5)

  data_chunk data = data_queue.front();

  data_queue.pop();

  lk.unlock(); ←(6)

  process(data);

  if (is_last_chunk(data))

   break;

 }

}

Итак, мы имеем очередь (1), которая служит для передачи данных между двумя потоками. Когда данные будут готовы, поток, отвечающий за их подготовку, помещает данные в очередь, предварительно захватив защищающий ее мьютекс с помощью std::lock_guard. Затем он вызывает функцию-член notify_one() объекта std::condition_variable, чтобы известить ожидающий поток (если таковой существует) (3).

По другую сторону забора находится поток, обрабатывающий данные. Он в самом начале захватывает мьютекс, но с помощью std::unique_lock, а не std::lock_guard (4) — почему, мы скоро увидим. Затем поток вызывает функцию-член wait() объекта std::condition_variable, передавая ей объект-блокировку и лямбда-функцию, выражающую ожидаемое условие (5). Лямбда-функции — это нововведение в С++11, они позволяют записать анонимную функцию как часть выражения и идеально подходят для задания предикатов для таких стандартных библиотечных функций, как wait(). В данном случае простая лямбда-функция []{ return !data_queue.empty(); } проверяет, что очередь data_queue не пуста (вызывая ее метод empty()), то есть что в ней имеются данные для обработки. Подробнее лямбда-функции описаны в разделе А.5 приложения А.

Затем функция wait() проверяет условие (вызывая переданную лямбда-функцию) и возвращает управление, если оно выполнено (то есть лямбда-функция вернула true). Если условие не выполнено (лямбда-функция вернула false), то wait() освобождает мьютекс и переводит поток в состояние ожидания. Когда условная переменная получит извещение, отправленное потоком подготовки данных с помощью notify_one(), поток обработки пробудится, вновь захватит мьютекс и еще раз проверит условие. Если условие выполнено, то wait() вернет управление, причём мьютекс в этот момент будет захвачен. Если же условие не выполнено, то поток снова освобождает мьютекс и возобновляет ожидание. Именно поэтому нам необходим std::unique_lock, а не std::lock_guard — ожидающий поток должен освобождать мьютекс, когда находится в состоянии ожидания, и захватывать его но выходе из этого состояния, a std::lock_guard такой гибкостью не обладает. Если бы мьютекс оставался захваченным в то время, когда поток обработки спит, поток подготовки данных не смог бы захватить его, чтобы поместить новые данные в очередь, а, значит, ожидаемое условие никогда не было бы выполнено.