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

Второй способ — сделать указатель 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;

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_count30 бит, мы ограничили длину поля счетчика 32 битами. В результате мы убиваем сразу двух зайцев: и значение внутреннего счетчика может быть достаточно велико, и вся структура помещается в машинное слово на 32- и 64-разрядных машинах. Очень важно изменять счетчики как единое целое, чтобы избежать гонки. Как это делается, мы покажем чуть ниже. На многих платформах хранение структуры в одном машинном слове повышает шансы на то, что атомарные операции окажутся свободными от блокировок.

При инициализации структуры nodeв поле internal_countзаписывается 0, а в поле external_counters— 2 (4), потому что сразу после добавления нового узла в очередь на него есть две ссылки: из tailи из указателя nextв предыдущем узле. Код самой функции push()похож на приведенный в листинге 7.14 с тем отличием, что перед тем как разыменовывать загруженное из tailзначение, чтобы вызвать compare_exchange_strong()для члена узла data (6), мы вызываем новую функцию increase_external_count()которая увеличивает счетчик (5), а затем функцию free_external_counter()для старого хвоста old_tail (7).

Разобравшись с push(), обратим наши взоры на pop(). В ее коде (см. листинг 7.16) логика подсчета ссылок из реализации pop()в листинге 7.11 комбинируется с логикой извлечения из очереди в листинге 7.13.

Листинг 7.16.Извлечение узла из очереди без блокировок с подсчётом ссылок на tail

template

class lock_free_queue {

private:

struct node {

void release_ref();

};

public:

std::unique_ptr pop() {

counted_node_ptr old_head =

head.load(std::memory_order_relaxed); ← (1)

for (;;) {

increase_external_count(head, old_head);← (2)

node* const ptr = old_head.ptr;

if (ptr == tail.load().ptr) {

ptr->release_ref();← (3)

return std::unique_ptr();

}

if (head.compare_exchange_strong(old_head, ptr->next)) {← (4)

T* const res = ptr->data.exchange(nullptr);