Параллельное программирование на С++ в действии

Алгоритмы и структуры данных, в которых для синхронизации доступа используются мьютексы, условные переменные и будущие результаты, называются блокирующими. Приложение вызывает библиотечные функции, которые приостанавливают выполнение потока до тех пор, пока другой поток не завершит некоторое действие. Такие библиотечные функции называются блокирующими, потому что поток не может продвинуться дальше некоторой точки, пока не будет снят блокировка. Обычно ОС полностью приостанавливает заблокированный поток (и отдает его временные кванты другому потоку) до тех пор, пока он не будет разблокирован в результате выполнения некоторого действия в другом потоке, будь то освобождение мьютекса, сигнал условной переменной или перевод будущего результата в состояние готов.

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

7.1.1. Типы неблокирующих структур данных

В главе 5 мы реализовали простой мьютекс-спинлок с помощью

std::atomic_flag

Листинг 7.1. Реализация мьютекса-спинлока с помощью

std::atomic_flag

class spinlock_mutex {

 std::atomic_flag flag;

public:

 spinlock_mutex():

  flag(ATOMIC_FLAG_INIT) {}

 void lock() {

  while (flag.test_and_set(std::memory_order_acquire));

 }

 void unlock() {

  flag.clear(std::memory_order_release);

 }

};

Здесь не вызываются никакие блокирующие функции;

lock()

test_and_set()

false

7.1.2. Структуры данных, свободные от блокировок

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

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

Свободные от блокировок алгоритмы с такими циклами могут приводить к застреванию (starvation) потоков, когда один поток, выполняющий операции с «неудачным» хронометражем, продвигается вперёд, а другой вынужден постоянно повторять одну и ту же операцию. Структуры данных, в которых такой проблемы не возникает, называются свободными от блокировок и ожидания.

7.1.3. Структуры данных, свободные от ожидания

Свободная от блокировок структура данных называется свободной от ожидания, если обладает дополнительным свойством: каждый обращающийся к ней поток может завершить свою работу за ограниченное количество шагов вне зависимости от поведения остальных потоков. Алгоритмы, в которых количество шагов может быть неограничено из-за конфликтов с другими потоками, не свободны от ожидания.

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

7.1.4. Плюсы и минусы структур данных, свободных от блокировок

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

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

Но у этой медали есть и оборотная сторона: если вы не можете запретить потокам одновременный доступ к структуре, то должны внимательно следить за соблюдением инвариантов или выбирать альтернативные инварианты, соблюдение которых можно гарантировать. Кроме того, следует обращать внимание на ограничения упорядочения, налагаемые на операции. Чтобы избежать неопределённого поведения вследствие гонки за данными, следует использовать для всех модификаций атомарные операции. Но и этого недостаточно — необходимо гарантировать, что изменения становятся видны другим потокам в правильном порядке. Все это означает, что написание потокобезопасных структур данных без использования блокировок гораздо сложнее, чем с блокировками.

Ввиду отсутствия блокировок невозможны и взаимоблокировки, однако вместо них появляется угроза активных блокировок. Активная блокировка (live lock) возникает, когда два потока одновременно пытаются изменить структуру данных, но каждый из них должен начинать свою операцию сначала из-за изменений, произведенных другим потоком. Таким образом, каждый поток беспрестанно повторяет попытки в цикле. Представьте себе двух людей, пытающихся разойтись в узком проходе. Войдя в него одновременно, они сталкиваются лбами, поэтому оба отступают назад и пробуют еще раз. И так будет повторяться до тех пор, пока кто-то не проскочит первым (по взаимному согласию, потому что оказался быстрее или просто благодаря удаче). Как и в этом простом примере, активные блокировки обычно существуют недолго, потому что зависят от случайных временных соотношений при планировании потоков. Поэтому они скорее «подъедают» производительность, чем вызывают долгосрочные проблемы, но остерегаться их все равно стоит. По определению программа, свободная от блокировок, не страдает от активных блокировок, потому что существует ограничение сверху на количество шагов, необходимых для выполнения операции. Зато и алгоритм, скорее всего, окажется сложнее альтернативного и может потребовать большего числа шагов даже в случае, когда никакой другой поток одновременно не обращается к структуре данных.

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

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

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

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

memory_order_seq_cst

memory_order_seq_cst

memory_order_acquire

memory_order_release

memory_order_relaxed

std::atomic_flag

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

7.2.1. Потокобезопасный стек без блокировок

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

head

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

2. Записать в его указатель

next

head

3. Записать в

head

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

head

head

head

Ну хорошо, а как все-таки быть с этим неприятным состоянием гонки? Ответ таков — использовать атомарную операцию сравнить-и-обменять на шаге 3, гарантирующую, что

