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

В одном из возможных способов используется не один, а два счетчика ссылок — внутренний и внешний. Их сумма равна общему количеству ссылок на узел. Внешний счетчик хранится вместе с указателем на узел и увеличивается при каждом чтении этого указателя. Когда читатель закапчивает работу с узлом, он уменьшает его внутренний счетчик. Таким образом, по завершении простой операции чтения указателя внешний счетчик увеличится на единицу, а внутренний уменьшится на единицу.

Когда необходимость в связи между внешним счетчиком и указателем отпадает (то есть узел невозможно получить из области памяти, доступной другим потокам), внутренний счетчик увеличивается на величину внешнего минус 1, а внешний отбрасывается. Если внутренний счетчик обратился в нуль, значит, никаких ссылок на узел извне не осталось и его можно удалять. Для обновления разделяемых данных по-прежнему необходимо применять атомарные операции. Теперь рассмотрим реализацию свободного от блокировок стека, в которой эта техника используется для гарантии того, что узлы освобождаются, только когда это безопасно.

В листинге ниже показала внутренняя структура данных и реализация функции push() — простая и элегантная.

Листинг 7.10.Помещение узла в свободный от блокировок стек с разделённым счётчиком ссылок

template

class lock_free_stack {

private:

struct node;

struct counted_node_ptr {← (1)

int external_count;

node* ptr;

};

struct node {

std::shared_ptr data;← (2)

std::atomic internal_count;

counted_node_ptr next; ← (3)

node(T const& data_) :

data(std::make_shared(data_)), internal_count(0) {}

};

std::atomic head;← (4)

public:

~lock_free_stack() {

while(pop());

}

void push(T const& data) {← (5)

counted_node_ptr new_node;

new_node.ptr = new node(data);

new_node.external_count = 1;

new_node.ptr->next = head.load();

while(

!head.compare_exchange_weak(new_node.ptr->next, new_node));

}

};

Обратите внимание, что внешний счетчик хранится вместе с указателем на узел в структуре counted_node_ptr (1). Эта структура затем используется для представления указателя nextв структуре node (3), где хранится также внешний счетчик (2). Поскольку counted_node_ptrопределена как struct, то ее можно использовать в шаблоне std::atomic<>для представления головы списка head (4).

На платформах, где поддерживается операция сравнения и обмена двойного слова, размер этой структуры достаточно мал для того, чтобы тип данных std::atomicбыл свободен от блокировок. Если вы работаете на другой платформе, то лучше пользоваться вариантом std::shared_ptr<>, приведенным в листинге 7.9, потому что в std::atomic<>для гарантирования атомарности используется мьютекс, когда тип слишком велик и с помощью машинных команд обеспечить атомарность невозможно (а, следовательно, ваш алгоритм, якобы «свободный от блокировок», на самом теле таковым не является). Можно поступить и по-другому — если вы готовы ограничить размер счетчика и знаете, что на данной платформе в указателе есть неиспользуемые биты (например, потому что адресное пространство представлено 48 битами, а под указатель отводится 64 бита), то счетчик можно хранить в незанятых битах указателя, и тогда оба поля поместятся в одно машинное слово. Но для таких трюков нужно хорошо знать особенности платформы, так что в этой книге мы их обсуждать не будем.

Функция push()относительно проста (5). Мы конструируем объект counted_node_ptr, ссылающийся на только что выделенный узел nodeс ассоциированными данными, и в поле nextузла nodeзаписываем текущее значение head. Затем с помощью compare_exchange_weak()устанавливаем новое значение head, как и раньше. Внутренний счетчик internal_countинициализируется нулем, а внешний external_count— единицей. Поскольку узел новый, на него пока существует только одна внешняя ссылка (из самого указателя head).

Как обычно, все сложности сосредоточены в реализации pop(), показанной в следующем листинге.

Листинг 7.11.Выталкивание узла из свободного от блокировок стека с разделённым счётчиком ссылок

template

class lock_free_stack {

private:

void increase_head_count(counted_node_ptr& old_counter) {

counted_node_ptr new_counter;

do {

new_counter = old_counter;

++new_counter.external_count;