queues.py
python
"""Очереди"""
__all__ = ['Queue', 'PriorityQueue', 'LifoQueue', 'QueueFull', 'QueueEmpty']
import collections
import heapq
from . import compat
from . import events
from . import locks
from .coroutines import coroutine
class QueueEmpty(Exception):
"""Исключение, возбуждаемое при вызове Queue.get_nowait() для объекта Queue,
который пуст.
"""
pass
class QueueFull(Exception):
"""Исключение, возбуждаемое при вызове метода Queue.put_nowait() для объекта Queue,
который полон.
"""
pass
class Queue:
"""Очередь, полезная для координации корутин-производителей и корутин-потребителей.
Если maxsize меньше или равно нулю, размер очереди бесконечен. Если это
целое число больше 0, то "yield from put()" будет блокироваться, когда
очередь достигает maxsize, пока элемент не будет извлечён с помощью get().
В отличие от Queue из стандартной библиотеки, вы можете надёжно узнать размер
этой очереди с помощью qsize(), так как ваше однопоточное приложение asyncio не будет
прервано между вызовом qsize() и выполнением операции над очередью.
"""
def __init__(self, maxsize=0, *, loop=None):
if loop is None:
self._loop = events.get_event_loop()
else:
self._loop = loop
self._maxsize = maxsize
# Future.
self._getters = collections.deque()
# Future.
self._putters = collections.deque()
self._unfinished_tasks = 0
self._finished = locks.Event(loop=self._loop)
self._finished.set()
self._init(maxsize)
# Эти три метода можно переопределить в подклассах.
def _init(self, maxsize):
self._queue = collections.deque()
def _get(self):
return self._queue.popleft()
def _put(self, item):
self._queue.append(item)
# Конец переопределяемых методов.
def _wakeup_next(self, waiters):
# Разбудить следующего ожидающего (если есть), который не отменен.
while waiters:
waiter = waiters.popleft()
if not waiter.done():
waiter.set_result(None)
break
def __repr__(self):
return '<{} at {:#x} {}>'.format(
type(self).__name__, id(self), self._format())
def __str__(self):
return '<{} {}>'.format(type(self).__name__, self._format())
def _format(self):
result = 'maxsize={!r}'.format(self._maxsize)
if getattr(self, '_queue', None):
result += ' _queue={!r}'.format(list(self._queue))
if self._getters:
result += ' _getters[{}]'.format(len(self._getters))
if self._putters:
result += ' _putters[{}]'.format(len(self._putters))
if self._unfinished_tasks:
result += ' tasks={}'.format(self._unfinished_tasks)
return result
def qsize(self):
"""Количество элементов в очереди."""
return len(self._queue)
@property
def maxsize(self):
"""Максимальное количество элементов в очереди."""
return self._maxsize
def empty(self):
"""Возвращает True, если очередь пуста, иначе False."""
return not self._queue
def full(self):
"""Возвращает True, если в очереди maxsize элементов.
Примечание: если Queue была инициализирована с maxsize=0 (по умолчанию),
то full() никогда не возвращает True.
"""
if self._maxsize <= 0:
return False
else:
return self.qsize() >= self._maxsize
@coroutine
def put(self, item):
"""Поместить элемент в очередь.
Поместить элемент в очередь. Если очередь полна, ждать, пока не появится
свободное место перед добавлением элемента.
Этот метод является корутиной.
"""
while self.full():
putter = self._loop.create_future()
self._putters.append(putter)
try:
yield from putter
except:
putter.cancel() # На случай, если putter еще не завершен.
if not self.full() and not putter.cancelled():
# Мы были разбужены get_nowait(), но не можем принять
# вызов. Пробуждаем следующего в очереди.
self._wakeup_next(self._putters)
raise
return self.put_nowait(item)
def put_nowait(self, item):
"""Помещает элемент в очередь без блокировки.
Если свободное место недоступно немедленно, вызывает QueueFull.
"""
if self.full():
raise QueueFull
self._put(item)
self._unfinished_tasks += 1
self._finished.clear()
self._wakeup_next(self._getters)
@coroutine
def get(self):
"""Удалить и вернуть элемент из очереди.
Если очередь пуста, ждать, пока элемент не станет доступен.
Этот метод является корутиной.
"""
while self.empty():
getter = self._loop.create_future()
self._getters.append(getter)
try:
yield from getter
except:
getter.cancel() # На случай, если getter еще не завершен.
try:
self._getters.remove(getter)
except ValueError:
pass
if not self.empty() and not getter.cancelled():
# Мы были пробуждены с помощью put_nowait(), но не можем взять
# вызов. Пробуждаем следующего в очереди.
self._wakeup_next(self._getters)
raise
return self.get_nowait()
def get_nowait(self):
"""Удалить и вернуть элемент из очереди.
Возвращает элемент, если он доступен немедленно, в противном случае вызывает QueueEmpty.
"""
if self.empty():
raise QueueEmpty
item = self._get()
self._wakeup_next(self._putters)
return item
def task_done(self):
"""Указывает, что ранее поставленная в очередь задача выполнена.
Используется потребителями очереди. Для каждого вызова get(), использованного для получения задачи,
последующий вызов task_done() сообщает очереди, что обработка задачи завершена.
Если join() в данный момент заблокирован, он возобновится, когда все элементы будут
обработаны (означает, что вызов task_done() был получен для каждого
элемента, который был помещен в очередь через put()).
Вызывает ValueError, если вызван больше раз, чем было элементов, помещенных в очередь.
"""
if self._unfinished_tasks <= 0:
raise ValueError('task_done() called too many times')
self._unfinished_tasks -= 1
if self._unfinished_tasks == 0:
self._finished.set()
@coroutine
def join(self):
"""Блокирует выполнение, пока все элементы в очереди не будут получены и обработаны.
Счетчик незавершенных задач увеличивается при добавлении элемента в очередь.
Счетчик уменьшается, когда потребитель вызывает task_done(), чтобы
указать, что элемент извлечен и вся работа над ним завершена.
Когда счетчик незавершенных задач падает до нуля, join() разблокируется.
"""
if self._unfinished_tasks > 0:
yield from self._finished.wait()
class PriorityQueue(Queue):
"""Подкласс Queue; извлекает элементы в порядке приоритета (сначала наименьший).
Элементы обычно представлены кортежами вида: (номер приоритета, данные).
"""
def _init(self, maxsize):
self._queue = []
def _put(self, item, heappush=heapq.heappush):
heappush(self._queue, item)
def _get(self, heappop=heapq.heappop):
return heappop(self._queue)
class LifoQueue(Queue):
"""Подкласс Queue, который первым извлекает самые последние добавленные элементы."""
def _init(self, maxsize):
self._queue = []
def _put(self, item):
self._queue.append(item)
def _get(self):
return self._queue.pop()
if not compat.PY35:
JoinableQueue = Queue
"""Deprecated alias for Queue."""
__all__.append('JoinableQueue')