Многопоточный Python на примерах: избавляемся от дедлоков

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

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

Примеры кода, которые вы здесь увидите, взяты из библиотеки locklib.

Как можно обнаружить дедлок?

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

1. На этапе статического анализа исходного кода или компиляции.

2. В рантайме.

К сожалению, статический анализ работает только в каких-то строго определённых случаях. Общего решения здесь не существует, поскольку оно было бы эквивалентно решению проблемы останова, а это считается невозможным. Кроме того, даже существующие способы статического обнаружения гораздо лучше работают в статически типизированных языках (например, в Go или C++). Для Python ни одного вменяемого решения при подготовке статьи я не нашёл.

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

Что такое граф ожидания?

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

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

Графы ожидания сегодня широко используются, к примеру, в СУБД (PostgreSQL таким образом ищет дедлоки в транзакциях, а в MySQL так делалось в старых версиях, сейчас же применяется похожий, но несколько модифицированный алгоритм), однако этот метод не лишён недостатков. Главный из них состоит в том, что получение каждой блокировки становится не просто дороже (поскольку включает в себя дополнительные действия), но и требует, по крайней мере на некоторое время, взять глобальную блокировку, которой, в свою очередь, защищён сам граф ожидания. Это приводит к увеличению доли последовательного кода в программе и, согласно закону Амдала, бьёт по выгоде от параллелизации вычислений вообще.

Приступим

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

«Верхушка» класса, объектом которого является граф, выглядит так:

from threading import Lock from collections import defaultdict  from locklib.errors import DeadLockError # Исключение, которое должно подниматься при обнаружении дедлока.   class LocksGraph:     def __init__(self):         self.links = defaultdict(set)         self.lock = Lock()

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

По сути, наш граф должен поддерживать только две базовые операции:

1. Создать связь между двумя потоками. При этом нужно проверить, не образуется ли от этого цикл, — и, если да, поднять DeadLockError.

2. Удалить её.

Добавление связи выглядит так:

def add_link(self, _from, to):     cycle_with = self.search_cycles(to, _from)     if cycle_with is not None:         cycle_with = ', '.join([str(x) for x in cycle_with])         raise DeadLockError(f'A cycle between {_from}th and {to}th threads has been detected. The full path of the cycle: {cycle_with}.')     self.links[_from].add(to)

А удаление — так:

def delete_link(self, _from, to):     if _from in self.links:         if to in self.links[_from]:             self.links[_from].remove(to)         if not self.links[_from]:             del self.links[_from]

Мы могли бы здесь не делать лишних проверок — тогда методы выглядели бы ещё проще.

Ну и центральная часть — обход графа для обнаружения циклов — это всего лишь три коротких метода:

def get_links_from(self, _from):     return self.links[_from]  def dfs(self, path, current_node, target):     path.append(current_node)      neighbors = self.get_links_from(current_node)      if neighbors:         for link in neighbors:             if link == target:                 path.append(target)                 return path             result_of_next_search = self.dfs(path, link, target)             if result_of_next_search is not None:                 return result_of_next_search      path.pop()  def search_cycles(self, _from, to):     return self.dfs([], _from, to)

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

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

«Верхушка» класса лока выглядит так:

from threading import Lock, get_native_id from collections import deque  from locklib.locks.smart_lock.graph import LocksGraph   graph = LocksGraph()  class SmartLock:     def __init__(self, local_graph=graph):         self.graph = local_graph         self.lock = Lock()         self.deque = deque()         self.local_locks = {}

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

self.deque и self.local_locks используются, как мы увидим дальше, для воспроизведения семантики обычного лока, внутри которого находится очередь ожидающих тредов. В словарь self.local_locks каждый поток кладёт экземпляр лока под ключом в виде собственного идентификатора. В очереди self.deque последовательно лежат идентификаторы потоков, которые попытались взять данную блокировку. Каждый новый поток добавляется в очередь, кладёт в словарь свой лок и берёт лок предыдущего потока. Получается что-то вроде этого:

Объект лока предлагает, по сути, два возможных действия: взять лок и отпустить его. В данном случае они представлены методами acquire() и release().

Начнём с acquire(). Алгоритм действий здесь следующий:

