А вот чего мы пока не рассмотрели, так это ожидание поступления входных данных из других потоков. Наш потокобезопасный стек просто возбуждает исключение при попытке извлечения из пустого стека. Поэтому если один поток хочет дождаться, пока другой поток поместит в стек какие-то данные (а это, собственно, и есть основное назначение потокобезопасного стека), то должен будет раз за разом пытаться извлечь значение, повторяя попытку в случае исключения. Это приводит лишь к бесцельной трате процессорного времени на проверку; более того, такая повторяющаяся проверка может замедлить работу программы, поскольку не дает выполняться другим потокам. Нам необходим какой-то способ, который позволил бы одному потоку ждать завершения операции в другом потоке, не потребляя процессорное время. В главе 4, которая опирается на рассмотренные выше средства защиты разделяемых данных, мы познакомимся с различными механизмами синхронизации операций между потоками в С++, а в главе 6 увидим, как с помощью этих механизмов можно строить более крупные структуры данных, допускающие повторное использование.
Глава 4.Синхронизация параллельных операций
В этой главе:
Ожидание события.
Ожидание однократного события с будущими результатами
Ожидание с ограничением по времени.
Использование синхронизации операций для упрощения программы.
В предыдущей главе мы рассмотрели различные способы защиты данных, разделяемых между потоками. Но иногда требуется не только защитить данные, но и синхронизировать действия, выполняемые в разных потоках. Например, возможно, что одному потоку перед тем как продолжить работу, нужно дождаться, пока другой поток завершит какую-то операцию. В общем случае, часто возникает ситуация, когда поток должен ожидать какого-то события или истинности некоторого условия. Конечно, это можно сделать, периодически проверяя разделяемый флаг «задача завершена» или что-то в этом роде, но такое решение далеко от идеала. Необходимость в синхронизации операцийнастолько распространенный сценарий, что в стандартную библиотеку С++ включены специальные механизмы для этой целиусловные переменные и будущие результаты (future).
В этой главе мы рассмотрим, как реализуется ожидание событий с помощью условных переменных и будущих результатов и как ими можно воспользоваться, чтобы упростить синхронизацию операций.
4.1. Ожидание события или иного условия
Представьте, что вы едете на поезде ночью. Чтобы не пропустить свою станцию, можно не спать всю ночь и читать названия всех пунктов, где поезд останавливается. Так вы, конечно, не проедете мимо, но сойдете с поезда сильно уставшим. Есть и другой способзаранее посмотреть в расписании, когда поезд прибывает в нужный вам пункт, поставить будильник и улечься спать. Так вы тоже свою остановку не пропустите, но если поезд задержится в пути, то проснётесь слишком рано. И еще одноесли в будильнике сядут батарейки, то вы можете проспать и проехать мимо нужной станции. В идеале хотелось бы, чтобы кто-то или что-то разбудило вас, когда поезд подъедет к станции,не раньше и не позже.
Какое отношение всё это имеет к потокам? Самое непосредственноеесли один поток хочет дождаться, когда другой завершит некую операцию, то может поступить несколькими способами. Во-первых, он может просто проверять разделяемый флаг (защищенный мьютексом), полагая, что второй поток поднимет этот флаг, когда завершит свою операцию. Это расточительно но двум причинам: на опрос флага уходит процессорное время, и мьютекс, захваченный ожидающим потоком, не может быть захвачен никаким другим потоком. То и другое работает против ожидающего потока, поскольку ограничивает ресурсы, доступные потоку, которого он так ждет, и даже не дает ему возможность поднять флаг, когда работа будет завершена. Это решение сродни бодрствованию всю ночь, скрашиваемому разговорами с машинистом: он вынужден вести поезд медленнее, потому что вы его постоянно отвлекаете, и, значит, до пункта назначения вы доберетесь позже. Вот и ожидающий поток потребляет ресурсы, которые пригодились бы другим потокам, в результате чего ждет дольше, чем необходимо.
Второй вариантзаставить ожидающий поток спать между проверками с помощью функции std::this_thread::sleep_for() (см. раздел 4.3):
bool flag;
std::mutex m;
void wait_for_flag() {
std::unique_lock<std::mutex> lk(m); (1) Освободить мьютекс
while (!flag) {
lk.unlock(); (2) Спать 100 мс
std::this_thread::sleep_for(std::chrono::milliseconds(100));
lk.lock(); (3) Снова захватить мьютекс
}
}
В этом цикле функция освобождает мьютекс (1) перед тем, как заснуть (2), и снова захватывает его, проснувшись, (3), оставляя другому потоку шанс захватить мьютекс и поднять флаг.
Это уже лучше, потому что во время сна поток не расходует процессорное время. Но трудно выбрать подходящий промежуток времени. Если он слишком короткий, то поток все равно впустую тратит время на проверку; если слишком длинныйто поток будет спать и после того, как ожидание завершилось, то есть появляется ненужная задержка. Редко бывает так, что слишком длительный сон прямо влияет на работу программу, но в динамичной игре это может привести к пропуску кадров, а в приложении реального временик исчерпанию выделенного временного кванта.
Третийи наиболее предпочтительный - способ состоит в том, чтобы воспользоваться средствами из стандартной библиотеки С++, которые позволяют потоку ждать события. Самый простой механизм ожидания события, возникающего в другом потоке (например, появления нового задания в упоминавшемся выше конвейере), дают условные переменные. Концептуально условная переменная ассоциирована с каким-то событием или иным условием, причём один или несколько потоков могут ждать, когда это условие окажется выполненным. Если некоторый поток решит, что условие выполнено, он может известить об этом один или несколько потоков, ожидающих условную переменную, в результате чего они возобновят работу.
4.1.1. Ожидание условия с помощью условных переменных
Стандартная библиотека С++ предоставляет не одну, а две реализации условных переменных: std::condition_variable и std::condition_variable_any. Оба класса объявлены в заголовке <condition_variable>. В обоих случаях для обеспечения синхронизации необходимо взаимодействие с мьютексом; первый класс может работать только с std::mutex, второй с любым классом, который отвечает минимальным требованиям к «мьютексоподобию», отсюда и суффикс _any. Поскольку класс std::condition_variable_any более общий, то его использование может обойтись дороже с точки зрения объема потребляемой памяти, производительности и ресурсов операционной системы. Поэтому, если дополнительная гибкость не требуется, то лучше ограничиться классом std::condition_variable.
Ну и как же воспользоваться классом std::condition_variable в примере, упомянутом во введении,как сделать, чтобы поток, ожидающий работу, спал, пока не поступят данные? В следующем листинге приведён пример реализации с использованием условной переменной.
Листинг 4.1. Ожидание данных с помощью std::condition_variable
std::mutex mut;
std::queue<data_chunk> 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<std::mutex> lk(mut);
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, удерживать мьютекс дольше необходимого неразумно.
Применение очереди для передачи данных между потоками (как в листинге 4.1) весьма распространенный прием. При правильной реализации синхронизацию можно ограничить только самой очередью, что уменьшает количество потенциальных проблем и состояний гонки. Поэтому покажем, как на основе листинга 4.1 построить обобщенную потокобезопасную очередь.
4.1.2. Потокобезопасная очередь на базе условных переменных
Приступая к проектированию обобщенной очереди, стоит потратить некоторое время на обдумывание того, какие понадобятся операции. Именно так мы подходили к разработке потокобезопасного стека в разделе 3.2.3. Возьмем в качестве образца адаптер контейнера std::queue<> из стандартной библиотеки С++, интерфейс которого показан в листинге ниже.
Листинг 4.2. Интерфейс класса std::queue
template <class T, class Container = std::deque<T>>
class queue {
public:
explicit queue(const Container&);
explicit queue(Container&& = Container());
template <class Alloc> explicit queue(const Alloc&);
template <class Alloc> queue(const Container&, const Alloc&);
template <class Alloc> queue(Container&&, const Alloc&);
template <class Alloc> queue(queue&&, const Alloc&);
void swap(queue& q);
bool empty() const;
size_type size() const;
T& front();
const T& front() const;
T& back();
const T& back() const;
void push(const T& x);
void push(T&& x);
void pop();
template <class... Args> void emplace(Args&&... args);
};
Если не обращать внимания на конструирование, присваивание и обмен, то останется три группы операций: опрос состояния очереди в целом (empty() и size()), опрос элементов очереди (front() и back()) модификация очереди (push(), pop() и emplace()). Ситуация аналогична той, что мы видели в разделе 3.2.3 для стека, поэтому возникают те же внутренне присущие интерфейсупроблемы с гонкой. Следовательно, front() и pop() необходимо объединить в одной функцииточно так же, как мы постудили с top() и pop() в случае стека. Но в коде в листинге 4.1 есть дополнительный нюанс: если очередь используется для передачи данных между потоками, то поток-получатель часто будет ожидать поступления данных. Поэтому включим два варианта pop(): try_pop() пытается извлечь значение из очереди, но сразу возвращает управление (с указанием ошибки), если в очереди ничего не было, a wait_and_pop() ждет, когда появятся данные. Взяв за образец сигнатуры функций из примера стека, представим интерфейс в следующем виде:
Листинг 4.3. Интерфейс класса threadsafe_queue
#include <memory>
template<typename T>
class threadsafe_queue {
public:
threadsafe_queue();
threadsafe_queue(const threadsafe_queue&);
threadsafe_queue& operator=(
const threadsafe_queue&) = delete; Для простоты
void push(T new_value); запрещаем присваивание
bool try_pop(T& value); (1)
std::shared_ptr<T> try_pop(); (2)
void wait_and_pop(T& value);
std::shared_ptr<T> wait_and_pop();
bool empty() const;
};
Как и в случае стека, мы для простоты уменьшили число конструкторов и запретили присваивание. И, как и раньше, предлагаем по два варианта функций try_pop() и wait_for_pop(). Первый перегруженный вариант try_pop() (1) сохраняет извлеченное значение в переданной по ссылке переменной, а возвращаемое значение использует для индикации ошибки: оно равно true, если значение получено, и falseв противном случае (см. раздел А.2). Во втором перегруженном варианте (2) так поступить нельзя, потому что возвращаемое значениеэто данные, извлеченные из очереди. Однако же можно возвращать указатель NULL, если в очереди ничего не оказалось.