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

Листинг 7.20.Модификация pop()с целью помочь при выполнении push()

template

class lock_free_queue {

private:

struct node {

std::atomic data;

std::atomic count;

std::atomic next;← (1)

};

public:

std::unique_ptr pop() {

counted_node_ptr old_head =

head.load(std::memory_order_relaxed);

for (;;) {

increase_external_count(head, old_head);

node* const ptr = old_head.ptr;

if (ptr == tail.load().ptr) {

return std::unique_ptr();

}

counted_node_ptr next = ptr->next.load();← (2)

if (head.compare_exchange_strong(old_head, next)) {

T* const res = ptr->data.exchange(nullptr);

free_external_counter(old_head);

return std::unique_ptr(res);

}

ptr->release_ref();

}

}

};

Как я и говорил, изменения тривиальны: указатель nextтеперь атомарный (1), поэтому операция loadв точке (2)атомарна. В данном примере мы используем упорядочение по умолчанию memory_order_seq_cst, поэтому явное обращение к load()можно было бы опустить, полагаясь на операцию загрузки в операторе неявного преобразования к типу counted_node_ptr, но явное обращение будет напоминать нам, куда впоследствии добавить явное задание порядка обращения к памяти.

Листинг 7.21.Пример реализации функции push(), освобождаемой от блокировок благодаря помощи извне

template

class lock_free_queue {

private:

void set_new_tail(counted_node_ptr &old_tail, ← (1)

counted_node_ptr const &new_tail) {

node* const current_tail_ptr = old_tail.ptr;

while (!tail.compare_exchange_weak(old_tail, new_tail) &&← (2)

old_tail.ptr == current_tail_ptr);

if (old_tail.ptr == current_tail_ptr)← (3)

free_external_counter(old_tail); ← (4)

else

current_tail_ptr->release_ref(); ← (5)

}

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

T* old_data = nullptr;

if (old_tail.ptr->data.compare_exchange_strong( ← (6)

old_data, new_data.get())) {

counted_node_ptr old_next = {0};

if (!old_tail.ptr->next.compare_exchange_strong(← (7)

old_next, new_next)) {

delete new_next.ptr;← (8)

new_next = old_next;← (9)

}

set_new_tail(old_tail, new_next);

new_data.release();

break;

} else { ← (10)

counted_node_ptr old_next = {0};

if (old_tail.ptr->next.compare_exchange_strong(← (11)

old_next, new_next)) {

old_next = new_next; ← (12)

new_next.ptr = new node;← (13)

}

set_new_tail(old_tail, old_next);← (14)

}

}

}

};

В целом похоже на исходную версию push()из листинга 7.15, но есть и несколько принципиальных отличий. Если указатель data действительно установлен (6), то нужно обработать случай, когда нам помог другой поток, и кроме того появилась ветвь else, в которой один поток оказывает помощь другому (10).

Установив указатель dataв узле (6), новая версия push()изменяет указатель next, вызывая compare_exchange_strong() (7). Мы используем compare_exchange_strong(), чтобы избежать цикла. Если обмен завершился неудачно, значит, другой поток уже установил указатель next, поэтому нам ни к чему узел, выделенный в начале, и его можно удалить (8). Мы также хотим использовать значение next, установленное другим потоком, для обновления tail (9).

Собственно обновление указателя tailвынесено в отдельную функцию set_new_tail() (1). В ней мы используем цикл по compare_exchange_weak() (2), потому что если другие потоки пытаются поместить в очередь новый узел с помощью push(), то значение external_countможет измениться, а нам не хотелось бы его потерять. Однако нужно позаботиться и о том, чтобы не затереть значение, которое другой поток уже успешно изменил, в противном случае в очереди могут возникнуть циклы, а это уже совершенно лишнее. Следовательно, нужно гарантировать, что часть ptrзагруженного значения осталась той же самой, если сравнение с обменом не прошло. Если ptrне изменился после выхода из цикла (3), то мы успешно установили tail, поэтому старый внешний счетчик нужно освободить (4). Если значение ptrизменилось, то счетчик уже освобождён другим потоком, поэтому нам нужно только освободить ссылку, которую удерживает наш поток (5).

Если поток, вызвавший push(), не сумел установить указатель dataна этой итерации цикла, то он может помочь более удачливому потоку завершить обновление. Сначала мы пытаемся записать в next указатель на новый узел, выделенный в этом потоке (11). Если это получилось, то выделенный нами узел будет использоваться как новый узел tail (12), а нам следует выделить еще один узел, поскольку поместить свои данные в очередь еще только предстоит (13). После этого мы можем попытаться установить узел tail, вызвав set_new_tailдо перехода к очередной итерации (14).

Вероятно, вы обратили внимание на чрезмерно большое для такого крохотного фрагмента количество newи delete. Вызвало это тем, что новые узлы создаются в push(), а уничтожаются в pop(). Поэтому быстродействие этого кода существенно зависит от того, насколько эффективно работает распределитель памяти; плохой распределитель может полностью свести на нет свойства масштабируемости, присущие свободному от блокировок контейнеру. Вопрос о выборе и реализации подобных распределителей выходит за рамки данной книги, но имейте в виду, что единственный способ узнать, какой распределитель лучше, — испытывать и замерять производительность. К числу стандартных приемов оптимизации выделения памяти можно отнести создание отдельного распределителя в каждом потоке и использование списка свободных узлов — освободившиеся узлы помещаются в этот список, а не возвращаются распределителю.

Ну и, пожалуй, хватит примеров. Давайте теперь на основе вышеизложенного сформулируем ряд рекомендаций по написанию структур данных, свободных от блокировок.

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