head

push()

Листинг 7.2. Реализация функции

push()

template

class lock_free_stack {

private:

 struct node {

  T data;

  node* next;

  node(T const& data_) : ←

  data(data_) {}

 };

 std::atomic head;

public:

 void push(T const& data) {

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

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

  while (!head.compare_exchange_weak(

      new_node->next, new_node));  ←

 }

};

В этом коде дотошно реализованы все три пункта изложенного выше плана: создать новый узел (2), записать в его поле

next

head

head

node

compare_exchange_weak()

head

new_node->next

new_node

false

head

new_node->next

head

head

compare_exchange_weak

compare_exchange_strong

Итак, операции

pop()

push()

node

compare_exchange_weak()

head

push()

Теперь, когда у нас есть средства для добавления данных в стек, надо научиться их извлекать обратно. На первый взгляд, тут всё просто.

1. Прочитать текущее значение

head

2. Прочитать

head->next

3. Записать в

head

head->next

4. Вернуть поле

data

node

5. Удалить извлеченный узел.

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

head

Однако на этом трудности не кончаются. Есть еще одна проблема: если два потока прочитают одно и то же значение

head

push()

head

head

Если сравнение с обменом завершилось успешно, то мы точно знаем, что больше ни один поток не пытался удалить данный узел из стека, поэтому можем без опаски выполнить шаг 4. Вот первая попытка написать код

pop()

template

class lock_free_stack {

public:

 void pop(T& result) {

  node* old_head = head.load();

  while (!head.compare_exchange_weak(old_head, old_head->next));

  result = old_head->data;

 }

};

Вроде бы всё красиво и лаконично, но, помимо утечки узлов, осталось еще две проблемы. Во-первых, этот код не работает для пустого списка: если указатель

head

next

while

head

nullptr

Вторая проблема касается безопасности относительно исключений. Впервые подступаясь к потокобезопасному стеку в главе 3, мы видели, что простой возврат объекта по значению небезопасен относительно исключений: если исключение возникает во время копирования возвращаемого значения, то значение будет потеряно. Тогда передача ссылки на результат оказалась приемлемым решением, которое гарантировало неизменность стека в случае исключения. К сожалению, сейчас мы лишены такой роскоши; безопасно скопировать данные можно только тогда, когда мы точно знаем, что больше никакой поток не пытается вернуть данный узел, а это означает, что узел уже удален из стека. Следовательно, передача возвращаемого значения по ссылке больше не является преимуществом, с тем же успехом можно было бы вернуть его и по значению. Чтобы безопасно вернуть значение, придется воспользоваться другим вариантом, описанным в главе 3: возвращать интеллектуальный указатель на данные.

Возврат

nullptr

pop()

push()

node

std::shared_ptr<>

pop()

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

template

class lock_free_stack {

private:

 struct node       

 {             │

  std::shared_ptr data;←┘

  node* next;

  node(T const& data_) :      

  data(std::make_shared(data))←┤

  {}                │

 };

 std::atomic head;

public:

 void push(T const& data) {

  node* const new_node = new node(data);

  new_node->next = head.load();

  while (!head.compare_exchange_weak(new_node->next, new_node));

 }

 std::shared_ptr pop()

 {

  node* old_head = head.load();│

  while (old_head &&      ←┘

  !head.compare_exchange_weak(old_head, old_head->next));

  return old_head ? old_head->data : std::shared_ptr();←

 }

};

Теперь данные удерживаются указателем (1), поэтому мы должны выделять память для них из кучи в конструкторе узле (2). Кроме того, перед тем как разыменовывать

old_head

compare_exchange_weak()

while

push()

pop()

compare_exchange_weak()

false

