Энтони Уильямс - Параллельное программирование на С++ в действии. Практика разработки многопоточных программ

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

Второй способ — сделать указатель dataатомарным и устанавливать его с помощью операции сравнения с обменом. Если она завершится успешно, то мы получили свой хвостовой узел и можем безопасно записать в nextуказатель на наш новый узел, а затем обновить tail. Если же сравнение с обменом завершается неудачно, потому что другой поток успел сохранить данные, мы возвращаемся в начало цикла, заново читаем tailи пробуем снова. Если атомарные операции над s td::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;

}

}

}

Применение схемы подсчета ссылок устраняет эту конкретную гонку, но в push()имеются и другие гонки. Взглянув на переработанную версию push()в листинге 7.14, вы обнаружите ту же ситуацию, что уже встречалась нам в стеке: загрузка атомарного указателя (1)и разыменование этого указателя (2). В промежутке между этими двумя операциями другой поток может изменить указатель (3), что в конечном итоге приведет к освобождению памяти, запятой узлом (в pop()). Если это произойдет раньше, чем мы разыменовываем указатель, то получится неопределенное поведение. Ой! Возникает искушение добавить в tailвнешний счетчик, как мы уже поступили для head, однако на каждый узел уже имеется внешний счетчик в указателе nextв предыдущем узле очереди. Если хранить два внешних счетчика для одного узла, то потребуется модифицировать схему подсчета ссылок, чтобы не удалить узел преждевременно. Проблему можно решить, подсчитывая число внешних счетчиков в структуре nodeи уменьшая это число при уничтожении внешнего счетчика (одновременно с прибавлением значения внешнего счетчика к значению внутреннего). Если внутренний счетчик равен нулю, а внешних не осталось, то узел можно удалять. Эту технику я впервые встретил в проекте Джо Сейга (Joe Seigh) Atomic Ptr Plus [16] Atomic Ptr Plus Project, http://atomic-ptr-plus.sourceforge.net/. . В следующем листинге показано, как выглядит 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;