3.2.5. Дополнительные рекомендации, как избежать взаимоблокировок

Взаимоблокировка может возникать не только при захвате мьютексов, хотя это и наиболее распространенная причина. Нарваться на взаимоблокировку можно и тогда, когда есть два потока и никаких мьютексов; достаточно, чтобы каждый поток вызвал функцию jоin() объекта std::thread, связанного с другим потоком. В этом случае ни один поток не сможет продолжить выполнение, потому что будет ждать завершения другого потока, — точно так же, как дети, ссорящиеся по поводу игрушек. Такой простой цикл может возникнуть всюду, где один поток ждет завершения другого для продолжения работы, а этот другой поток одновременно ждет завершения первого. Причем потоков необязательно должно быть два — цикл, в котором участвуют три и более потоков, также приведёт к взаимоблокировке. Общая рекомендация во всех случаях одна: не ждите завершения другого потока, если есть малейшая возможность, что он будет дожидаться вас. Ниже приведены конкретные советы, как выявить и устранить такую возможность.

Избегайте вложенных блокировок

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

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

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

Захватывайте мьютексы в фиксированном порядке

Если без захвата нескольких мьютексов никак не обойтись и захватить их в одной операции типа std::lock не получается, то следует прибегнуть к другому способу — захватывать их во всех потоках в одном и том же порядке. Мы уже говорили об этом в разделе 3.2.4, как о способе избежать взаимоблокировки при захвате двух мьютексов; идея в том, чтобы четко определить порядок захвата и соблюдать его во всех потоках. Иногда это сравнительно просто. Например, в случае стека из раздела 3.2.3 мьютекс хранится в каждом экземпляре стека, но для операций над хранящимися в стеке элементами необходимо вызывать пользовательский код. Однако можно добавить ограничение: никакая операция над хранящимися в стеке данными не должна производить какие-либо действия с самим стеком. Это возлагает определенную ответственность на пользователя стека, но на практике редко бывает, чтобы хранящимся в контейнере данным нужно было обращаться к самому контейнеру, а если такое и случается, то сразу видно. Поэтому бремя ответственности не слишком тяжело.

Но не всегда всё так просто, и пример мы видели при рассмотрении оператора сравнения в разделе 3.2.4. В этом конкретном случае есть возможность захватить мьютексы одновременно, но так бывает не всегда. Пример связанного списка из раздела 3.1 дает еще один способ защитить список — хранить мьютекс в каждом узле. Тогда, чтобы получить доступ к списку, поток должен будет захватить мьютекс для каждого интересующего его узла. Так, чтобы удалить элемент, надо будет захватить мьютексы трех узлов — удаляемого, предшествующего и последующего, — постольку все они так или иначе модифицируются. Аналогично для обхода списка поток должен удерживать мьютекс текущего узла, пока не захватит мьютекс следующего за ним; это гарантирует, что никто не может изменить указатель на следующий узел. Захватив мьютекс следующего узла, можно освободить мьютекс текущего, так как больше он не понадобится.

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

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

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

Пользуйтесь иерархией блокировок

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

Листинг 3.7. Использование иерархии блокировок для предотвращения взаимоблокировки

hierarchical_mutex high_level_mutex(10000); ← (1)

hierarchical_mutex low_level_mutex(5000);   ← (2)

int do_low_level_stuff();

int low_level_func() {

 std::lock_guard<hierarchical_mutex> lk(low_level_mutex); ← (3)

 return do_low_level_stuff();

}

void high_level_stuff(int some_param);

void high_level_func() {

 std::lock_guard<hierarchical_mutex> lk(high_level_mutex); ← (4)

 high_level_stuff(low_level_func());                       ← (5)

}

void thread_a() { ← (6)

 high_level_func();

}

hierarchical_mutex other_mutex(100); ← (7)

void do_other_stuff();

void other_stuff() {

 high_level_func(); ← (8)

 do_other_stuff();

}

void thread_b() { ← (9)

 std::lock_guard<hierarchical_mutex> lk(other_mutex); ← (10)

 other_stuff();

}

Поток thread_a() (6) соблюдает правила и выполняется беспрепятственно. Напротив, поток thread_b() (9) нарушает правила, поэтому во время выполнения столкнется с трудностями. Функция thread_a() вызывает high_level_func(), которая захватывает мьютекс high_level_mutex (4) (со значением уровня иерархии 10000 (1)), а затем вызывает low_level_func() (5) (мьютекс в этот момент уже захвачен), чтобы получить параметр, необходимый функции high_level_stuff(). Далее функция low_level_func() захватывает мьютекс low_level_mutex (3), и в этом нет ничего плохого, так как уровень иерархии для него равен 5000 (2), то есть меньше, чем для high_level_mutex.

