locks.py
python
"""Примитивы синхронизации."""
__all__ = ['Lock', 'Event', 'Condition', 'Semaphore', 'BoundedSemaphore']
import collections
from . import compat
from . import events
from . import futures
from .coroutines import coroutine
class _ContextManager:
"""Менеджер контекста.
Это позволяет использовать следующий идиоматический код для захвата и освобождения блокировки вокруг блока:
with (yield from lock):
<block>
в то время как при ошибочном использовании:
with lock:
<block>
вызовет явную ошибку.
"""
def __init__(self, lock):
self._lock = lock
def __enter__(self):
# Нам не нужна конструкция "as ..." в with
# инструкция для блокировок.
return None
def __exit__(self, *args):
try:
self._lock.release()
finally:
self._lock = None # Грубо предотвращает повторное использование.
class _ContextManagerMixin:
def __enter__(self):
raise RuntimeError(
'"yield from" should be used as context manager expression')
def __exit__(self, *args):
# Это должно существовать, потому что __enter__ существует, даже если он
# всегда вызывает исключение; так работает with-оператор.
pass
@coroutine
def __iter__(self):
# Это не корутина. Она предназначена для использования идиомы:
#
# with (yield from lock):
<block>
# <block>
#
# как альтернатива:
#
# yield from lock.acquire()
# try:
# <block>
# finally:
# lock.release()
yield from self.acquire()
return _ContextManager(self)
if compat.PY35:
def __await__(self):
# Чтобы "with await lock" работало.
yield from self.acquire()
return _ContextManager(self)
@coroutine
def __aenter__(self):
yield from self.acquire()
# Нам не нужна конструкция "as ..." в with
# инструкция для блокировок.
return None
@coroutine
def __aexit__(self, exc_type, exc, tb):
self.release()
class Lock(_ContextManagerMixin):
"""Примитивные объекты блокировки.
Примитивная блокировка — это примитив синхронизации, который не принадлежит
конкретной корутине, когда заблокирована. Примитивная блокировка находится в одном
из двух состояний: 'заблокирована' или 'разблокирована'.
Она создаётся в разблокированном состоянии. У неё есть два основных метода:
acquire() и release(). Когда состояние разблокировано, acquire()
меняет состояние на заблокирована и немедленно возвращается. Когда
состояние заблокирована, acquire() блокируется до тех пор, пока вызов release() в
другой корутине не изменит его на разблокирована, после чего вызов acquire()
снова устанавливает заблокирована и возвращается. Метод release() должен вызываться
только в состоянии заблокирована; он меняет состояние на разблокирована
и немедленно возвращается. При попытке освободить
разблокированную блокировку будет возбуждено RuntimeError.
Когда более одной корутины заблокированы в acquire() в ожидании
перехода состояния в разблокирована, только одна корутина продолжает выполнение, когда
вызов release() сбрасывает состояние в разблокирована; обрабатывается первая корутина,
которая заблокирована в acquire().
acquire() является корутиной и должна вызываться с 'yield from'.
Блокировки также поддерживают протокол управления контекстом. Выражение '(yield from lock)'
должно использоваться в качестве выражения менеджера контекста.
Использование:
lock = Lock()
...
yield from lock
try:
...
finally:
lock.release()
Использование с менеджером контекста:
lock = Lock()
...
with (yield from lock):
...
Объекты Lock можно проверять на состояние блокировки:
if not lock.locked():
yield from lock
else:
# lock is acquired
...
"""
def __init__(self, *, loop=None):
self._waiters = collections.deque()
self._locked = False
if loop is not None:
self._loop = loop
else:
self._loop = events.get_event_loop()
def __repr__(self):
res = super().__repr__()
extra = 'locked' if self._locked else 'unlocked'
if self._waiters:
extra = '{},waiters:{}'.format(extra, len(self._waiters))
return '<{} [{}]>'.format(res[1:-1], extra)
def locked(self):
"""Возвращает True, если блокировка получена."""
return self._locked
@coroutine
def acquire(self):
"""Получить блокировку.
Этот метод блокируется до тех пор, пока блокировка не будет освобождена, затем переводит её в состояние заблокировано и возвращает True.
"""
if not self._locked and all(w.cancelled() for w in self._waiters):
self._locked = True
return True
fut = self._loop.create_future()
self._waiters.append(fut)
# Блок finally должен вызываться до CancelledError
# обработки, так как мы не хотим, чтобы CancelledError вызывал
# _wake_up_first() и пытался разбудить сам себя.
try:
try:
yield from fut
finally:
self._waiters.remove(fut)
except futures.CancelledError:
if not self._locked:
self._wake_up_first()
raise
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):
"""Разбудить первого ожидающего, если он ещё не завершён."""
try:
fut = next(iter(self._waiters))
except StopIteration:
return
# .done() обязательно означает, что ожидающий проснётся позже и
# либо получит блокировку, либо, если он был отменён и блокировка ещё не была
# захвачена, снова попадёт сюда и разбудит нового ожидающего.
if not fut.done():
fut.set_result(True)
class Event:
"""Асинхронный эквивалент threading.Event.
Класс, реализующий объекты событий. Событие управляет флагом, который может быть установлен
в true методом set() и сброшен в false методом clear().
Метод wait() блокируется до тех пор, пока флаг не станет true. Изначально флаг установлен
в false.
"""
def __init__(self, *, loop=None):
self._waiters = collections.deque()
self._value = False
if loop is not None:
self._loop = loop
else:
self._loop = events.get_event_loop()
def __repr__(self):
res = super().__repr__()
extra = 'set' if self._value else 'unset'
if self._waiters:
extra = '{},waiters:{}'.format(extra, len(self._waiters))
return '<{} [{}]>'.format(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
@coroutine
def wait(self):
"""Блокироваться, пока внутренний флаг не станет true.
Если внутренний флаг уже true на входе, немедленно вернуть True.
В противном случае блокироваться, пока другая корутина не вызовет
set() для установки флага в true, затем вернуть True.
"""
if self._value:
return True
fut = self._loop.create_future()
self._waiters.append(fut)
try:
yield from fut
return True
finally:
self._waiters.remove(fut)
class Condition(_ContextManagerMixin):
"""Асинхронный эквивалент threading.Condition.
Этот класс реализует объекты условных переменных. Условная переменная
позволяет одной или нескольким корутинам ожидать, пока они не будут уведомлены другой
корутиной.
Создаётся новый объект Lock, который используется как базовая блокировка.
"""
def __init__(self, lock=None, *, loop=None):
if loop is not None:
self._loop = loop
else:
self._loop = events.get_event_loop()
if lock is None:
lock = Lock(loop=self._loop)
elif lock._loop is not self._loop:
raise ValueError("loop argument must agree with 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 = '{},waiters:{}'.format(extra, len(self._waiters))
return '<{} [{}]>'.format(res[1:-1], extra)
@coroutine
def wait(self):
"""Ожидание уведомления.
Если вызвавшая корутина не захватила блокировку к моменту вызова этого
метода, возбуждается RuntimeError.
Этот метод освобождает нижележащую блокировку, а затем блокируется
до тех пор, пока не будет пробуждена вызовом notify() или notify_all() для
той же переменной условия в другой корутине. После пробуждения
она повторно захватывает блокировку и возвращает True.
"""
if not self.locked():
raise RuntimeError('cannot wait on un-acquired lock')
self.release()
try:
fut = self._loop.create_future()
self._waiters.append(fut)
try:
yield from fut
return True
finally:
self._waiters.remove(fut)
finally:
# Необходимо повторно захватить блокировку, даже если ожидание отменено
cancelled = False
while True:
try:
yield from self.acquire()
break
except futures.CancelledError:
cancelled = True
if cancelled:
raise futures.CancelledError
@coroutine
def wait_for(self, predicate):
"""Ожидание, пока предикат не станет истинным.
Предикат должен быть вызываемым объектом, результат которого будет
интерпретирован как логическое значение. Итоговое значение предиката является
возвращаемым значением.
"""
result = predicate()
while not result:
yield from self.wait()
result = predicate()
return result
def notify(self, n=1):
"""По умолчанию пробуждает одну корутину, ожидающую это условие, если таковая имеется.
Если вызвавшая корутина не захватила блокировку к моменту вызова этого метода,
возбуждается RuntimeError.
Этот метод пробуждает не более n корутин, ожидающих
переменную условия; он ничего не делает, если ожидающих корутин нет.
Примечание: пробужденная корутина не возвращается из своего
вызова wait() до тех пор, пока не сможет повторно захватить блокировку. Поскольку notify() не
освобождает блокировку, это должен сделать его вызывающий.
"""
if not self.locked():
raise RuntimeError('cannot notify on un-acquired lock')
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):
"""Реализация семафора.
Семафор управляет внутренним счётчиком, который уменьшается при каждом
вызове acquire() и увеличивается при каждом вызове release(). Счётчик
никогда не может стать меньше нуля; когда acquire() обнаруживает, что он равен нулю, он блокируется,
ожидая, пока какой-нибудь другой поток вызовет release().
Семафоры также поддерживают протокол управления контекстом.
Необязательный аргумент задаёт начальное значение внутреннего
счётчика; по умолчанию 1. Если указанное значение меньше 0,
возбуждается ValueError.
"""
def __init__(self, value=1, *, loop=None):
if value < 0:
raise ValueError("Semaphore initial value must be >= 0")
self._value = value
self._waiters = collections.deque()
if loop is not None:
self._loop = loop
else:
self._loop = events.get_event_loop()
def __repr__(self):
res = super().__repr__()
extra = 'locked' if self.locked() else 'unlocked,value:{}'.format(
self._value)
if self._waiters:
extra = '{},waiters:{}'.format(extra, len(self._waiters))
return '<{} [{}]>'.format(res[1:-1], extra)
def _wake_up_next(self):
while self._waiters:
waiter = self._waiters.popleft()
if not waiter.done():
waiter.set_result(None)
return
def locked(self):
"""Возвращает True, если семафор не может быть захвачен немедленно."""
return self._value == 0
@coroutine
def acquire(self):
"""Захват семафора.
Если внутренний счётчик больше нуля при входе,
уменьшить его на единицу и немедленно вернуть True. Если он
равен нулю при входе, заблокироваться, ожидая, пока другая корутина
вызовет release(), чтобы сделать его больше 0, и затем вернуть
True.
"""
while self._value <= 0:
fut = self._loop.create_future()
self._waiters.append(fut)
try:
yield from fut
except:
# Смотрите аналогичный код в Queue.get.
fut.cancel()
if self._value > 0 and not fut.cancelled():
self._wake_up_next()
raise
self._value -= 1
return True
def release(self):
"""Освободить семафор, увеличив внутренний счётчик на единицу.
Если при входе он был равен нулю и другая корутина ожидает, когда он
снова станет больше нуля, разбудить эту корутину.
"""
self._value += 1
self._wake_up_next()
class BoundedSemaphore(Semaphore):
"""Реализация ограниченного семафора.
Этот класс возбуждает ValueError в release(), если это увеличило бы значение
выше начального.
"""
def __init__(self, value=1, *, loop=None):
self._bound_value = value
super().__init__(value, loop=loop)
def release(self):
if self._value >= self._bound_value:
raise ValueError('BoundedSemaphore released too many times')
super().release()