Если бы у нас был сборщик мусора (как в управляемых языках типа С# или Java), то на этом можно было бы ставить точку — старый узел был бы убран и повторно использован после того, как все потоки перестанут к нему обращаться. Но сегодня мало найдётся компиляторов С++ с встроенным сборщиком мусора, поэтому прибираться за собой нужно самостоятельно.

7.2.2. Устранение утечек: управление памятью в структурах данных без блокировок

При первом подходе к

pop()

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

pop()

push()

pop()

С другой стороны, если мы все-таки хотим, чтобы несколько потоков могли одновременно вызывать

pop()

node

node

pop()

push()

pop()

Если потоков, вызывающих

pop()

pop()

pop()

pop()

Листинг 7.4. Освобождение занятой узлами памяти в момент, когда нет потоков, вызывающих

pop()

template

class lock_free_stack {

private:

 std::atomic threads_in_pop;←┐

 void try_reclaim(node* old_head);  

public:

 std::shared_ptr pop()

 {               

  ++threads_in_pop;      ←┤

  node* old_head = head.load();│

  while (old_head &&

  !head.compare_exchange_weak(old_head, old_head->next));

  std::shared_ptr res;

  if (old_head)

  {             

  res.swap(old_head->data);←┤

  }              │

  try_reclaim(old_head);←┐

  return res;      

 }

};

Атомарная переменная

threads_in_pop

pop()

try_reclaim()

swap()

try_reclaim()

Листинг 7.5. Механизм освобождения памяти на основе подсчёта ссылок

template

class lock_free_stack {

private:

 std::atomic to_be_deleted;

 static void delete_nodes(node* nodes) {

  while (nodes) {

  node* next = nodes->next;

  delete nodes;

  nodes = next;

  }

 }

 void try_reclaim(node* old_head) {

  if (threads_in_pop == 1) ←

  {     

  node* nodes_to_delete = to_be_deleted.exchange(nullptr);←┘

  if (!--threads_in_pop)←┐

  {            

   delete_nodes(nodes_to_delete);    ←

  } else if(nodes_to_delete) {      ←

   chain_pending_nodes(nodes_to_delete);←

  }

  delete old_head; ←

  } else {

  chain_pending_node(old_head); ←

  --threads_in_pop;

  }

 }

 void chain_pending_nodes(node* nodes) {

  node* last = nodes;

  while (node* const next =

      last->next) {←┐  

            ├

  last = next;     │  

  }

  chain_pending_nodes(nodes, last);

 }

 void chain_pending_nodes(node* first, node* last) {

  last->next = to_be_deleted;←

  while (

  !to_be_deleted.compare_exchange_weak(←┐  

  last->next, first));         ├

  }                   │  

 void chain_pending_node(node* n) {

  chain_pending_nodes(n, n);←

 }

};

Если при попытке освободить занятую узлом (1) память счетчик

threads_in_pop

pop()

Предположим, что

threads_in_pop

exchange

threads_in_pop

delete_nodes

Если счетчик после уменьшения не равен нулю, то освобождать узлы небезопасно, поэтому если такие узлы есть (5), то нужно поместить их в начало списка ожидающих (6). Такое может случиться, если к структуре данных одновременно обращаются несколько потоков. Другие потоки могли вызвать

pop()

threads_in_pop

to_be_deleted

old_head

next

Рис. 7.1. На примере трех потоков, вызывающих

pop()

threads_in_pop

try_reclaim()

Чтобы поместить узлы, ожидающие удаления, в список ожидающих, мы используем уже имеющийся указатель

next

next

to_be_deleted

to_be_deleted

compare_exchange_weak

next

Этот алгоритм работает вполне приемлемо, если нагрузка невелика, то есть существуют моменты затишья, когда в

pop()

threads_in_pop

threads_in_pop

pop()

threads_in_pop

Если нагрузка высока, то затишье может не наступить никогда, поскольку новые потоки входят в

pop()

to_be_deleted

7.2.3. Обнаружение узлов, не подлежащих освобождению, с помощью указателей опасности

Термин указатели опасности откосится к технике, предложенной Магедом Майклом (Maged Michael)[12]. Они называются так потому, что удаление узла, на который все еще могут ссылаться другие потоки, — опасное предприятие. Если действительно существует поток, ссылающийся на данный узел, и этот поток попытается обратиться к нему по ссылке, то мы получим неопределенное поведение. Основная идея заключается в том, что поток, собирающийся получить доступ к объекту, который другой поток может захотеть удалить, сначала устанавливает указатель опасности, ссылающийся на этот объект, информируя тем самым другой поток, что удалять этот объект действительно опасно. После того как объект перестает быть нужным, указатель опасности очищается. Если вам доводилось наблюдать гребную регату между командами Оксфорда и Кембриджа, то вы могли видеть аналогичный механизм, применяемый в начале заезда: рулевой в лодке может поднять руку, сообщая, что экипаж еще не готов. Пока хотя бы один рулевой держит руку поднятой, судья не может дать старт заезду. После того как оба рулевых опустят руки, судья может давать старт, однако рулевой вправе снова поднять руку, если заезд еще не начался, а ситуация, на его взгляд, изменилась.

Собираясь удалить объект, поток должен сначала проверить указатели опасности в других имеющихся в системе потоках. Если ни один из указателей опасности не ссылается на данный объект, то его можно спокойно удалять. В противном случае удаление следует отложить. Периодически поток просматривает список отложенных объектов в поисках тех, которые уже можно удалить.

Высокоуровневое описание выглядит достаточно простым, но как это сделать на С++?

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

get_hazard_pointer_for_current_thread()

head

std::shared_ptr pop() {

 std::atomic& hp =

 get_hazard_pointer_for_current_thread();

 node* old_head = head.load();←

 node* temp;

 do {

  temp = old_head;

  hp.store(old_head);     ←

  old_head = head.load();

 } while (old_head != temp); ←

 // ...

}

Это необходимо делать в цикле

while

node

head

head

head

head

new

delete

Установив указатель опасности, мы можем продолжить выполнение

pop()

old_head

pop()

Листинг 7.6. Реализация функции

pop()

std::shared_ptr pop() {

 std::atomic& hp =

  get_hazard_pointer_for_current_thread();

 node* old_head = head.load();

 do {

  node* temp;

  do     ←┤

  {      │

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

 std::shared_ptr res;

 if (old_head) {

  res.swap(old_head->data);

  if (outstanding_hazard_pointers_for(old_head))←┐

  {                        ├

  reclaim_later(old_head);           │

←

  }                        │

  else                      │

  {                        │

                         │

  delete old_head; ←

  }

  delete_nodes_with_no_hazards();←

 }

 return res;

}

Начнём с того, что мы перенесли цикл, в котором устанавливается указатель опасности, во внешний цикл, где перечитывается

old_head

compare_exchange_strong()

while

compare_exchange_weak()

old_head

reclaim_later()

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

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) {←

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

  }

 }

 std::atomic& get_pointer() {

  return hp->pointer;

 }

 ~hp_owner() {←

  hp->pointer.store(nullptr);

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

 }

};

