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

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

Стандартная библиотека С++ предоставляет не одну, а две реализации условных переменных: std::condition_variableи std::condition_variable_any. Оба класса объявлены в заголовке . В обоих случаях для обеспечения синхронизации необходимо взаимодействие с мьютексом; первый класс может работать только с std::mutex, второй — с любым классом, который отвечает минимальным требованиям к «мьютексоподобию», отсюда и суффикс _any. Поскольку класс std::condition_variable_anyболее общий, то его использование может обойтись дороже с точки зрения объема потребляемой памяти, производительности и ресурсов операционной системы. Поэтому, если дополнительная гибкость не требуется, то лучше ограничиться классом std::condition_variable.

Ну и как же воспользоваться классом 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такой гибкостью не обладает. Если бы мьютекс оставался захваченным в то время, когда поток обработки спит, поток подготовки данных не смог бы захватить его, чтобы поместить новые данные в очередь, а, значит, ожидаемое условие никогда не было бы выполнено.

В листинге 4.1 используется простая лямбда-функция (5), которая проверяет, что очередь не пуста. Однако с тем же успехом можно было бы передать любую функцию или объект, допускающий вызов. Если функция проверки условия уже существует (быть может, она сложнее показанного в примере простенького теста), то передавайте ее напрямую — нет никакой необходимости обертывать ее лямбда-функцией. Внутри wait()условная переменная может проверять условие многократно, но всякий раз это делается после захвата мьютекса, и, как только функция проверки условия вернет true(и лишь в этом случае), wait()возвращает управление вызывающей программе. Ситуация, когда ожидающий поток захватывает мьютекс и проверяет условие не в ответ на извещение от другого потока, называется ложным пробуждением (spurious wake). Поскольку количество и частота ложных пробуждений по определению недетерминированы, нежелательно использовать для проверки условия функцию с побочными эффектами. В противном случае будьте готовы к тому, что побочный эффект может возникать более одного раза.

Присущая std::unique_lockвозможность освобождать мьютекс используется не только при обращении к wait(), но и непосредственно перед обработкой поступивших данных (6). Обработка может занимать много времени, а, как было отмечено в главе 3, удерживать мьютекс дольше необходимого неразумно.

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