+++ /dev/null
-"""Queues"""
-
-__all__ = ['Queue', 'PriorityQueue', 'LifoQueue', 'QueueFull', 'QueueEmpty']
-
-import collections
-import heapq
-
-from . import compat
-from . import events
-from . import futures
-from . import locks
-from .coroutines import coroutine, From, Return
-
-
-class QueueEmpty(Exception):
- """Exception raised when Queue.get_nowait() is called on a Queue object
- which is empty.
- """
- pass
-
-
-class QueueFull(Exception):
- """Exception raised when the Queue.put_nowait() method is called on a Queue
- object which is full.
- """
- pass
-
-
-class Queue(object):
- """A queue, useful for coordinating producer and consumer coroutines.
-
- If maxsize is less than or equal to zero, the queue size is infinite. If it
- is an integer greater than 0, then "yield from put()" will block when the
- queue reaches maxsize, until an item is removed by get().
-
- Unlike the standard library Queue, you can reliably know this Queue's size
- with qsize(), since your single-threaded asyncio application won't be
- interrupted between calling qsize() and doing an operation on the Queue.
- """
-
- def __init__(self, maxsize=0, loop=None):
- if loop is None:
- self._loop = events.get_event_loop()
- else:
- self._loop = loop
- self._maxsize = maxsize
-
- # Futures.
- self._getters = collections.deque()
- # Futures
- self._putters = collections.deque()
- self._unfinished_tasks = 0
- self._finished = locks.Event(loop=self._loop)
- self._finished.set()
- self._init(maxsize)
-
- # These three are overridable in subclasses.
-
- def _init(self, maxsize):
- self._queue = collections.deque()
-
- def _get(self):
- return self._queue.popleft()
-
- def _put(self, item):
- self._queue.append(item)
-
- # End of the overridable methods.
-
- def __put_internal(self, item):
- self._put(item)
- self._unfinished_tasks += 1
- self._finished.clear()
-
- def __repr__(self):
- return '<{0} at {1:#x} {2}>'.format(
- type(self).__name__, id(self), self._format())
-
- def __str__(self):
- return '<{0} {1}>'.format(type(self).__name__, self._format())
-
- def _format(self):
- result = 'maxsize={0!r}'.format(self._maxsize)
- if getattr(self, '_queue', None):
- result += ' _queue={0!r}'.format(list(self._queue))
- if self._getters:
- result += ' _getters[{0}]'.format(len(self._getters))
- if self._putters:
- result += ' _putters[{0}]'.format(len(self._putters))
- if self._unfinished_tasks:
- result += ' tasks={0}'.format(self._unfinished_tasks)
- return result
-
- def _consume_done_getters(self):
- # Delete waiters at the head of the get() queue who've timed out.
- while self._getters and self._getters[0].done():
- self._getters.popleft()
-
- def _consume_done_putters(self):
- # Delete waiters at the head of the put() queue who've timed out.
- while self._putters and self._putters[0].done():
- self._putters.popleft()
-
- def qsize(self):
- """Number of items in the queue."""
- return len(self._queue)
-
- @property
- def maxsize(self):
- """Number of items allowed in the queue."""
- return self._maxsize
-
- def empty(self):
- """Return True if the queue is empty, False otherwise."""
- return not self._queue
-
- def full(self):
- """Return True if there are maxsize items in the queue.
-
- Note: if the Queue was initialized with maxsize=0 (the default),
- then full() is never True.
- """
- if self._maxsize <= 0:
- return False
- else:
- return self.qsize() >= self._maxsize
-
- @coroutine
- def put(self, item):
- """Put an item into the queue.
-
- Put an item into the queue. If the queue is full, wait until a free
- slot is available before adding item.
-
- This method is a coroutine.
- """
- self._consume_done_getters()
- if self._getters:
- assert not self._queue, (
- 'queue non-empty, why are getters waiting?')
-
- getter = self._getters.popleft()
- self.__put_internal(item)
-
- # getter cannot be cancelled, we just removed done getters
- getter.set_result(self._get())
-
- elif self._maxsize > 0 and self._maxsize <= self.qsize():
- waiter = futures.Future(loop=self._loop)
-
- self._putters.append(waiter)
- yield From(waiter)
- self._put(item)
-
- else:
- self.__put_internal(item)
-
- def put_nowait(self, item):
- """Put an item into the queue without blocking.
-
- If no free slot is immediately available, raise QueueFull.
- """
- self._consume_done_getters()
- if self._getters:
- assert not self._queue, (
- 'queue non-empty, why are getters waiting?')
-
- getter = self._getters.popleft()
- self.__put_internal(item)
-
- # getter cannot be cancelled, we just removed done getters
- getter.set_result(self._get())
-
- elif self._maxsize > 0 and self._maxsize <= self.qsize():
- raise QueueFull
- else:
- self.__put_internal(item)
-
- @coroutine
- def get(self):
- """Remove and return an item from the queue.
-
- If queue is empty, wait until an item is available.
-
- This method is a coroutine.
- """
- self._consume_done_putters()
- if self._putters:
- assert self.full(), 'queue not full, why are putters waiting?'
- putter = self._putters.popleft()
-
- # When a getter runs and frees up a slot so this putter can
- # run, we need to defer the put for a tick to ensure that
- # getters and putters alternate perfectly. See
- # ChannelTest.test_wait.
- self._loop.call_soon(putter._set_result_unless_cancelled, None)
-
- raise Return(self._get())
-
- elif self.qsize():
- raise Return(self._get())
- else:
- waiter = futures.Future(loop=self._loop)
- self._getters.append(waiter)
- try:
- value = (yield From(waiter))
- raise Return(value)
- except futures.CancelledError:
- # if we get CancelledError, it means someone cancelled this
- # get() coroutine. But there is a chance that the waiter
- # already is ready and contains an item that has just been
- # removed from the queue. In this case, we need to put the item
- # back into the front of the queue. This get() must either
- # succeed without fault or, if it gets cancelled, it must be as
- # if it never happened.
- if waiter.done():
- self._put_it_back(waiter.result())
- raise
-
- def _put_it_back(self, item):
- """
- This is called when we have a waiter to get() an item and this waiter
- gets cancelled. In this case, we put the item back: wake up another
- waiter or put it in the _queue.
- """
- self._consume_done_getters()
- if self._getters:
- assert not self._queue, (
- 'queue non-empty, why are getters waiting?')
-
- getter = self._getters.popleft()
- self.__put_internal(item)
-
- # getter cannot be cancelled, we just removed done getters
- getter.set_result(item)
- else:
- self._queue.appendleft(item)
-
- def get_nowait(self):
- """Remove and return an item from the queue.
-
- Return an item if one is immediately available, else raise QueueEmpty.
- """
- self._consume_done_putters()
- if self._putters:
- assert self.full(), 'queue not full, why are putters waiting?'
- putter = self._putters.popleft()
- # Wake putter on next tick.
-
- # getter cannot be cancelled, we just removed done putters
- putter.set_result(None)
-
- return self._get()
-
- elif self.qsize():
- return self._get()
- else:
- raise QueueEmpty
-
- def task_done(self):
- """Indicate that a formerly enqueued task is complete.
-
- Used by queue consumers. For each get() used to fetch a task,
- a subsequent call to task_done() tells the queue that the processing
- on the task is complete.
-
- If a join() is currently blocking, it will resume when all items have
- been processed (meaning that a task_done() call was received for every
- item that had been put() into the queue).
-
- Raises ValueError if called more times than there were items placed in
- the queue.
- """
- 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):
- """Block until all items in the queue have been gotten and processed.
-
- The count of unfinished tasks goes up whenever an item is added to the
- queue. The count goes down whenever a consumer calls task_done() to
- indicate that the item was retrieved and all work on it is complete.
- When the count of unfinished tasks drops to zero, join() unblocks.
- """
- if self._unfinished_tasks > 0:
- yield From(self._finished.wait())
-
-
-class PriorityQueue(Queue):
- """A subclass of Queue; retrieves entries in priority order (lowest first).
-
- Entries are typically tuples of the form: (priority number, data).
- """
-
- 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):
- """A subclass of Queue that retrieves most recently added entries first."""
-
- 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')
projects / dotfiles / .git / blobdiff