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

Применение схемы подсчета ссылок устраняет эту конкретную гонку, но в push() имеются и другие гонки. Взглянув на переработанную версию push() в листинге 7.14, вы обнаружите ту же ситуацию, что уже встречалась нам в стеке: загрузка атомарного указателя (1) и разыменование этого указателя (2). В промежутке между этими двумя операциями другой поток может изменить указатель (3), что в конечном итоге приведет к освобождению памяти, запятой узлом (в pop()). Если это произойдет раньше, чем мы разыменовываем указатель, то получится неопределенное поведение. Ой! Возникает искушение добавить в tail внешний счетчик, как мы уже поступили для head, однако на каждый узел уже имеется внешний счетчик в указателе next в предыдущем узле очереди. Если хранить два внешних счетчика для одного узла, то потребуется модифицировать схему подсчета ссылок, чтобы не удалить узел преждевременно. Проблему можно решить, подсчитывая число внешних счетчиков в структуре node и уменьшая это число при уничтожении внешнего счетчика (одновременно с прибавлением значения внешнего счетчика к значению внутреннего). Если внутренний счетчик равен нулю, а внешних не осталось, то узел можно удалять. Эту технику я впервые встретил в проекте Джо Сейга (Joe Seigh) Atomic Ptr Plus[16]. В следующем листинге показано, как выглядит push() при использовании такой схемы.

Листинг 7.15. Реализация push() для очереди без блокировок с подсчётом ссылок на tail

template

class lock_free_queue {

private:

 struct node;

 struct counted_node_ptr {

  int external_count;

  node* ptr;

 };

 std::atomic head;

 std::atomic tail;←(1)

 struct node_counter {

  unsigned internal_count:30;

  unsigned external_counters:2;←(2)

 };

 struct node {

  std::atomic data;

  std::atomic count;←(3)

  counted_node_ptr next;

  node() {

   node_counter new_count;

   new_count.internal_count = 0;

   new_count.external_counters = 2;←(4)

   count.store(new_count);

   next.ptr = nullptr;

   next.external_count = 0;

  }

 };

public:

 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;

  counted_node_ptr old_tail = tail.load();

  for (;;) {

   increase_external_count(tail, old_tail);       ←(5)

   T* old_data = nullptr;

   if (old_tail.ptr->data.compare_exchange_strong(←(6)

       old_data, new_data.get())) {

    old_tail.ptr->next = new_next;

    old_tail = tail.exchange(new_next);

    free_external_counter(old_tail);              ←(7)

    new_data.release();

    break;

   }

   old_tail.ptr->release_ref();

  }

 }

};

В листинге 7.15 tail теперь имеет такой же тип atomic, как и head (1), а в структуру node добавлен член count вместо прежнего internal_count (3). Член count сам представляет собой структуру с двумя полями: internal_count и external_counters (2). Под поле external_counters отведено только 2 бита, потому что внешних счетчиков может быть не более двух. Воспользовавшись битовыми полями и отведя под internal_count 30 бит, мы ограничили длину поля счетчика 32 битами. В результате мы убиваем сразу двух зайцев: и значение внутреннего счетчика может быть достаточно велико, и вся структура помещается в машинное слово на 32- и 64-разрядных машинах. Очень важно изменять счетчики как единое целое, чтобы избежать гонки. Как это делается, мы покажем чуть ниже. На многих платформах хранение структуры в одном машинном слове повышает шансы на то, что атомарные операции окажутся свободными от блокировок.