1. Взять лок внутри нашего объекта лока. 2. Взять лок на граф. 3. Если данный поток берёт эту блокировку первым (то есть он первый в очереди): 3.1. Положить id текущего потока в очередь. 3.2. Используя id потока в качестве ключа, положить в словарь self.local_locks объект обычного лока и сразу взять его. 4. Если очередь ожидающих потоков не пуста (то есть данный поток не первый, кто взял этот лок): 4.1. Создать связь в графе между идентификатором текущего потока и идентификатором последнего потока в очереди. 4.2. Положить id текущего потока в очередь. 4.3. Используя id потока в качестве ключа, положить в словарь self.local_locks объект обычного лока и сразу взять его. 4.4. Взять из очереди id предыдущего потока и по нему извлечь из словаря объект блокировки, который тот туда положил. 5. Отпустить лок на граф. 6. Отпустить лок внутри нашего лока. 7. Если очередь потоков была не пуста: 7.1. Взять блокировку, оставленную предыдущим потоком, которую мы успели извлечь из словаря на шаге 4.4.

В коде же это выглядит так:

def acquire(self):     id = get_native_id()     previous_element_lock = None      with self.lock:         with self.graph.lock:             if not self.deque:                 self.deque.appendleft(id)                 self.local_locks[id] = Lock()                 self.local_locks[id].acquire()             else:                 previous_element = self.deque[0]                 self.graph.add_link(id, previous_element)                 self.deque.appendleft(id)                 self.local_locks[id] = Lock()                 self.local_locks[id].acquire()                 previous_element_lock = self.local_locks[previous_element]      if previous_element_lock is not None:         previous_element_lock.acquire()

Теперь release(). Начнём снова с алгоритма:

1, 2. Взять локи внутри нашего объекта лока и внутри графа. 3. Удалить id текущего потока из очереди. 4. Достать из словаря self.local_locks объект лока по id текущего потока и сохранить его в переменную. 5. Удалить из словаря self.local_locks ключ, соответствующий id текущего потока. 6. Если за текущим потоком в очереди есть другие потоки: 6.1. Получить из очереди id потока, который попытался взять блокировку следующим. 6.2. Удалить из графа связь между текущим потоком и следующим. 7. Отпустить лок, извлечённый из словаря на шаге 4. 8, 9. Отпустить лок на граф и лок на лок.

И снова код:

def release(self):     id = get_native_id()      with self.lock:         with self.graph.lock:             if id not in self.local_locks:                 raise RuntimeError('Release unlocked lock.')              self.deque.pop()             lock = self.local_locks[id]             del self.local_locks[id]              if len(self.deque) != 0:                 next_element = self.deque[-1]                 self.graph.delete_link(next_element, id)              lock.release()

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

Что с производительностью?

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

За основу возьмём скорость работы стандартного Lock, которую замерим с помощью следующего кода:

from time import perf_counter from threading import Lock  lock = Lock()  t1 = perf_counter()  for _ in range(100000000):     lock.acquire()     lock.release()  print(perf_counter() - t1)

На моём компьютере (MacBook Air на M1) при первом запуске этого кода вышло 18,812158417 секунды на 100 000 000 итераций, то есть примерно 5 315 711 итераций в секунду.

Теперь протестируем наш лок (код отличается только во второй строчке):

from time import perf_counter from locklib import SmartLock as Lock  lock = Lock()  t1 = perf_counter()  for _ in range(100000000):     lock.acquire()     lock.release()  print(perf_counter() - t1)

У меня при первом запуске вышло 216,22566979200002 секунды, то есть примерно 462 479 итераций в секунду. Это чуть более, чем в десять раз медленнее.

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

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

А это можно скачать бесплатно из интернета?

Библиотека доступна в PyPI:

$ pip install locklib

Пробуем импортировать из нее лок и спровоцировать состояние гонки:

from threading import Thread from locklib import SmartLock   lock_1 = SmartLock() lock_2 = SmartLock()  def function_1():   while True:     with lock_1:       with lock_2:         pass  def function_2():   while True:     with lock_2:       with lock_1:         pass  thread_1 = Thread(target=function_1) thread_2 = Thread(target=function_2) thread_1.start() thread_2.start()

Получаем исключение!

... locklib.errors.DeadLockError: A cycle between 1970256th and 1970257th threads has been detected.

Итак, мы разобрались:

1. Какие вообще существуют способы обнаружения дедлоков и что такое граф ожидания.

2. Как реализовать граф ожидания на Python.

3. Как он влияет на производительность в сравнении с обычной блокировкой.

4. Где можно скачать библиотеку с приведённым в статье кодом.

Теперь, когда у вас появилось новое знание, помните, что знание — сила, а большая сила — это большая ответственность.