std::atomic& get_hazard_pointer_for_current_thread()←

{                   

 thread_local static hp_owner hazard;←┘

 return hazard.get_pointer();←

}

Реализация самой функции

get_hazard_pointer_for_current_thread()

hp_owner

hp_owner

compare_exchange_strong()

compare_exchange_strong()

false

true

После того как экземпляр

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

delete_nodes_with_no_hazards()

Листинг 7.8. Простая реализация функций освобождения узлов

template

void do_delete(void* p) {

 delete static_cast(p);

}

struct data_to_reclaim {

 void* data;

 std::function deleter;

 data_to_reclaim* next;

 template

 data_to_reclaim(T* p) : ←

  data(p),

  deleter(&do_delete), next(0) {}

 ~data_to_reclaim() {

  deleter(data); ←

 }

};

std::atomic nodes_to_reclaim;

void add_to_reclaim_list(data_to_reclaim* node) {←

 node->next = nodes_to_reclaim.load();

 while (

  !nodes_to_reclaim.compare_exchange_weak(node->next, node));

}

template

void reclaim_later(T* data) {          ←

 add_to_reclaim_list(new data_to_reclaim(data));←

}

void delete_nodes_with_no_hazards() {

 data_to_reclaim* current =

 nodes_to_reclaim.exchange(nullptr);          ←

 while(current) {

  data_to_reclaim* const next = current->next;

  if (!outstanding_hazard_pointers_for(current->data)) {←

  delete current;                    ←

  } else {

  add_to_reclaim_list(current);             ←

  }

  current = next;

 }

}

Полагаю, вы обратили внимание, что

reclaim_later()

std::atomic

void*

delete

data_to_reclaim

reclaim_later()

data_to_reclaim

add_to_reclaim_list()

compare_exchange_weak()

Но вернёмся к конструктору

data_to_reclaim

void*

data

do_delete()

void*

std::function<>

data_to_reclaim

Деструктор

data_to_reclaim

delete_nodes_with_no_hazards()

Эта функция сначала заявляет права на владение всем списком подлежащих освобождению узлов, вызывая

exchange()

Далее мы по очереди просматриваем все узлы списка, проверяя, ссылаются ли на них не сброшенные указатели опасности (7). Если нет, то запись можно удалить (очистив хранящиеся в ней данные) (8). В противном случае мы возвращаем элемент в список, чтобы освободить его позже (9).

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

max_hazard_pointers

pop()

pop()

max_hazard_pointers

max_hazard_pointers

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

pop()

max_hazard_pointers

max_hazard_pointers+1

max_hazard_pointers

pop()

2*max_hazard_pointers

max_hazard_pointers

max_hazard_pointers

pop()

max_hazard_pointers

pop()

2*max_hazard_pointers

max_hazard_pointers

pop()

max_hazard_pointers

pop()

Но и у этого решения есть недостаток (помимо увеличенного расхода памяти): теперь мы должны подсчитывать узлы в списке освобождения, то есть использовать атомарный счетчик, а, кроме того, за доступ к самому списку конкурируют несколько потоков. Если память позволяет, то можно предложить еще более эффективную схему освобождения: каждый поток хранит собственный список освобождения в поточно-локальной памяти. Тогда ни для счетчика, ни для доступа к списку не понадобятся атомарные переменные. Но в обмен придется выделить память для

