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

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 {

new_counter = old_counter;

--new_counter.internal_count; ← (1)

}

while (!count.compare_exchange_strong(← (2)

old_counter, new_counter,

std::memory_order_acquire, std::memory_order_relaxed));

if (

!new_counter.internal_count &&

!new_counter.external_counters) {

delete this; ← (3)

}

}

};

};

Реализация node::release_ref()лишь немногим отличается от аналогичного кода в lock_free_stack::pop()(см. листинг 7.11). Там мы работали с единственным внешним счетчиком, поэтому достаточно было вызвать fetch_sub. Здесь же необходимо атомарно обновить всю структуру count, хотя в действительности мы хотим модифицировать только поле internal_count (1). Поэтому никуда не деться от цикла сравнения с обменом (2). Если после уменьшения internal_countоказалось, что и внутренний, и внешний счетчик равны нулю, то это была последняя ссылка, и мы можем удалять узел (3).

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

template

class lock_free_queue {

private:

static void increase_external_count(

std::atomic& counter,

counted_node_ptr& old_counter) {

counted_node_ptr new_counter;

do {

new_counter = old_counter;

++new_counter.external_count;

}

while (!counter.compare_exchange_strong(

old_counter, new_counter,

std::memory_order_acquire, std::memory_order_relaxed));

old_counter.external_count = new_counter.external_count;

}

};

Листинг 7.18 завершает картину. На этот раз мы не освобождаем ссылку, а получаем новую и увеличиваем внешний счетчик. Функция increase_external_count()аналогична increase_head_count()из листинга 7.12, отличаясь от нее тем, что преобразована в статическую функцию-член, которая принимает подлежащий обновлению внешний счетчик извне, а не оперирует внутренним членом класса.

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

template

class lock_free_queue {

private:

static void free_external_counter(

counted_node_ptr &old_node_ptr) {

node* const ptr = old_node_ptr.ptr;

int const count_increase = old_node_ptr.external_count — 2;

node_counter old_counter =

ptr->count.load(std::memory_order_relaxed);

node_counter new_counter;

do {

new_counter = old_counter;

--new_counter.external_counters; ← (1)

new_counter.internal_count += count_increase;← (2)

}

while (!ptr->count.compare_exchange_strong( ← (3)

old_counter, new_counter,

std::memory_order_acquire, std::memory_order_relaxed));

if (!new_counter.internal_count &&

!new_counter.external_counters) {

delete ptr;← (4)

}

}

};

Функция free_external_counter()дополняет increase_external_count(). Она аналогична эквивалентной функции из реализации lock_free_stack::pop()в листинге 7.11, но модифицировала с учетом появления поля external_counters. Она обновляет оба счетчика в одном вызове compare_exchange_strong()для всей структуры count (3)— точно так же мы поступали при уменьшении internal_countв release_ref(). Значение internal_countобновляется, как в листинге 7.11 (2), a external_countersуменьшается на единицу (1). Если теперь оба значения равны нулю, значит, ссылок на узел не осталось, поэтому его можно удалять (4). Оба обновления необходимо выполнять в одном действии (потому и нужен цикл сравнения с обменом), чтобы избежать гонки. Если бы счетчики обновлялись порознь, то два разных потока могли бы решить, что владеют последней ссылкой на узел, и удалить его, что привело бы к неопределенному поведению.

Хотя теперь функция работает и свободна от блокировок, осталась еще одна проблема, касающаяся производительности. После того как один поток начал операцию push(), успешно выполнив compare_exchange_strong()от имени old_tail.ptr->data(точка (5)в листинге 7.15), никакой другой войти в push()не может. Попытавшись это сделать, поток увидит новое значение, отличное от nullptr, в результате чего вызов compare_exchange_strong()вернет false, и потоку придется начать цикл заново. Это активное ожидание, которое только потребляет время процессора, не продвигаясь вперед ни на йоту. По сути дела, это блокировка. Первый удачный вызов push()блокирует все остальные потоки, пока не завершится, так что этот код более не свободен от блокировок . Хуже того — обычно операционная система может отдать приоритет потоку, удерживающему мьютекс, если существуют заблокированные потоки, но только не в данном случае, поэтому остальные потоки так и будут пожирать процессорное время, пока первый не закончит. И тут мы вытащим на свет очередной припасенный для освобождения от блокировок трюк: ожидающий поток может помочь потоку, который выполняет push().

Чтобы вновь сделать код свободным от блокировок, нам нужно придумать, как ожидающий поток может продвигаться вперед, даже если поток, находящийся в push(), застрял. Один из способов — помочь застрявшему потоку, выполнив за него часть работы.

В данном случае мы точно знаем, что нужно сделать: указатель nextв хвостовом узле требуется установить на новый фиктивный узел, и тогда сам указатель tailможно будет обновить. Все фиктивные узлы эквивалентны, поэтому не имеет значения, какой из них использовать — созданный потоком, который успешно поместил в очередь данные, или потоком, ожидающим входа в push(). Если сделать указатель nextв узле атомарным, то для его установки можно будет применить compare_exchange_strong(). После того как указатель nextустановлен, в цикле по compare_exchange_weak()можно будет установить tail, проверяя при этом, указывает ли он по-прежнему на тот же самый исходный узел. Если это не так, значит, узел обновил какой-то другой поток, так что можно прекратить попытки и перейти в начало цикла. Реализация этой идеи потребует также небольшого изменения pop(), где нужно будет загрузить указатель next; эта модификация показана в листинге ниже.