data_queue.push(data); ←
(2)
data_cond.notify_one(); ←
(3)
}
}
void data_processing_thread() {
while(true) {
std::unique_lock<std::mutex> 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, удерживать мьютекс дольше необходимого неразумно.