max_hazard_pointers * max_hazard_pointers

Еще один недостаток указателей опасности состоит в том, что они защищены патент пой заявкой, поданной IBM[13]. Если вы пишете программное обеспечение, которое будет применяться в стране, где эти патенты признаются, то придется получить соответствующую лицензию. Это проблема, общая для многих методов освобождения памяти без блокировок; поскольку в этой области ведутся активные исследования, крупные компании берут патенты всюду, где могут. Возможно, вы задаетесь вопросом, зачем я посвятил так много страниц описанию техники, которой многие не смогут воспользоваться. Что ж, вопрос не праздный. Во-первых, в некоторых случаях ей можно воспользоваться, не платя лицензионных отчислений. Например, если вы разрабатываете бесплатную программу на условиях лицензии GPL[14], то она может подпадать под заявление IBM об отказе от патентных притязаний[15]. Во-вторых — и это более существенно — объяснение техники помогает высветить вещи, о которых надо помнить при написании кода, свободного от блокировок, например, о плате за атомарные операции.

А существуют ли непатентованные методы освобождения памяти, применимые в программах без блокировок? К счастью, да. Один из них — подсчет ссылок.

7.2.4. Нахождение используемых узлов с помощью подсчета ссылок

В разделе 7.2.2 мы видели, что проблема удаления узлов сводится к задаче нахождения узлов, к которым еще обращаются потоки-читатели. Если бы можно было точно узнать, на какие узлы есть ссылки и когда количество ссылок обращается в нуль, то узлы можно было бы удалить. Указатели опасности решают эту проблему путем хранения списка используемых узлов, а механизм подсчета ссылок — путем хранения числа потоков, обращающихся к каждому узлу.

Выглядит просто и элегантно, но реализовать на практике очень трудно. Сразу приходит в голову мысль, что для такой задачи подошло бы что-то вроде

std::shared_ptr<>

std::shared_ptr<>

std::shared_ptr<>

std::atomic_is_lock_free(&some_shared_ptr)

true

std::shared_ptr

Листинг 7.9. Свободный от блокировок стек на основе свободной от блокировок реализации

std::shared_ptr<>

template

class lock_free_stack {

private:

 struct node {

  std::shared_ptr data;

  std::shared_ptr next;

  node(T const& data_) :

  data(std::make_shared(data_)) {}

 };

 std::shared_ptr head;

public:

 void push(T const& data) {

  std::shared_ptr const new_node =

  std::make_shared(data);

  new_node->next = head.load();

  while (!std::atomic_compare_exchange_weak(

  &head, &new_node->next, new_node));

 }

 std::shared_ptr pop() {

  std::shared_ptr old_head = std::atomic_load(&head);

  while(old_head && !std::atomic_compare_exchange_weak(

  &head, &old_head, old_head->next));

  return old_head ? old_head->data : std::shared_ptr();

 }

};

В том весьма вероятном случае, когда реализация

std::shared_ptr<>

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

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

В листинге ниже показала внутренняя структура данных и реализация функции

push()

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

template

class lock_free_stack {

private:

 struct node;

 struct counted_node_ptr {←

  int external_count;

  node* ptr;

 };

 struct node {

  std::shared_ptr data;←

  std::atomic internal_count;

  counted_node_ptr next;  ←

  node(T const& data_) :

  data(std::make_shared(data_)), internal_count(0) {}

 };

 std::atomic head;←

public:

 ~lock_free_stack() {

  while(pop());

 }

 void push(T const& data) {←

  counted_node_ptr new_node;

  new_node.ptr = new node(data);

  new_node.external_count = 1;

  new_node.ptr->next = head.load();

  while(

  !head.compare_exchange_weak(new_node.ptr->next, new_node));

 }

};

Обратите внимание, что внешний счетчик хранится вместе с указателем на узел в структуре

counted_node_ptr

next

node

counted_node_ptr

struct

std::atomic<>

head

На платформах, где поддерживается операция сравнения и обмена двойного слова, размер этой структуры достаточно мал для того, чтобы тип данных

std::atomic

std::shared_ptr<>

std::atomic<>

Функция

push()

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

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

  if (!ptr) {

   return std::shared_ptr();

  }

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

   std::shared_ptr res;

   res.swap(ptr->data);                  ←

   int const count_increase = old_head.external_count - 2;←

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

     -count_increase) {                 ←

   delete ptr;

   }

   return res; ←

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

   delete ptr; ←

  }

  }

 }

};

Теперь, загрузив значение

head

head

