Блокировки и реентерабельные мьютексы в параллельном программировании на Python
Блокировки (Lock) в Python
В многопоточных приложениях одновременный доступ к общим ресурсам может приводить к состоянию гонки. Для синхронизации доступа в модуле threading предусмотрен примитив Lock. Он позволяет гарантировать, что в каждый момент времени только один поток выполняет критическую секцию кода.
Основное использование Lock с контекстным менеджером
Наиболее простой и безопасный способ - использование объекта Lock через оператор with. Это автоматически захватывает и освобождает блокировку, даже если в блоке возникло исключение.
import threading
lock = threading.Lock()
shared_data = 0
def increment():
global shared_data
with lock:
shared_data += 1
threads = [threading.Thread(target=increment) for _ in range(10)]
for t in threads:
t.start()
for t in threads:
t.join()
print(shared_data) # 10замок python (блокировки (lock) в python)
Такой подход исключает забытый вызов release() и является предпочтительным.
Вариант с явными acquire/release и try/finally
Как гарантировать освобождение блокировки без контекстного менеджера?
Если по какой-то причине использование with невозможно, применяют пару методов acquire() и release() в блоке try...finally.
lock.acquire()
try:
shared_data += 1
finally:
lock.release()Типичная ошибка:
Забыть вызвать release() в случае исключения, что приведет к вечной блокировке других потоков (deadlock). Всегда используйте try...finally или with.
Реентерабельная блокировка RLock
Как поступать, если один и тот же поток должен повторно захватить блокировку?
Обычный Lock не допускает повторного захвата тем же потоком - возникнет deadlock. Для таких ситуаций предназначен RLock (reentrant lock). Он позволяет потоку входить в критическую секцию несколько раз, при этом количество захватов должно быть равно количеству освобождений.
import threading
rlock = threading.RLock()
def recursive_func(n):
with rlock:
if n > 0:
recursive_func(n - 1)
thread = threading.Thread(target=recursive_func, args=(3,))
thread.start()
thread.join()Используется, когда вызов функции происходит внутри другой функции, уже захватившей блокировку.
Проблема:
Чрезмерное использование RLock может скрыть ошибки проектирования. Обычно рекурсивные вызовы с блокировкой стоит пересмотреть.
Блокировка с условием (Condition)
Как организовать ожидание наступления определенного события под блокировкой?
Объект Condition связывает блокировку с возможностью ожидания (wait) и уведомления (notify/notify_all). Это полезно для шаблона «производитель-потребитель».
import threading
cv = threading.Condition()
buffer = []
def producer():
with cv:
buffer.append('data')
cv.notify()
def consumer():
with cv:
while not buffer:
cv.wait()
item = buffer.pop()
print(f"Consumed {item}")Вызов wait() временно освобождает блокировку и ждет уведомления. После пробуждения блокировка снова захватывается.
Частая ошибка:
Использовать wait() без while цикла для проверки условия. Это может привести к ложным пробуждениям (spurious wakeups). Всегда проверяйте условие в цикле.
Собственная реализация блокировки чтения-записи (RWLock)
Как разрешить одновременное чтение, но исключить запись?
Стандартная библиотека Python не содержит готового RWLock, но его можно реализовать с помощью Lock и Condition. Такая блокировка позволяет нескольким потокам читать данные одновременно, но запись блокирует всех.
class RWLock:
def __init__(self):
self._read_lock = threading.Lock()
self._write_lock = threading.Lock()
self._readers = 0
def acquire_read(self):
with self._read_lock:
self._readers += 1
if self._readers == 1:
self._write_lock.acquire()
def release_read(self):
with self._read_lock:
self._readers -= 1
if self._readers == 0:
self._write_lock.release()
def acquire_write(self):
self._write_lock.acquire()
def release_write(self):
self._write_lock.release()Простое решение подходит для сценариев, где чтений гораздо больше, чем записей.
Риск:
Голодание записи: если постоянно приходят читатели, писатель может никогда не получить доступ. В сложных системах используют fair RWLock или приоритеты.
Расширенные примеры работы с блокировками
1. Демонстрация deadlock с двумя блокировками
import threading
lock1 = threading.Lock()
lock2 = threading.Lock()
def worker1():
with lock1:
print("Worker1 захватил lock1")
with lock2:
print("Worker1 захватил lock2")
def worker2():
with lock2:
print("Worker2 захватил lock2")
with lock1:
print("Worker2 захватил lock1")
t1 = threading.Thread(target=worker1)
t2 = threading.Thread(target=worker2)
t1.start()
t2.start()
t1.join()
t2.join()Worker1 захватил lock1 Worker2 захватил lock2 (программа зависает)
Если потоки захватывают блокировки в разном порядке, возникает взаимная блокировка. Решение - всегда захватывать блокировки в одном и том же порядке.
2. Использование RLock для рекурсивного вызова
import threading
lock = threading.RLock()
def update_counter(n):
with lock:
if n > 0:
print(f"Глубина {n}, захват RLock")
update_counter(n - 1)
thread = threading.Thread(target=update_counter, args=(5,))
thread.start()
thread.join()Глубина 5, захват RLock Глубина 4, захват RLock Глубина 3, захват RLock Глубина 2, захват RLock Глубина 1, захват RLock
С Lock тот же код вызвал бы deadlock, так как поток попытается повторно захватить уже занятую им блокировку.
3. Acquire с таймаутом (timeout)
import threading
lock = threading.Lock()
lock.acquire() # захват без таймаута
def try_acquire():
result = lock.acquire(timeout=2)
if result:
print("Блокировка получена")
lock.release()
else:
print("Не удалось получить блокировку за 2 секунды")
thread = threading.Thread(target=try_acquire)
thread.start()
thread.join()
lock.release() # освобождаем в главном потокеНе удалось получить блокировку за 2 секунды
Таймаут позволяет избежать вечного ожидания. Возвращаемое значение True при успешном захвате, False при превышении времени.
4. Несколько потоков ожидают уведомления от Condition
import threading
import time
cv = threading.Condition()
ready = False
def waiter():
with cv:
while not ready:
cv.wait()
print("Поток проснулся")
def setter():
time.sleep(1)
with cv:
global ready
ready = True
cv.notify_all()
threads = [threading.Thread(target=waiter) for _ in range(3)]
for t in threads:
t.start()
threading.Thread(target=setter).start()
for t in threads:
t.join()(через 1 секунду) Поток проснулся Поток проснулся Поток проснулся
notify_all() пробуждает все ожидающие потоки. Каждый повторно проверяет условие в цикле.
5. Собственная реализация RWLock с приоритетом записи (Fair RWLock)
import threading
class FairRWLock:
def __init__(self):
self._lock = threading.Lock()
self._read_cond = threading.Condition(self._lock)
self._write_cond = threading.Condition(self._lock)
self._readers = 0
self._writers_waiting = 0
self._writing = False
def read_acquire(self):
with self._lock:
while self._writers_waiting > 0 or self._writing:
self._read_cond.wait()
self._readers += 1
def read_release(self):
with self._lock:
self._readers -= 1
if self._readers == 0:
self._write_cond.notify()
def write_acquire(self):
with self._lock:
self._writers_waiting += 1
while self._readers > 0 or self._writing:
self._write_cond.wait()
self._writers_waiting -= 1
self._writing = True
def write_release(self):
with self._lock:
self._writing = False
self._read_cond.notify_all()
self._write_cond.notify()Такая реализация отдает приоритет потокам, ожидающим записи: новые читатели не могут начать чтение, пока есть ожидающие писатели. Это предотвращает голодание записи.