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

Проблемы начинаются, когда несколько потоков вызывают push() или pop() одновременно. Сначала рассмотрим push(). Если два потока одновременно вызывают push(), то оба выделяют память для нового фиктивного узла (3), оба читают одно и то же значение tail (4) и, следовательно, оба изменяют данные-члены data и next одного и того же узла (5), (6). А это уже гонка за данными!

Аналогичные проблемы возникают в pop_head(). Если два потока вызывают эту функцию одновременно, то оба читают одно и то же значение head, и оба перезаписывают старое значение одним и тем же указателем next. Оба потока теперь думают, что получили один и тот же узел, — прямой путь к катастрофе. Мы должны не только сделать так, чтобы лишь один поток извлекал данный элемент, но и позаботиться о том, чтобы другие потоки могли безопасно обращаться к члену next узла, который прочитали из head. Это точно та же проблема, с которой мы сталкивались при написании pop() для свободного от блокировок стека, поэтому и любое из предложенных тогда решений можно применить здесь.

Итак, проблему pop() можно считать решенной, но как быть с push()? Здесь трудность заключается в том, что для получения требуемого отношения происходит-раньше между push() и pop() мы должны заполнить поля фиктивного узла до обновления tail. Но это означает, что одновременные вызовы push() конкурируют за те же самые данные, так как был прочитал один и тот же указатель tail.

Один из способов — добавить фиктивный узел между реальными. Тогда единственной частью текущего узла tail, нуждающейся в обновлении, будет указатель next, который, следовательно, можно было бы сделать атомарным. Если потоку удалось записать в next указатель на свой новый узел вместо nullptr, то он успешно добавил узел; в противном случае ему придется начать сначала и снова прочитать tail. Это потребует небольшого изменения в pop() — нужно будет игнорировать узлы с нулевым указателем на данные и возвращаться в начало цикла. Недостаток этого решения в том, что при каждом вызове pop() придется как правило исключать из списка два узла и производить в два раза больше операций выделения памяти.

Второй способ — сделать указатель data атомарным и устанавливать его с помощью операции сравнения с обменом. Если она завершится успешно, то мы получили свой хвостовой узел и можем безопасно записать в next указатель на наш новый узел, а затем обновить tail. Если же сравнение с обменом завершается неудачно, потому что другой поток успел сохранить данные, мы возвращаемся в начало цикла, заново читаем tail и пробуем снова. Если атомарные операции над std::shared_ptr<> свободны от блокировок, то дело сделано. Если нет, нужна альтернатива. Можно, например, заставить pop() возвращать std::unique_ptr<> (в конце концов, это ведь единственная ссылка на объект) и сохранять данные в очереди в виде простого указателя. Тогда его можно было бы хранить как std::atomic и впоследствии обновлять с помощью compare_exchange_strong(). Если воспользоваться для поддержки нескольких потоков в pop() схемой подсчета ссылок из листинга 7.11, то push() будет выглядеть следующим образом.

Листинг 7.14. Первая (неудачная) попытка переработки push()

void push(T new_value) {

 std::unique_ptr new_data(new T(new_value));

 counted_node_ptr new_next;

 new_next.ptr = new node;

 new_next.external_count = 1;

 for (;;) {

  node* const old_tail = tail.load();←(1)

  T* old_data = nullptr;

  if (old_tail->data.compare_exchange_strong(

   old_data, new_data.get())) {      ←(2)

   old_tail->next = new_next;

   tail.store(new_next.ptr);         ←(3)

   new_data.release();

   break;

  }

 }

}