compare_exchange_strong()

Увеличив счетчик, мы можем без опаски разыменовать поле

ptr

head

compare_exchange_strong()

head

Если

compare_exchange_strong()

true

swap()

fetch_add

fetch_add

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

head

compare_exchange_strong()

До сих мы задавали для всех атомарных операций упорядочение доступа к памяти

std::memory_order_seq_cst

7.2.5. Применение модели памяти к свободному от блокировок стеку

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

В этом сценарии есть три существенных участника. Во-первых, структура

counted_node_ptr

head

node

head

Поток, выполняющий

push()

node

head

pop()

head

node

next

next

push()

compare_exchange_weak()

compare_exchange_weak()

std::memory_order_release

compare_exchange_weak()

false

std::memory_order_relaxed

void push(T const& data) {

 counted_node_ptr new_node;

 new_node.ptr = new node(data);

 new_node.external_count = 1;

 new_node.ptr->next = head.load(std::memory_order_relaxed);

 while (!head.compare_exchange_weak(

  new_node.ptr->next, new_node,

  std::memory_order_release, std::memory_order_relaxed));

}

А что можно сказать о коде

pop()

next

std::memory_order_acquire

next

compare_exchange_strong()

increase_head_count()

push()

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,

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

 old_counter.external_count = new_counter.external_count;

}

Если вызов

compare_exchange_strong()

ptr

old_counter

push()

compare_exchange_strong()

ptr

push()

ptr->next

pop()

Отметим, что для этого анализа порядок доступа к памяти в начальном вызове

head.load()

std::memory_order_relaxed

Далее на очереди операция

compare_exchange_strong()

head

old_head.ptr->next

ptr->data

ptr->data

push()

increase_head_count()

push()

push()

head

increase_head_count()

ptr->data

pop()

std::memory_order_relaxed

ptr->data

swap()

Если

compare_exchange_strong()

old_head

std::memory_order_acquire

increase_head_count()

std::memory_order_relaxed

Что можно сказать насчёт других потоков? Нужны ли более сильные ограничения, чтобы и другие потоки работали безопасно? Нет, не нужны, потому что

head

push()

compare_exchange_weak()

push()

compare_exchange_strong()

increase_head_count()

head

Мы почти закончили, осталось только рассмотреть функции, в которых используются операции

fetch_add()

swap()

swap()

delete

std::memory_order_release

fetch_add()

std::memory_order_acquire

delete

fetch_add()

load()

std::memory_order_acquire

fetch_add()

std::memory_order_relaxed

pop()

Листинг 7.12. Свободный от блокировок стек с подсчётом ссылок и ослабленными атомарными операциями

template

class lock_free_stack {

private:

 struct node;

 struct counted_node_ptr {

  int external_count;

  node* ptr;

 };

 struct node {

  std::shared_ptr data;

  std::atomic internal_count;

  counted_node_ptr next;

  node(T const& data_):

  data(std::make_shared(data_)), internal_count(0) {}

 };

 std::atomic head;

 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,

     std::memory_order_acquire,

     std::memory_order_relaxed));

 old_counter.external_count = new_counter.external_count;

 }

public:

 ~lock_free_stack() {

  while(pop());

 }

 void push(T const& data) {

  counted_node_ptr new_node;

  new_node.ptr = new node(data);

  new_node.external_count = 1;

  new_node.ptr->next = head.load(std::memory_order_relaxed);

  while (!head.compare_exchange_weak(

      new_node.ptr->next, new_node,

      std::memory_order_release,

      std::memory_order_relaxed));

 }

 std::shared_ptr pop() {

  counted_node_ptr old_head =

  head.load(std::memory_order_relaxed);

  for (;;) {

  increase_head_count(old_head);

  node* const ptr = old_head.ptr;

  if (!ptr) {

   return std::shared_ptr();

  }

  if (head.compare_exchange_strong(old_head, ptr->next,

    std::memory_order_relaxed)) {

   std::shared_ptr res;

   res.swap(ptr->data);

   int const count_increase = old_head.external_count — 2;

   if (ptr->internal_count.fetch_add(count_increase,

     std::memory_order_release) == -count_increase) {

   delete ptr;

   }

   return res;

  }

  else if (ptr->internal_count.fetch_add(-1,

       std::memory_order_relaxed) == 1) {

   ptr->internal_count.load(std::memory_order_acquire);

   delete ptr;

  }

  }

 }

};

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

pop()

pop()

7.2.6. Потокобезопасная очередь без блокировок

Очередь отличается от стека прежде всего тем, что операции

push()

pop()

try_pop()

pop()

Если взять листинг 6.6 за основу, то нам понадобятся два указателя на

node

head

