На этом шаге мы рассмотрим решение этой задачи.
Задача
Вы пишете многопоточную программу, где потоки должны получать более чем одну блокировку за раз, избегая дедлока.
Решение
В многопоточных программах обычной причиной возникновения дедлока являются потоки, которые пытаются получить несколько блокировок за раз. Например, если поток получает первую блокировку, но затем блокируется, пытаясь получить вторую блокировку, этот поток может заблокировать выполнение других потоков и заставить программу приостановиться. Одно из решений для избежания дедлока - присвоить каждой блокировке в программе уникальный номер и заставить соблюдать выполнение правила приобретения по порядку: несколько блокировок может быть получено только по возрастанию номеров. Это очень легко реализовать с помощью менеджера контекста:
import threading from contextlib import contextmanager # Локальное для потока состояние для хранения информации # об уже полученных блокировках _local = threading.local() @contextmanager def acquire(*locks): # Сортирует блокировки по идентификатору объекта locks = sorted(locks, key=lambda x: id(x)) # Убеждается, что порядок блокировки ранее # # приобретенных блокировок не нарушен acquired = getattr(_local, 'acquired', []) if acquired and max(id(lock) for lock in acquired) >= id(locks[0]): raise RuntimeError('Lock Order Violation') # Получает все блокировки acquired.extend(locks) _local.acquired = acquired try: for lock in locks: lock.acquire() yield finally: # Освобождает блокировки в порядке, обратном их получению for lock in reversed(locks): lock.release() del acquired[-len(locks):]
Чтобы использовать этот менеджер контекста, вы просто выделяете объекты блокировок обычным образом, но используете функцию acquire(), когда хотите работать с одной или более блокировками. Например:
import threading from deadlock import acquire x_lock = threading.Lock() y_lock = threading.Lock() def thread_1(): while True: with acquire(x_lock, y_lock): print("Thread-1") def thread_2(): while True: with acquire(y_lock, x_lock): print("Thread-2") input('This program runs forever. Press [return] to start, Ctrl-C to exit') t1 = threading.Thread(target=thread_1) t1.daemon = True t1.start() t2 = threading.Thread(target=thread_2) t2.daemon = True t2.start()
Если вы запустите эту программу, то обнаружите, что она отлично работает без остановки и дедлока - даже если приобретение блокировок определено в каждой функции в разном порядке.
Основной секрет данного рецепта - в первой инструкции, которая сортирует блокировки в соответствии с идентификатором объекта. Путем сортировки блокировок мы добиваемся того, что они всегда приобретаются в строгом порядке, независимо от того, как пользователь может передать их в acquire().
Решение использует локальное для каждого потока хранилище, чтобы решить тонкую проблему с определением потенциального дедлока при нескольких вложенных операциях acquire(). Предположим, например, что вы пишете такой код:
import threading import time from deadlock import acquire x_lock = threading.Lock() y_lock = threading.Lock() def thread_1(): while True: with acquire(x_lock): with acquire(y_lock): print("Thread-1") time.sleep(1) def thread_2(): while True: with acquire(y_lock): with acquire(x_lock): print("Thread-2") time.sleep(1) input('This program crashes with an exception. Press [return] to start') t1 = threading.Thread(target=thread_1) t1.daemon = True t1.start() t2 = threading.Thread(target=thread_2) t2.daemon = True t2.start()
Если вы запустите эту версию программы, один из потоков упадет с исключением:
Exception in thread Thread-1:
Traceback (most recent call last):
. . .
RuntimeError: Lock Order Violation
>>>
Это падение вызвано тем, что каждый поток запоминает блокировку, которую он уже приобрел. Функция acquire() проверяет список ранее
приобретенных блокировок и принудительно внедряет ограничение упорядочивания: приобретенные ранее блокировки должны иметь ID объекта,
который меньше, чем у новых приобретаемых блокировок.
Обсуждение
Проблема дедлока в программах с использованием потоков широко известна (это общая тема для всех учебников по операционным системам). Эмпирическое правило: пока вы можете удостовериться, что потоки могут удерживать только одну блокировку в данный момент времени, дедлока в вашей программе не будет. Однако если приобретается несколько блокировок, то все меняется.
Поиск дедлока и восстановление после него - крайне хитрая проблема с небольшим количеством элегантных решений. Например, обычная схема поиска дедлока и восстановления работоспособности подразумевает использование сторожевого таймера. Пока потоки выполняются, они периодически сбрасывают таймер, и пока это работает гладко, все хорошо. Однако если возникает дедлок, сторожевой таймер достигнет порогового значения. В этой точке программа выполняет восстановление, убивая себя и перезапускаясь.
Еще одна стратегия избежания дедлока - это когда операции блокировки выполняются таким образом, что просто не позволяют программе войти в состояние дедлока. Может быть математически доказано, что решение, при котором блокировки приобретаются в строгом порядке увеличения номеров ID объектов, позволяет избежать дедлока. Доказательство мы не приводим (суть в том, что получение блокировок строго в возрастающем порядке позволяет избежать циклических зависимостей блокировок, а это является необходимым условием возникновения дедлока).
В качестве последнего примера приведем классическую задачу о дедлоке, которая называется "проблема обедающих философов". В этой задаче пять философов сидят вокруг стола, на котором стоят пять мисок риса и лежат пять палочек для еды. Каждый философ представляет независимый поток, а каждая палочка - блокировку. В задаче философы либо размышляют, либо едят. Но чтобы поесть риса, философу нужны две палочки. К сожалению, если все философы протянут руку и возьмут палочку слева от себя, все они останутся с одной палочкой и умрут от голода. Ужасная картина.
Используя приведенное решение, мы можем написать простую реализацию проблемы пяти философов, избавленную от дедлока:
import threading from deadlock import acquire # Поток-философ def philosopher(left, right): while True: with acquire(left, right): print(threading.currentThread(), 'eating') # Палочки (представлены блокировками) NSTICKS = 5 chopsticks = [threading.Lock() for n in range(NSTICKS)] # Создаем всех философов for n in range(NSTICKS): t = threading.Thread(target=philosopher, args=(chopsticks[n], chopsticks[(n + 1) % NSTICKS])) t.daemon = True t.start()
И последнее: стоит отметить, что для избежания дедлока все блокирующие операции должны выполняться с использованием нашей функции acquire(). Если какой-то фрагмент кода решит приобрести блокировку напрямую, алгоритм избежания дедлока перестанет работать.
На следующем шаге мы рассмотрим хранение специфичного состояния потока.
