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

А теперь перейдём к примерам.

Для демонстрации некоторых приёмов проектирования структур данных, свободных от блокировок, мы рассмотрим реализации ряда простых структур.

Как уже отмечалось, структуры данных, свободные от блокировок, опираются на использование атомарных операций и связанные с ними гарантии упорядочения доступа к памяти, благодаря которым можно быть уверенным, что изменения данных становятся видны потокам в правильном порядке. Сначала мы будем использовать во всех атомарных операциях принимаемое по умолчанию упорядочение memory_order_seq_cst, потому что оно проще всего для понимания (напомним, что семантика memory_order_seq_cstустанавливает полное упорядочение всех операций). Но позже мы посмотрим, как ослабить некоторые ограничения с помощью семантик memory_order_acquire, memory_order_releaseи даже memory_order_relaxed. Хотя ни в одном примере мьютексы не используются напрямую, не стоит забывать, что отсутствие блокировок гарантируется только для типа std::atomic_flag. На некоторых платформах в казалось бы свободном от блокировок коде могут использоваться внутренние блокировки, скрытые в реализации стандартной библиотеки С++ (детали см. в главе 5). В этом случае простая структура данных с блокировками может оказаться предпочтительнее, но дело не только в этом; прежде чем выбирать ту или иную реализацию, нужно четко сформулировать требования, а затем подвергнуть профилированию различные решения, удовлетворяющие этим требованиям.

Итак, снова начнем с простейшей структуры данных — стека.

Основное свойство стека понятно: элементы извлекаются в порядке, обратном тому, в котором помещались — последним пришёл, первым ушел (LIFO). Поэтому важно убедиться, что после добавления значения в стек оно может быть сразу же безопасно извлечено другим потоком и что только один поток получает данное значение. Простейшая реализация стека основана на связанном списке; указатель headнаправлен на первый узел (который будет извлечен следующим), и каждый узел указывает на следующий в списке. При такой схеме добавление узла реализуется просто.

1. Создать новый узел.

2. Записать в его указатель nextтекущее значение head.

3. Записать в headуказатель на новый узел.

Все это прекрасно работает в однопоточной программе, но, когда стек могут модифицировать сразу несколько потоков, этого недостаточно. Существенно, что если узлы добавляют два потока, то между шагами 2 и 3 возможна гонка: второй поток может модифицировать значение headпосле того, как первый прочитает его на шаге 2, но до изменения на шаге 3. В таком случае изменения, произведенные вторым потоком, будут отброшены или случится еще что-нибудь похуже. Прежде чем решать эту проблему, следует отметить, что после того, как указатель headбудет изменен и станет указывать на новый узел, этот узел может быть прочитан другим потоком. Поэтому крайне важно, чтобы новый узел был аккуратно подготовлен до того , как на него начнет указывать head; потом изменять узел уже нельзя.

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

Листинг 7.2.Реализация функции push()без блокировок

template

class lock_free_stack {

private:

struct node {

T data;

node* next;

node(T const& data_) : ← (1)

data(data_) {}

};

std::atomic head;

public:

void push(T const& data) {

node* const new_node = new node(data);← (2)

new_node->next = head.load(); ← (3)

while (!head.compare_exchange_weak(

new_node->next, new_node)); ← (4)

}

};

В этом коде дотошно реализованы все три пункта изложенного выше плана: создать новый узел (2), записать в его поле nextтекущее значение head (3)и записать в headуказатель на новый узел (4). Заполнив данные самой структуры nodeв конструкторе (1), мы гарантируем, что узел готов к использованию сразу после конструирования, так что легкая проблема решена. Затем мы вызываем функцию compare_exchange_weak(), которая проверяет, что указатель headпо-прежнему содержит то значение, которое было сохранено в new_node->next (3), и, если это так, то записывает его в new_node. В этой части программы используется также полезное свойство сравнения с обменом: если функция возвращает false, означающее, что сравнение не прошло (например, потому что значение headбыло изменено другим потоком), то в переменную, которая передана в первом параметре ( new_node->next) записывается текущее значение head. Поэтому нам не нужно перезагружать headна каждой итерации цикла — это сделает за нас компилятор. Кроме того, поскольку мы сразу переходим в начало цикла в случае неудачного сравнения, можно использовать функцию compare_exchange_weak, которая в некоторых архитектурах дает более оптимальный код, чем compare_exchange_strong(см. главу 5).