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

Функция wait_and_pop() сложнее, потому что мы должны решить, где поместить ожидание, какой задать предикат и какой мьютекс захватить. Мы ждем условия «очередь не пуста», оно представляется выражением head != tail. Если записать его в таком виде, то придется захватывать и head_mutex, и tail_mutex, но, разбирая код в листинге 6.6, мы уже поняли, что захватывать tail_mutex нужно только для чтения tail, а не для самого сравнения, та же логика применима и здесь. Если записать предикат в виде head != get_tail(), то нужно будет захватить только head_mutex и использовать уже полученную блокировку для защиты data_cond.wait(). Прочий код такой же, как в try_pop().

Второй перегруженный вариант try_pop() и соответствующий ему вариант wait_and_pop() нуждаются в тщательном осмыслении. Если просто заменить возврат указателя std::shared_ptr<>, полученного из old_head, копирующим присваиванием параметру value, то функция перестанет быть безопасной относительно исключений. В этот момент элемент данных уже удален из очереди и мьютекс освобожден, осталось только вернуть данные вызывающей программе. Однако, если копирующее присваивание возбудит исключение (а почему бы и нет?), то элемент данных будет потерян, потому что вернуть его в то же место очереди, где он был, уже невозможно.

Если фактический тип T, которым конкретизируется шаблон, обладает не возбуждающими исключений оператором перемещающего присваивания или операцией обмена (swap), то так поступить можно, но ведь мы ищем общее решение, применимое к любому типу T. В таком случае следует поместить операции, способные возбудить исключения, в защищенную область перед тем, как удалять узел из списка. Это означает, что нам необходим еще один перегруженный вариант pop_head(), который извлекает сохраненное значение до модификации списка.

Напротив, модификация функции empty() тривиальна: нужно просто захватить head_mutex и выполнить проверку head == get_tail() (см. листинг 6.10). Окончательный код очереди приведён в листингах 6.7, 6.8, 6.9 и 6.10.

Листинг 6.7. Потокобезопасная очередь с блокировкой и ожиданием: внутренние данные и интерфейс

template

class threadsafe_queue {

private:

 struct node {

  std::shared_ptr data;

  std::unique_ptr next;

 };

 std::mutex head_mutex;

 std::unique_ptr head;

 std::mutex tail_mutex;

 node* tail;

 std::condition_variable data_cond;

public:

 threadsafe_queue():

  head(new node), tail(head.get()) {}

 threadsafe_queue(const threadsafe_queue& other) = delete;

 threadsafe_queue& operator=(

  const threadsafe_queue& other) = delete;

 std::shared_ptr try_pop();

 bool try_pop(T& value);

 std::shared_ptr wait_and_pop();

 void wait_and_pop(T& value);

 void push(T new_value);

 void empty();

};

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

Листинг 6.8. Потокобезопасная очередь с блокировкой и ожиданием: добавление новых значений

template

void threadsafe_queue::push(T new_value) {

 std::shared_ptr new_data(

  std::make_shared(std::move(new_value)));

 std::unique_ptr p(new node);

 {

  std::lock_guard tail_lock(tail_mutex);

  tail->data = new_data;

  node* const new_tail = p.get();

  tail->next = std::move(p);

  tail = new_tail;

 }

 data_cond.notify_one();

}

Как уже отмечалось, вся сложность сосредоточена в части pop. В листинге ниже показана реализация функции-члена wait_and_pop() и относящихся к ней вспомогательных функций.

Листинг 6.9. Потокобезопасная очередь с блокировкой и ожиданием: wait_and_pop

template

class threadsafe_queue {

private:

 node* get_tail() {

  std::lock_guard tail_lock(tail_mutex);

  return tail;

 }

 std::unique_ptr pop_head() {←(1)

  std::unique_ptr old_head = std::move(head);

  head = std::move(old_head->next);

  return old_head;

 }