locks.py
python
"""Примитивы синхронизации."""
__all__ = ('Lock', 'Event', 'Condition', 'Semaphore',
'BoundedSemaphore', 'Barrier')
import collections
import enum
from . import exceptions
from . import mixins
class _ContextManagerMixin:
async def __aenter__(self):
await self.acquire()
# Нам не нужна конструкция "as ..." в with
# инструкция для блокировок.
return None
async def __aexit__(self, exc_type, exc, tb):
self.release()
class Lock(_ContextManagerMixin, mixins._LoopBoundMixin):
"""Примитивные объекты блокировки.
Примитивная блокировка — это примитив синхронизации, который не принадлежит
конкретной задаче, когда заблокирован. Примитивная блокировка находится в одном
из двух состояний: 'locked' или 'unlocked'.
Она создается в разблокированном состоянии. У нее есть два основных метода:
acquire() и release(). Когда состояние разблокировано, acquire()
переводит состояние в заблокированное и немедленно возвращается. Когда
состояние заблокировано, acquire() блокируется до тех пор, пока вызов release() в
другой задаче не переведет его в разблокированное, затем вызов acquire()
возвращает его в заблокированное и завершается. Метод release() следует
вызывать только в заблокированном состоянии; он переводит состояние в разблокированное
и немедленно возвращается. Если попытаться освободить разблокированную
блокировку, будет возбуждено RuntimeError.
Когда более одной задачи заблокировано в acquire(), ожидая
перехода состояния в разблокированное, только одна задача продолжает выполнение, когда
вызов release() переводит состояние в разблокированное; последующие вызовы release()
будут разблокировать задачи в порядке FIFO.
Блокировки также поддерживают асинхронный протокол управления контекстом.
Следует использовать оператор 'async with lock'.
Использование:
lock = Lock()
...
await lock.acquire()
try:
...
finally:
lock.release()
Использование с менеджером контекста:
lock = Lock()
...
async with lock:
...
Объекты блокировки можно проверить на состояние блокировки:
if not lock.locked():
await lock.acquire()
else:
# блокировка захвачена
...
"""
def __init__(self):
self._waiters = None
self._locked = False
def __repr__(self):
res = super().__repr__()
extra = 'locked' if self._locked else 'unlocked'
if self._waiters:
extra = f'{extra}, waiters:{len(self._waiters)}'
return f'<{res[1:-1]} [{extra}]>'
def locked(self):
"""Возвращает True, если блокировка получена."""
return self._locked
async def acquire(self):
"""Получить блокировку.
Этот метод блокируется до тех пор, пока блокировка не будет освобождена, затем переводит её в состояние заблокировано и возвращает True.
"""
# Реализовать честное планирование, где поток всегда ожидает
# своей очереди. Перепрыгивание очереди, если все отменены, является оптимизацией.
if (not self._locked and (self._waiters is None or
all(w.cancelled() for w in self._waiters))):
self._locked = True
return True
if self._waiters is None:
self._waiters = collections.deque()
fut = self._get_loop().create_future()
self._waiters.append(fut)
try:
try:
await fut
finally:
self._waiters.remove(fut)
except exceptions.CancelledError:
# В настоящее время единственное исключение, спроектированное так, чтобы возникать здесь.
# Обеспечить инвариант блокировки: если блокировка не захвачена (или собирается
# быть захваченной нами) и в ожидающих есть задача,
# обеспечить, что задача во главе списка выполнится.
if not self._locked:
self._wake_up_first()
raise
# assert self._locked is False
self._locked = True
return True
def release(self):
"""Освободить блокировку.
Когда блокировка заблокирована, сбросить ее в разблокированное состояние и завершиться.
Если какие-либо другие задачи заблокированы в ожидании разблокировки блокировки,
разрешить ровно одной из них продолжить.
При вызове на разблокированной блокировке возбуждается RuntimeError.
Возвращаемого значения нет.
"""
if self._locked:
self._locked = False
self._wake_up_first()
else:
raise RuntimeError('Lock is not acquired.')
def _wake_up_first(self):
"""Обеспечить, что первый ожидающий проснется."""
if not self._waiters:
return
fut = next(iter(self._waiters))
# .done() означает, что ожидающий уже настроен на пробуждение.
if not fut.done():
fut.set_result(True)
class Event(mixins._LoopBoundMixin):
"""Асинхронный эквивалент threading.Event.
Класс, реализующий объекты событий. Событие управляет флагом, который может быть
установлен в true методом set() и сброшен в false методом clear().
Метод wait() блокируется до тех пор, пока флаг не станет true. Флаг изначально
false.
"""
def __init__(self):
self._waiters = collections.deque()
self._value = False
def __repr__(self):
res = super().__repr__()
extra = 'set' if self._value else 'unset'
if self._waiters:
extra = f'{extra}, waiters:{len(self._waiters)}'
return f'<{res[1:-1]} [{extra}]>'
def is_set(self):
"""Возвращает True тогда и только тогда, когда внутренний флаг равен true."""
return self._value
def set(self):
"""Установить внутренний флаг в true. Все задачи, ожидающие его
установки в true, пробуждаются. Задачи, вызывающие wait(), когда флаг уже true,
не будут блокироваться вообще.
"""
if not self._value:
self._value = True
for fut in self._waiters:
if not fut.done():
fut.set_result(True)
def clear(self):
"""Сбросить внутренний флаг в false. После этого задачи, вызывающие
wait(), будут блокироваться до тех пор, пока set() не будет вызван для установки
внутреннего флага снова в true."""
self._value = False
async def wait(self):
"""Блокироваться, пока внутренний флаг не станет true.
Если при входе внутренний флаг true, немедленно вернуть True.
Иначе блокироваться, пока другая задача не вызовет
set() для установки флага в true, затем вернуть True.
"""
if self._value:
return True
fut = self._get_loop().create_future()
self._waiters.append(fut)
try:
await fut
return True
finally:
self._waiters.remove(fut)
class Condition(_ContextManagerMixin, mixins._LoopBoundMixin):
"""Асинхронный эквивалент threading.Condition.
Этот класс реализует объекты условных переменных. Условная переменная
позволяет одной или нескольким задачам ждать, пока они не будут уведомлены другой
задачей.
Создается новый объект Lock, который используется как базовая блокировка.
"""
def __init__(self, lock=None):
if lock is None:
lock = Lock()
self._lock = lock
# Экспорт методов locked(), acquire() и release() блокировки.
self.locked = lock.locked
self.acquire = lock.acquire
self.release = lock.release
self._waiters = collections.deque()
def __repr__(self):
res = super().__repr__()
extra = 'locked' if self.locked() else 'unlocked'
if self._waiters:
extra = f'{extra}, waiters:{len(self._waiters)}'
return f'<{res[1:-1]} [{extra}]>'
async def wait(self):
"""Ожидать до уведомления.
Если вызывающая задача не захватила блокировку при вызове этого метода, возникает RuntimeError.
Этот метод освобождает базовую блокировку, а затем блокируется, пока не будет разбужен вызовом notify() или notify_all() для той же переменной условия в другой задаче. После пробуждения он повторно захватывает блокировку и возвращает True.
Этот метод может вернуться ложно (spuriously), поэтому вызывающий код всегда должен повторно проверять состояние и быть готовым снова вызвать wait().
"""
if not self.locked():
raise RuntimeError('cannot wait on un-acquired lock')
fut = self._get_loop().create_future()
self.release()
try:
try:
self._waiters.append(fut)
try:
await fut
return True
finally:
self._waiters.remove(fut)
finally:
# Необходимо повторно захватить блокировку, даже если ожидание было отменено.
# Мы перехватываем здесь только CancelledError, так как не хотим, чтобы
# другие (фатальные) ошибки с future заставляли нас крутиться в цикле.
err = None
while True:
try:
await self.acquire()
break
except exceptions.CancelledError as e:
err = e
if err is not None:
try:
raise err # Повторно возбудить последний экземпляр исключения.
finally:
err = None # Разорвать циклические ссылки.
except BaseException:
# Любая ошибка, возникшая здесь, _может_ произойти после того, как эта задача
# считалась успешно уведомлённой.
# Убедитесь, что вместо этого уведомляется другая задача. Это может привести
# к "ложному пробуждению", что допускается в рамках
# протокола переменной условия.
self._notify(1)
raise
async def wait_for(self, predicate):
"""Ожидать, пока предикат не станет истинным.
Предикат должен быть вызываемым, чей результат будет интерпретироваться как булево значение. Метод будет повторно вызывать wait(), пока предикат не станет истинным. Конечное значение предиката является возвращаемым значением.
"""
result = predicate()
while not result:
await self.wait()
result = predicate()
return result
def notify(self, n=1):
"""По умолчанию пробуждает одну задачу, ожидающую на этой переменной условия, если такая есть.
Если вызывающая задача не захватила блокировку при вызове этого метода, возникает RuntimeError.
Этот метод пробуждает n задач, ожидающих переменную условия; если ожидает меньше n, пробуждаются все.
Примечание: пробуждённая задача фактически не возвращается из вызова wait(), пока не сможет повторно захватить блокировку. Поскольку notify() не освобождает блокировку, это должен сделать его вызывающий код.
"""
if not self.locked():
raise RuntimeError('cannot notify on un-acquired lock')
self._notify(n)
def _notify(self, n):
idx = 0
for fut in self._waiters:
if idx >= n:
break
if not fut.done():
idx += 1
fut.set_result(False)
def notify_all(self):
"""Пробуждает все задачи, ожидающие на этом условии. Этот метод действует
как notify(), но пробуждает все ожидающие задачи вместо одной. Если
вызывающая задача не захватила блокировку при вызове этого метода,
возникает исключение RuntimeError.
"""
self.notify(len(self._waiters))
class Semaphore(_ContextManagerMixin, mixins._LoopBoundMixin):
"""Реализация семафора.
Семафор управляет внутренним счётчиком, который уменьшается при каждом вызове acquire() и увеличивается при каждом вызове release(). Счётчик никогда не может стать меньше нуля; когда acquire() обнаруживает, что он равен нулю, он блокируется, ожидая, пока другой поток вызовет release().
Семафоры также поддерживают протокол управления контекстом.
Необязательный аргумент задаёт начальное значение внутреннего счётчика; по умолчанию 1. Если значение меньше 0, возникает ValueError.
"""
def __init__(self, value=1):
if value < 0:
raise ValueError("Semaphore initial value must be >= 0")
self._waiters = None
self._value = value
def __repr__(self):
res = super().__repr__()
extra = 'locked' if self.locked() else f'unlocked, value:{self._value}'
if self._waiters:
extra = f'{extra}, waiters:{len(self._waiters)}'
return f'<{res[1:-1]} [{extra}]>'
def locked(self):
"""Возвращает True, если семафор не может быть захвачен немедленно."""
# Из-за состояния или правил FIFO (необходимо дать другим выполниться первыми).
return self._value == 0 or (
any(not w.cancelled() for w in (self._waiters or ())))
async def acquire(self):
"""Захватить семафор.
Если внутренний счётчик больше нуля при входе, уменьшить его на единицу и сразу вернуть True. Если он равен нулю при входе, заблокироваться, ожидая, пока другая задача не вызовет release(), чтобы сделать его больше 0, а затем вернуть True.
"""
if not self.locked():
# Соблюдать FIFO, ожидать, пока другие начнут, даже если _value > 0.
self._value -= 1
return True
if self._waiters is None:
self._waiters = collections.deque()
fut = self._get_loop().create_future()
self._waiters.append(fut)
try:
try:
await fut
finally:
self._waiters.remove(fut)
except exceptions.CancelledError:
# В настоящее время единственное исключение, спроектированное так, чтобы возникать здесь.
if fut.done() and not fut.cancelled():
# Наш Future был успешно установлен в True через _wake_up_next(),
# но мы не собираемся успешно выполнить acquire(). Поэтому мы
# должны отменить уже сделанные учётные записи и попытаться разбудить
# кого-то другого.
self._value += 1
raise
finally:
# Новые ожидающие могли прибыть, но им пришлось ждать из-за FIFO.
# Разбудить столько, сколько разрешено.
while self._value > 0:
if not self._wake_up_next():
break # Некого было будить.
return True
def release(self):
"""Освободить семафор, увеличив внутренний счётчик на единицу.
Когда он был равен нулю на входе и другая задача ожидает, пока он
снова станет больше нуля, разбудить эту задачу.
"""
self._value += 1
self._wake_up_next()
def _wake_up_next(self):
"""Пробудить первого ожидающего, который ещё не завершён."""
if not self._waiters:
return False
for fut in self._waiters:
if not fut.done():
self._value -= 1
fut.set_result(True)
# `fut` теперь имеет статус `done()` и не является `cancelled()`.
return True
return False
class BoundedSemaphore(Semaphore):
"""Реализация ограниченного семафора.
Этот класс возбуждает ValueError в release(), если это увеличило бы значение
выше начального.
"""
def __init__(self, value=1):
self._bound_value = value
super().__init__(value)
def release(self):
if self._value >= self._bound_value:
raise ValueError('BoundedSemaphore released too many times')
super().release()
class _BarrierState(enum.Enum):
FILLING = 'filling'
DRAINING = 'draining'
RESETTING = 'resetting'
BROKEN = 'broken'
class Barrier(mixins._LoopBoundMixin):
"""Asyncio эквивалент threading.Barrier
Реализует примитив «Барьер».
Полезен для синхронизации фиксированного числа задач в известных точках синхронизации.
Задачи блокируются на 'wait()' и одновременно пробуждаются, когда все они вызывают этот метод.
"""
def __init__(self, parties):
"""Создать барьер, инициализированный для 'parties' задач."""
if parties < 1:
raise ValueError('parties must be >= 1')
self._cond = Condition() # уведомить все задачи при изменении состояния
self._parties = parties
self._state = _BarrierState.FILLING
self._count = 0 # подсчет задач в барьере
def __repr__(self):
res = super().__repr__()
extra = f'{self._state.value}'
if not self.broken:
extra += f', waiters:{self.n_waiting}/{self.parties}'
return f'<{res[1:-1]} [{extra}]>'
async def __aenter__(self):
# ждать, пока барьер достигнет числа parties
# при начале освобождения отпустить и вернуть индекс ожидающей задачи
return await self.wait()
async def __aexit__(self, *args):
pass
async def wait(self):
"""Ожидать барьер.
Когда указанное количество задач начали ожидание, они все
одновременно пробуждаются.
Возвращает уникальный и индивидуальный индексный номер от 0 до 'parties-1'.
"""
async with self._cond:
await self._block() # Блокировать, пока барьер освобождается или сбрасывается.
try:
index = self._count
self._count += 1
if index + 1 == self._parties:
# Мы освобождаем барьер
await self._release()
else:
await self._wait()
return index
finally:
self._count -= 1
# Пробудить все задачи, ожидающие освобождения барьера.
self._exit()
async def _block(self):
# Блокироваться, пока барьер не будет готов для нас,
# или вызвать исключение, если он сломан.
#
# Он освобождается или сбрасывается, ждать до завершения
# если только не возникнет CancelledError
await self._cond.wait_for(
lambda: self._state not in (
_BarrierState.DRAINING, _BarrierState.RESETTING
)
)
# проверить, находится ли барьер в сломанном состоянии
if self._state is _BarrierState.BROKEN:
raise exceptions.BrokenBarrierError("Barrier aborted")
async def _release(self):
# Освободить задачи, ожидающие на барьере.
# Войти в состояние слива.
# Следующие ожидающие задачи будут заблокированы до завершения слива.
self._state = _BarrierState.DRAINING
self._cond.notify_all()
async def _wait(self):
# Ожидание на барьере до тех пор, пока нас не освободят. Вызвать исключение
# если барьер сброшен или сломан.
# ожидание завершения заполнения
# если только не возникнет CancelledError
await self._cond.wait_for(lambda: self._state is not _BarrierState.FILLING)
if self._state in (_BarrierState.BROKEN, _BarrierState.RESETTING):
raise exceptions.BrokenBarrierError("Abort or reset of barrier")
def _exit(self):
# Если мы являемся последними задачами, выходящими из барьера, оповестить любые задачи
# ожидающие слива барьера.
if self._count == 0:
if self._state in (_BarrierState.RESETTING, _BarrierState.DRAINING):
self._state = _BarrierState.FILLING
self._cond.notify_all()
async def reset(self):
"""Сбросить барьер в исходное состояние.
Любые задачи, ожидающие в данный момент, получат исключение BrokenBarrier.
"""
async with self._cond:
if self._count > 0:
if self._state is not _BarrierState.RESETTING:
#сбросить барьер, пробуждая задачи
self._state = _BarrierState.RESETTING
else:
self._state = _BarrierState.FILLING
self._cond.notify_all()
async def abort(self):
"""Перевести барьер в сломанное состояние.
Полезно в случае ошибки. Любые ожидающие задачи и задачи, пытающиеся
вызвать 'wait()', возбудят BrokenBarrierError.
"""
async with self._cond:
self._state = _BarrierState.BROKEN
self._cond.notify_all()
@property
def parties(self):
"""Возвращает количество задач, необходимое для срабатывания барьера."""
return self._parties
@property
def n_waiting(self):
"""Возвращает количество задач, ожидающих в данный момент на барьере."""
if self._state is _BarrierState.FILLING:
return self._count
return 0
@property
def broken(self):
"""Возвращает True, если барьер находится в сломанном состоянии."""
return self._state is _BarrierState.BROKEN