tail

Листинг 7.13. Свободная от блокировок очередь с одним производителем и одним потребителем

template

class lock_free_queue {

private:

 struct node {

  std::shared_ptr data;

  node* next;

  node():

  next(nullptr) {}

 };

 std::atomic head;

 std::atomic tail;

 node* pop_head() {

  node* const old_head = head.load();

 if (old_head == tail.load()) {←

  return nullptr;

  }

 head.store(old_head->next);

  return old_head;

 }

public:

 lock_free_queue():

  head(new node), tail(head.load()) {}

 lock_free_queue(const lock_free_queue& other) = delete;

 lock_free_queue& operator=(

  const lock_free_queue& other) = delete;

 ~lock_free_queue() {

  while(node* const old_head = head.load()) {

  head.store(old_head->next);

  delete old_head;

  }

 }

 std::shared_ptr pop() {

  node* old_head = pop_head();

  if (!old_head) {

  return std::shared_ptr();

  }

  std::shared_ptr const res(old_head->data);←

  delete old_head;

  return res;

 }

 void push(T new_value) {

  std::shared_ptr new_data(std::make_shared(new_value));

  node* p = new node;         ←

  node* const old_tail = tail.load(); ←

  old_tail->data.swap(new_data);    ←

  old_tail->next = p;         ←

  tail.store(p);            ←

 }

};

На первый взгляд, неплохо, и если в каждый момент времени существует только один поток, вызывающий

push()

pop()

push()

pop()

tail

tail

data

tail

tail

data

data

Проблемы начинаются, когда несколько потоков вызывают

push()

pop()

push()

push()

tail

data

next

Аналогичные проблемы возникают в

pop_head()

head

next

next

head

pop()

Итак, проблему

pop()

push()

push()

pop()

tail

push()

tail

push()

Один из способов — добавить фиктивный узел между реальными. Тогда единственной частью текущего узла

tail

next

next

nullptr

tail

pop()

pop()

Второй способ — сделать указатель

data

next

tail

tail

td::shared_ptr<>

pop()

std::unique_ptr<>

std::atomic

compare_exchange_strong()

pop()

push()

Листинг 7.14. Первая (неудачная) попытка переработки

push()

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;

 for (;;) {

  node* const old_tail = tail.load();←

  T* old_data = nullptr;

  if (old_tail->data.compare_exchange_strong(

  old_data, new_data.get())) {    ←

  old_tail->next = new_next;

  tail.store(new_next.ptr);     ←

  new_data.release();

  break;

  }

 }

}

Применение схемы подсчета ссылок устраняет эту конкретную гонку, но в

push()

push()

pop()

tail

head

next

node

push()

Листинг 7.15. Реализация

push()

tail

template

class lock_free_queue {

private:

 struct node;

 struct counted_node_ptr {

  int external_count;

  node* ptr;

 };

 std::atomic head;

 std::atomic tail;←

 struct node_counter {

  unsigned internal_count:30;

  unsigned external_counters:2;←

 };

 struct node {

  std::atomic data;

  std::atomic count;←

 counted_node_ptr next;

  node() {

  node_counter new_count;

  new_count.internal_count = 0;

  new_count.external_counters = 2;←

  count.store(new_count);

  next.ptr = nullptr;

  next.external_count = 0;

  }

 };

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(←

    old_data, new_data.get())) {

   old_tail.ptr->next = new_next;

   old_tail = tail.exchange(new_next);

   free_external_counter(old_tail);       ←

   new_data.release();

   break;

  }

  old_tail.ptr->release_ref();

  }

 }

};

В листинге 7.15

tail

atomic

head

node

count

internal_count

count

internal_count

external_counters

external_counters

internal_count

При инициализации структуры

node

internal_count

external_counters

tail

next

push()

tail

compare_exchange_strong()

data

increase_external_count()

free_external_counter()

old_tail

Разобравшись с

push()

pop()

pop()

Листинг 7.16. Извлечение узла из очереди без блокировок с подсчётом ссылок на

tail

template

class lock_free_queue {

private:

 struct node {

 void release_ref();

};

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

   ptr->release_ref();←

   return std::unique_ptr();

  }

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

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

   free_external_counter(old_head);←

   return std::unique_ptr(res);

  }

  ptr->release_ref();

  }

 }

};

Все начинается с загрузки значения

old_head

head

tail

compare_exchange_strong()

Листинг 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;     ←

  }

  while (!count.compare_exchange_strong(←

     old_counter, new_counter,

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

  if (

   !new_counter.internal_count &&

   !new_counter.external_counters) {

   delete this; ←

  }

  }

 };

};

Реализация

node::release_ref()

lock_free_stack::pop()

