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

Листинг 7.6.Реализация функции pop()с помощью указателей опасности

std::shared_ptr pop() {

std::atomic& hp =

get_hazard_pointer_for_current_thread();

node* old_head = head.load();

do {

node* temp; (1) Цикл, пока указатель

do ←┤ опасности не установлен

{ │ на head

temp = old_head;

hp.store(old_head);

old_head = head.load();

} while (old_head != temp);

}

while (old_head &&

!head.compare_exchange_strong(old_head, old_head->next))

hp.store(nullptr);← (2) Закончив, очищаем указатель опасности

std::shared_ptr res;

if (old_head) {

res.swap(old_head->data);

if (outstanding_hazard_pointers_for(old_head))←┐ Прежде чем

{ ├ (3) удалять узел,

reclaim_later(old_head); │ проверяем, ← (4)

} │ нет ли ссы-

else │ лающихся на

{ │ него указате-

│ лей опасности

delete old_head; ← (5)

}

delete_nodes_with_no_hazards();← (6)

}

return res;

}

Начнём с того, что мы перенесли цикл, в котором устанавливается указатель опасности, во внешний цикл, где перечитывается old_head, если операция сравнения с обменом завершается неудачно (1). Здесь мы используем функцию compare_exchange_strong(), потому что фактическая работа делается внутри цикла while: ложный отказ в compare_exchange_weak()привел бы к ненужному сбросу указателя опасности. Таким образом, гарантируется, что указатель опасности установлен перед разыменованием old_head. Заявив свои права на узел, мы можем очистить указатель опасности (2). Получив узел в свое распоряжение, мы должны проверить, не ссылаются ли на него указатели опасности, принадлежащие другим потокам (3). Если это так, то удалять узел пока нельзя, а нужно поместить его в список ожидающих (4); в противном случае узел можно удалять немедленно (5). Наконец, мы добавили вызов функции, в которой проверяется, существуют ли узлы, для которых мы ранее вызывали reclaim_later(). Если не осталось указателей опасности, ссылающихся на эти узлы, то мы можем спокойно удалить их (6). Те же узлы, на которые еще ссылается хотя бы один указатель опасности, остаются в списке и будут проверены следующим потоком, вызвавшим pop().

Разумеется, в новых функциях — get_hazard_pointer_for_current_thread(), reclaim_later(), outstanding_hazard_pointers_for()и delete_nodes_with_no_hazards()— скрыта масса деталей, поэтому отдёрнем занавес и посмотрим, как они работают.

Как именно в функции get_hazard_pointer_for_current_thread()выделяется память для принадлежащих потокам указателей опасности, несущественно для логики программы (хотя, как будет показано ниже, может влиять на эффективность). Поэтому пока ограничимся простой структурой: массивом фиксированного размера, в котором хранятся пары (идентификатор потока, указатель). Функция get_hazard_pointer_for_current_thread()ищет в этом массиве первую свободную позицию и записывает в поле ID идентификатор текущего потока. Когда поток завершается, эта позиция освобождается — в поле ID заносится сконструированное по умолчанию значение std::thread::id(). Этот алгоритм показан в следующем листинге.

Листинг 7.7.Простая реализация функции get_hazard_pointer_for_current_thread

unsigned const max_hazard_pointers = 100;

struct hazard_pointer {

std::atomic id;

std::atomic pointer;

};

hazard_pointer hazard_pointers[max_hazard_pointers];

class hp_owner {

hazard_pointer* hp;

public:

hp_owner(hp_owner const&) = delete;

hp_owner operator=(hp_owner const&) = delete;

hp_owner() :

hp(nullptr) {

for (unsigned i = 0; i < max_hazard_pointers; ++i) {

std::thread::id old_id;

if (

hazard_pointers[i].

id.compare_exchange_strong( ←┐

old_id, std::this_thread::get_id()))│ Пытаемся заявить

{ │ права на владе-

hp = &hazard_pointers[i]; │ ние указателем

break; │ опасности

}

}

if (!hp) {← (1)

throw std::runtime_error("No hazard pointers available");

}

}

std::atomic& get_pointer() {

return hp->pointer;

}

~hp_owner() {← (2)

hp->pointer.store(nullptr);

hp->id.store(std::thread::id());

}

};

std::atomic& get_hazard_pointer_for_current_thread()← (3)

{ (4) У каждого потока

thread_local static hp_owner hazard;←┘ свой указатель опасности

return hazard.get_pointer();← (5)

}

Реализация самой функции get_hazard_pointer_for_current_thread()обманчиво проста (3): в ней объявлена переменная типа hp_ownerв поточно-локальной памяти (4), в которой хранится принадлежащий данному потоку указатель опасности. Затем она просто возвращает полученный от этого объекта указатель (5). Работает это следующим образом: в первый раз, когда каждый поток вызывает эту функцию, создается новый экземпляр hp_owner. Его конструктор (1)ищет в таблице пар (владелец, указатель) незанятую запись (такую, у которой нет владельца). На каждой итерации цикла он с помощью compare_exchange_strong()атомарно выполняет два действия: проверяет, что у текущей записи нет владельца, и делает владельцем себя (2). Если compare_exchange_strong()возвращает false, значит, записью владеет другой поток, поэтому мы идем дальше. Если же функция вернула true, то мы успешно зарезервировали запись для текущего потока, поэтому можем сохранить ее адрес и прекратить поиск (3). Если мы дошли до конца списка и не обнаружили свободной записи (4), значит, потоков, использующих указатель опасности, слишком много, так что приходится возбуждать исключение.

После того как экземпляр hp_ownerдля данного потока создан, последующие обращения происходят гораздо быстрее, потому что указатель запомнен и просматривать таблицу снова нет нужды.

Когда завершается поток, для которого был создан объект hp_owner, этот объект уничтожается. Прежде чем сохранить в идентификаторе владельца значение std::thread::id(), деструктор записывает в сам указатель значение nullptr, чтобы другие потоки могли повторно использовать эту запись. При такой реализации get_hazard_pointer_for_current_thread()реализация функции outstanding_hazard_pointers_for()совсем проста: требуется только найти переданное значение в таблице указателей опасности:

bool outstanding_hazard_pointers_for(void* p) {

for (unsigned i = 0; i < max_hazard_pointers; ++i) {

if (hazard_pointers[i].pointer.load() == p) {

return true;

}

}

return false;

}

He нужно даже проверять, есть ли у записи владелец, так как в бесхозных записях все равно хранятся нулевые указатели, поэтому сравнение заведомо вернёт false; это еще упрощает код. Теперь функции reclaim_later()и delete_nodes_with_no_hazards()могут работать с простым связанным списком; reclaim_later()добавляет в него узлы, a delete_nodes_with_no_hazards()удаляет узлы, на которые не ссылаются указатели опасности. Реализация обеих функций приведена в следующем листинге.