С другой стороны, функция thread_b() некорректна. Первым делом она захватывает мьютекс other_mutex (10), для которого уровень иерархии равен всего 100 (7). Это означает, что мьютекс призван защищать только данные очень низкого уровня. Следовательно, когда функция other_stuff() вызывает high_level_func() (8), она нарушает иерархию — high_level_func() пытается захватить мьютекс high_level_mutex, уровень иерархии которого (10000) намного больше текущего уровня иерархии 100. Поэтому hierarchical_mutex сообщит об ошибке, возбудив исключение или аварийно завершив программу. Таким образом, взаимоблокировки между иерархическими мьютексами невозможны, так как они сами следят за порядком захвата. Это означает, что программа не может удерживать одновременно два мьютекса, находящихся на одном уровне иерархии, поэтому в схемах «передачи из рук в руки» требуется, чтобы каждый мьютекс в цепочке имел меньшее значение уровня иерархии, чем предыдущий, — на практике удовлетворить такому требованию не всегда возможно.

На этом примере демонстрируется еще один момент — использование шаблона std::lock_guard<>, конкретизированного определенным пользователем типом мьютекса. Тип hierarchical_mutex не определен в стандарте, но написать его несложно — простая реализация приведена в листинге 3.8. Хотя этот тип определен пользователем, его можно употреблять совместно с std::lock_guard<>, потому что в нем имеются все три функции-члена, необходимые для удовлетворения требований концепции мьютекса: lock(), unlock() и try_lock(). Мы еще не видели, как используется функция try_lock(), но ничего хитрого в ней нет — если мьютекс захвачен другим потоком, то функция сразу возвращает false, а не блокирует вызывающий поток в ожидании освобождения мьютекса. Она может вызываться также из функции std::lock() для реализации алгоритма предотвращения взаимоблокировок.

Листинг 3.8. Простая реализация иерархического мьютекса

class hierarchical_mutex {

 std::mutex internal_mutex;

 unsigned long const hierarchy_value;

 unsigned previous_hierarchy_value;

 static thread_local

  unsigned long this_thread_hierarchy_value;← (1)

 void check_for_hierarchy_violation() {

  if (this_thread_hierarchy_value <= hierarchy_value) ← (2)

  {

   throw std::logic_error("mutex hierarchy violated");

  }

 }

 void update_hierarchy_value() {

  previous_hierarchy_value = this_thread_hierarchy_value; ← (3)

  this_thread_hierarchy_value = hierarchy_value;

 }

public:

 explicit hierarchical_mutex(unsigned long value):

  hierarchy_value(value),

  previous_hierarchy_value(0) {}

 void lock() {

  check_for_hierarchy_violation();

  internal_mutex.lock();    ← (4)

  update_hierarchy_value(); ← (5)

 }

 void unlock() {

  this_thread_hierarchy_value = previous_hierarchy_value; ← (6)

  internal_mutex.unlock();

 }

 bool try_lock() {

  check_for_hierarchy_violation();

  if (!internal_mutex.try_lock()) ← (7)

   return false;

  update_hierarchy_value();

  return true;

 }

};

thread_local unsigned long

hierarchical_mutex::this_thread_hierarchy_value(ULONG_MAX);← (8)

Главное здесь — использование значения типа thread_local для представления уровня иерархии в текущем потоке, this_thread_hierarchy_value (1). Оно инициализируется максимально возможным значением (8), так что в начальный момент можно захватить любой мьютекс. Поскольку переменная имеет тип thread_local, то в каждом потоке хранится отдельная ее копия, то есть состояние этой переменной в одном потоке не зависит от ее состояния в любом другом. Дополнительные сведения о thread_local см. в разделе А.8 приложения А.

Итак, при первом захвате потоком объекта hierarchical_mutex значение this_thread_hierarchy_value в нем будет равно ULONG_MAX. Это число по определению больше любого другого представимого в программе, потому проверка в функции check_for_hierarchy_violation() (2) проходит. Раз так, то функция lock() просто захватывает внутренний мьютекс (4). Успешно выполнив эту операцию, мы можем изменить значение уровня иерархии (5).

Если теперь попытаться захватить другой объект hierarchical_mutex, не освободив первый, то в переменной this_thread_hierarchy_value будет находиться уровень иерархии первого мьютекса. Чтобы проверка (2) завершилась успешно, уровень иерархии второго мьютекса должен быть меньше уровня уже удерживаемого.

Теперь мы должны сохранить предыдущее значение уровня иерархии в текущем потоке, чтобы его можно было восстановить в функции unlock() (6). В противном случае нам больше никогда не удалось бы захватить мьютекс с более высоким уровнем иерархии, даже если поток не удерживает ни одного мьютекса. Поскольку мы сохраняем предыдущий уровень иерархии только в случае, когда удерживаем internal_mutex (3), и восстанавливаем его перед тем, как освободить этот внутренний мьютекс (6), то можем безопасно сохранить его в самом объекте hierarchical_mutex, где его защищает захваченный внутренний мьютекс.

Функция try_lock() работает так же, как lock(), с одним отличием — если вызов try_lock() для internal_mutex завершается ошибкой (7), то мы не владеем мьютексом и, следовательно, не изменяем уровень иерархии, а вместо true возвращаем false.

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

Применение данных рекомендаций не ограничивается блокировками

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

Функция std::lock() и шаблон класса std::lock_guard покрывают большую часть простых случаев блокировки, по иногда этого недостаточно. Поэтому в стандартную библиотеку включен также шаблон std::unique_lock. Подобно std::lock_guard, этот шаблон класса параметризован типом мьютекса и реализует такое же управление блокировками в духе RAII, что и std::lock_guard, однако обладает чуть большей гибкостью.