fetch_sub

count

internal_count

internal_count

Листинг 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.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;       ←

  new_counter.internal_count += count_increase;←

  }

  while (!ptr->count.compare_exchange_strong(  ←

  old_counter, new_counter,

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

  if (!new_counter.internal_count &&

    !new_counter.external_counters) {

  delete ptr;←

  }

 }

};

Функция

free_external_counter()

increase_external_count()

lock_free_stack::pop()

external_counters

compare_exchange_strong()

count

internal_count

release_ref()

internal_count

external_counters

Хотя теперь функция работает и свободна от блокировок, осталась еще одна проблема, касающаяся производительности. После того как один поток начал операцию

push()

compare_exchange_strong()

old_tail.ptr->data

push()

nullptr

compare_exchange_strong()

false

push()

push()

Чтобы вновь сделать код свободным от блокировок, нам нужно придумать, как ожидающий поток может продвигаться вперед, даже если поток, находящийся в

push()

В данном случае мы точно знаем, что нужно сделать: указатель

next

tail

push()

next

compare_exchange_strong()

next

compare_exchange_weak()

tail

pop()

next

Листинг 7.20. Модификация

pop()

push()

template

class lock_free_queue {

private:

 struct node {

  std::atomic data;

  std::atomic count;

  std::atomic next;←

 };

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

  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

load

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, ←

          counted_node_ptr const &new_tail) {

  node* const current_tail_ptr = old_tail.ptr;

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

     old_tail.ptr == current_tail_ptr);

  if (old_tail.ptr == current_tail_ptr)←

  free_external_counter(old_tail);   ←

  else

  current_tail_ptr->release_ref();  ←

 }

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(  ←

   old_data, new_data.get())) {

   counted_node_ptr old_next = {0};

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

     old_next, new_next)) {

   delete new_next.ptr;←

   new_next = old_next;←

   }

   set_new_tail(old_tail, new_next);

   new_data.release();

   break;

  } else {       ←

   counted_node_ptr old_next = {0};

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

     old_next, new_next)) {

   old_next = new_next;   ←

   new_next.ptr = new node;←

   }

   set_new_tail(old_tail, old_next);←

  }

  }

 }

};

В целом похоже на исходную версию

push()

data

else

Установив указатель

data

push()

next

compare_exchange_strong()

compare_exchange_strong()

next

next

tail

Собственно обновление указателя

tail

set_new_tail()

compare_exchange_weak()

push()

external_count

ptr

ptr

tail

ptr

Если поток, вызвавший

push()

data

tail

tail

set_new_tail

Вероятно, вы обратили внимание на чрезмерно большое для такого крохотного фрагмента количество

new

delete

push()

pop()

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

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

7.3.1. Используйте

std::memory_order_seq_cst

для создания прототипа

Порядок доступа к памяти

std::memory_order_seq_cst

std::memory_order_seq_cst

7.3.2. Используйте подходящую схему освобождения памяти

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

• Дождаться, когда к структуре данных не будет обращаться ни один поток, и удалить разом все объекты, ожидающие удаления.

• Использовать указатели опасности для выявления потока, обращающегося к конкретному объекту.

• Подсчитывать ссылки на объекты и не удалять их, пока ссылки существуют.

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

Другой способ — использовать узлы повторно и полностью освобождать их только тогда, когда уничтожается вся структура данных. Раз узлы используются повторно, то память никогда не становится недействительной, поэтому некоторые проблемы, сопряженные с неопределенным поведением, вообще не возникают. Недостаток же в том, что взамен появляется так называемая проблема ABA.

7.3.3. Помните о проблеме ABA

Проблема ABA свойственна любому алгоритму, основанному на сравнении с обменом. Проявляется она следующим образом.

1. Поток 1 читает атомарную переменную

x

А

2. Поток 1 выполняет некоторую операцию, исходя из этого значения, например разыменовывает его (если это указатель), выполняет поиск или делает еще что-то.

3. Операционная система приостанавливает поток 1.

4. Другой поток выполняет некоторые операции с

x

B

5. Затем этот поток изменяет данные, ассоциированные со значением

A

6. Далее поток снова изменяет значение

x

A

7. Поток 1 возобновляется и выполняет сравнение с обменом для переменной

x

A

A

А

Ни один из представленных в этой главе алгоритмов не страдает от этой проблемы, но совсем нетрудно написать свободный от блокировок алгоритм, в котором она проявится. Самый распространенный способ избежать ее — хранить вместе с переменной

x

x

x

x

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

7.3.4. Выявляйте циклы активного ожидания и помогайте другим потокам

В последнем примере очереди мы видели, что поток, выполняющий операцию

push()

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

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

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