C++17 STL Стандартная библиотека шаблонов

Сторона-потребитель, ждущая выполнения конкретного условия, выглядит так. Поток-потребитель проходит в цикле по блоку, который изначально блокирует мьютекс mut с помощью unique_lock. Затем вызывает cv.wait:

while (!finished) {

  unique_lock l {mut};

  cv.wait(l, [] { return !q.empty() || finished; });

  while (!q.empty()) {

    // consume

  }

}

Данный код чем-то похож на следующий эквивалентный код. Вскоре мы рассмотрим, почему эти фрагменты не похожи друг на друга:

while (!finished) {

  unique_lock l {mut};

  while (q.empty() && !finished) {

    l.unlock();

    l.lock();

  }

  while (!q.empty()) {

    // consume

  }

}

Это значит, что сначала мы получили блокировку, а затем проверили, с каким сценарием сейчас работаем.

1. Есть ли элементы, которые можно использовать? В таком случае сохраняем блокировку, потребляем эти элементы, возвращаем блокировку и начинаем сначала.

2. В противном случае, если элементов для потребления нет, но производитель все еще жив, то возвращаем мьютекс с целью дать ему шанс добавить новые элементы в очередь. Далее снова пытаемся получить блокировку в надежде, что ситуация изменится и мы перейдем к ситуации 1.

Реальная причина, по которой строка cv.wait не эквивалентна конструкции while (q.empty() && ...), заключается в том, что мы не можем просто пройти по циклу l.unlock(); l.lock();. Отсутствие активности потока-производителя в течение какого-то промежутка времени приводит к постоянным блокировкам/ разблокировкам мьютекса, что не имеет смысла, поскольку мы впустую тратим циклы процессора.

Выражение наподобие cv.wait(lock, predicate) будет ждать до тех пор, пока вызов predicate() не вернет значение true. Но это не делается путем постоянной блокировки/разблокировки. Чтобы возобновить поток, который блокируется вызовом wait объекта condition_variable, другой поток должен вызывать методы notify_one() или notify_all() для одного объекта. Только тогда ожидающий поток будет возобновлен и проверит условие predicate(). (Последнее действительно и для нескольких потоков.)

Положительный момент таков: после вызова wait, проверившего предикат так, словно поступил внезапный сигнал к пробуждению, поток будет снова мгновенно приостановлен. Это значит, что мы не навредим рабочему потоку программы (но, возможно, пострадает производительность), если добавим в код слишком много вызовов notify.

На стороне производителя мы просто вызвали cv.notify_all() после того, как производитель добавил новый элемент в очередь, а также вслед за тем, как создал последний элемент и установил значение флага finished, равное true. Этого было достаточно для того, чтобы направить потребителя. 

Реализуем идиому «несколько производителей/потребителей» с помощью std::condition_variable

Возьмем задачу из прошлого примера и усложним ее: создадим несколько производителей и несколько потребителей. Кроме того, укажем, что размер очереди не должен превышать определенное значение.

Таким образом, нужно приостанавливать не только потребителей, при отсутствии в очереди элементов, но и производителей, если элементов в ней слишком много.

Мы увидим, как решить эту проблему с помощью нескольких объектов типа std::condition_variable, а также воспользуемся ими несколько иным образом, нежели это было показано в предыдущем примере.

Как это делается

В данном примере мы реализуем программу, похожую на программу из предыдущего примера, но в этот раз у нас будет несколько производителей и несколько потребителей.

1. Сначала включим все необходимые заголовочные файлы и объявим об использовании пространств имен std и chrono_literals:

#include

#include

#include

#include

#include

#include

#include

#include

#include

using namespace std;

using namespace chrono_literals;

2. Затем реализуем синхронизированную вспомогательную функцию для вывода сообщений на экран, показанную в другом примере этой главы, поскольку будем выводить множество сообщений на конкурентной основе:

struct pcout : public stringstream {

  static inline mutex cout_mutex;

  ~pcout() {

    lock_guard l {cout_mutex};

    cout << rdbuf();

  }

};