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

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);

free_external_counter(old_head);← (5)

return std::unique_ptr(res);

}

ptr->release_ref();

}

}

};

Все начинается с загрузки значения old_headперед входом в цикл (1)и до увеличения внешнего счетчика в загруженном значении (2). Если узел headсовпадает с tail, то можно освободить ссылку (3)и вернуть нулевой указатель, потому что очередь пуста. Если же в очереди есть данные, то мы пытаемся заявить на них свои права с помощью compare_exchange_strong()(4). Как и в случае стека в листинге 7.11, мы при этом сравниваем внешний счетчик и указатель как единое целое; если хотя бы один из них изменился, то мы должны вернуться в начало цикла, освободив предварительно ссылку 6. Если обмен завершился удачно, то мы получили в свое распоряжение данные в узле, поэтому можем вернуть их вызывающей программе, освободив предварительно внешний счетчик ссылок на извлеченный узел (5). После того как оба внешних счетчика освобождены, а внутренний счетчик обратился в нуль, сам узел можно удалять. Вспомогательные функции подсчета ссылок приведены в листингах 7.17, 7.18 и 7.19.

Листинг 7.17.Освобождение ссылки на узел в очереди без блокировок

template

class lock_free_queue {

private:

struct node {

void release_ref() {

node_counter old_counter =

count.load(std::memory_order_relaxed);

node_counter new_counter;

do {