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

Функция 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;

}

while (

!head.compare_exchange_strong(old_counter, new_counter));← (1)

old_counter.external_count = new_counter.external_count;

}

public:

std::shared_ptr pop() {

counted_node_ptr old_head = head.load();

for (;;) {

increase_head_count(old_head);

node* const ptr = old_head.ptr;← (2)

if (!ptr) {

return std::shared_ptr();

}

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

std::shared_ptr res;

res.swap(ptr->data); ← (4)

int const count_increase = old_head.external_count - 2;← (5)

if (ptr->internal_count.fetch_add(count_increase) ==

-count_increase) { ← (6)

delete ptr;

}

return res; ← (7)

} else if (ptr->internal_count.fetch_sub(1) == 1) {

delete ptr; ← (8)

}

}

}

};

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

Увеличив счетчик, мы можем без опаски разыменовать поле ptrобъекта, загруженного из head, и получить тем самым доступ к адресуемому узлу (2). Если оказалось, что указатель нулевой, то мы находимся в конце списка — больше записей нет. В противном случае мы можем попытаться исключить узел из списка, выполнив compare_exchange_strong()с головным узлом head (3).

Если compare_exchange_strong()возвращает true, то мы приняли на себя владение узлом и можем с помощью функции swap()вытащить из него данные, которые впоследствии вернём (4). Тем самым гарантируется, что данные случайно не изменятся, если вдруг другие обращающиеся к стеку другие потоки удерживают указатели на этот узел. Затем можно прибавить внешний счетчик к внутреннему с помощью атомарной операции fetch_add (6). Если теперь счетчик ссылок стал равен нулю, то предыдущее значение (то, которое возвращает fetch_add) было противоположно только что прибавленному, и тогда узел можно удалять. Важно отметить, что прибавленное значение на самом деле на 2 меньше внешнего счетчика (5); мы исключили узел из списка, вследствие чего значение счетчика уменьшилось на 1, и больше не обращаемся к узлу из данного потока, что дает уменьшение еще на 1. Неважно, удаляется узел или нет, наша работа закончена, и мы можем вернуть данные (7).

Если сравнение с обменом (3) не проходит , значит, другой поток сумел удалить узел раньше нас, либо другой поток добавил в стек новый узел. В любом случае нужно начать с начала — с новым значением head, которое вернула функция compare_exchange_strong(). Но прежде необходимо уменьшить счетчик ссылок на узел, который мы пытались исключить раньше. Этот поток больше не будет к нему обращаться. Если наш поток — последний, удерживавший ссылку на этот узел (потому что другой поток вытолкнул его из стека), то внутренний счетчик ссылок равен 1, так что после вычитания 1 он обратится в нуль. В таком случае мы можем удалить узел прямо здесь, не дожидаясь перехода к следующей итерации цикла (8).

До сих мы задавали для всех атомарных операций упорядочение доступа к памяти std::memory_order_seq_cst. В большинстве систем это самый неэффективный режим с точки зрения времени выполнения и накладных расходов на синхронизацию, причем в ряде случаев разница весьма ощутима. Но теперь, определившись с логикой структуры данных, можно подумать и о том, чтобы ослабить некоторые требования к упорядочению, — все-таки не хотелось бы, чтобы пользователи стека несли лишние расходы. Итак, перед тем как расстаться со стеком и перейти к проектированию свободной от блокировок очереди, еще раз присмотримся к операциям стека и спросим себя, нельзя ли где-нибудь использовать более слабое упорядочение доступа, сохранив тот же уровень безопасности?

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

В этом сценарии есть три существенных участника. Во-первых, структура counted_node_ptr, используемая для передачи данных — узла head. Во-вторых, структура node, на которую headссылается. И, в-третьих, сами данные, на которые указывает узел.

Поток, выполняющий push(), сначала конструирует элемент данных и объект node, затем устанавливает head. Поток, выполняющий pop(), сначала загружает значение head, затем в цикле сравнения с обменом увеличивает хранящийся в нем счетчик ссылок, после чего читает структуру node, чтобы извлечь из нее значение next. Из этой последовательности можно вывести требуемое отношение; значение next— простой неатомарный объект, поэтому для его безопасного чтения должно существовать отношение происходит-раньше между операциями сохранения (в заталкивающем потоке) и загрузки (в выталкивающем потоке). Поскольку в push()имеется единственная атомарная операция — compare_exchange_weak(), а для существования отношения происходит-раньше между потоками нам нужна операция освобождения (release), то для функции compare_exchange_weak()необходимо задать упорядочение std::memory_order_releaseили более сильное. Если compare_exchange_weak()вернула false, то ничего не было изменено, и мы можем продолжить цикл, следовательно в этом случае нужна только семантика std::memory_order_relaxed: