"""A multi-producer, multi-consumer queue.""" | |
from time import time as _time | |
try: | |
import threading as _threading | |
except ImportError: | |
import dummy_threading as _threading | |
from collections import deque | |
import heapq | |
__all__ = ['Empty', 'Full', 'Queue', 'PriorityQueue', 'LifoQueue'] | |
class Empty(Exception): | |
"Exception raised by Queue.get(block=0)/get_nowait()." | |
pass | |
class Full(Exception): | |
"Exception raised by Queue.put(block=0)/put_nowait()." | |
pass | |
class Queue: | |
"""Create a queue object with a given maximum size. | |
If maxsize is <= 0, the queue size is infinite. | |
""" | |
def __init__(self, maxsize=0): | |
self.maxsize = maxsize | |
self._init(maxsize) | |
# mutex must be held whenever the queue is mutating. All methods | |
# that acquire mutex must release it before returning. mutex | |
# is shared between the three conditions, so acquiring and | |
# releasing the conditions also acquires and releases mutex. | |
self.mutex = _threading.Lock() | |
# Notify not_empty whenever an item is added to the queue; a | |
# thread waiting to get is notified then. | |
self.not_empty = _threading.Condition(self.mutex) | |
# Notify not_full whenever an item is removed from the queue; | |
# a thread waiting to put is notified then. | |
self.not_full = _threading.Condition(self.mutex) | |
# Notify all_tasks_done whenever the number of unfinished tasks | |
# drops to zero; thread waiting to join() is notified to resume | |
self.all_tasks_done = _threading.Condition(self.mutex) | |
self.unfinished_tasks = 0 | |
def task_done(self): | |
"""Indicate that a formerly enqueued task is complete. | |
Used by Queue consumer threads. 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 a ValueError if called more times than there were items | |
placed in the queue. | |
""" | |
self.all_tasks_done.acquire() | |
try: | |
unfinished = self.unfinished_tasks - 1 | |
if unfinished <= 0: | |
if unfinished < 0: | |
raise ValueError('task_done() called too many times') | |
self.all_tasks_done.notify_all() | |
self.unfinished_tasks = unfinished | |
finally: | |
self.all_tasks_done.release() | |
def join(self): | |
"""Blocks 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 thread calls task_done() | |
to indicate the item was retrieved and all work on it is complete. | |
When the count of unfinished tasks drops to zero, join() unblocks. | |
""" | |
self.all_tasks_done.acquire() | |
try: | |
while self.unfinished_tasks: | |
self.all_tasks_done.wait() | |
finally: | |
self.all_tasks_done.release() | |
def qsize(self): | |
"""Return the approximate size of the queue (not reliable!).""" | |
self.mutex.acquire() | |
n = self._qsize() | |
self.mutex.release() | |
return n | |
def empty(self): | |
"""Return True if the queue is empty, False otherwise (not reliable!).""" | |
self.mutex.acquire() | |
n = not self._qsize() | |
self.mutex.release() | |
return n | |
def full(self): | |
"""Return True if the queue is full, False otherwise (not reliable!).""" | |
self.mutex.acquire() | |
n = 0 < self.maxsize == self._qsize() | |
self.mutex.release() | |
return n | |
def put(self, item, block=True, timeout=None): | |
"""Put an item into the queue. | |
If optional args 'block' is true and 'timeout' is None (the default), | |
block if necessary until a free slot is available. If 'timeout' is | |
a positive number, it blocks at most 'timeout' seconds and raises | |
the Full exception if no free slot was available within that time. | |
Otherwise ('block' is false), put an item on the queue if a free slot | |
is immediately available, else raise the Full exception ('timeout' | |
is ignored in that case). | |
""" | |
self.not_full.acquire() | |
try: | |
if self.maxsize > 0: | |
if not block: | |
if self._qsize() == self.maxsize: | |
raise Full | |
elif timeout is None: | |
while self._qsize() == self.maxsize: | |
self.not_full.wait() | |
elif timeout < 0: | |
raise ValueError("'timeout' must be a positive number") | |
else: | |
endtime = _time() + timeout | |
while self._qsize() == self.maxsize: | |
remaining = endtime - _time() | |
if remaining <= 0.0: | |
raise Full | |
self.not_full.wait(remaining) | |
self._put(item) | |
self.unfinished_tasks += 1 | |
self.not_empty.notify() | |
finally: | |
self.not_full.release() | |
def put_nowait(self, item): | |
"""Put an item into the queue without blocking. | |
Only enqueue the item if a free slot is immediately available. | |
Otherwise raise the Full exception. | |
""" | |
return self.put(item, False) | |
def get(self, block=True, timeout=None): | |
"""Remove and return an item from the queue. | |
If optional args 'block' is true and 'timeout' is None (the default), | |
block if necessary until an item is available. If 'timeout' is | |
a positive number, it blocks at most 'timeout' seconds and raises | |
the Empty exception if no item was available within that time. | |
Otherwise ('block' is false), return an item if one is immediately | |
available, else raise the Empty exception ('timeout' is ignored | |
in that case). | |
""" | |
self.not_empty.acquire() | |
try: | |
if not block: | |
if not self._qsize(): | |
raise Empty | |
elif timeout is None: | |
while not self._qsize(): | |
self.not_empty.wait() | |
elif timeout < 0: | |
raise ValueError("'timeout' must be a positive number") | |
else: | |
endtime = _time() + timeout | |
while not self._qsize(): | |
remaining = endtime - _time() | |
if remaining <= 0.0: | |
raise Empty | |
self.not_empty.wait(remaining) | |
item = self._get() | |
self.not_full.notify() | |
return item | |
finally: | |
self.not_empty.release() | |
def get_nowait(self): | |
"""Remove and return an item from the queue without blocking. | |
Only get an item if one is immediately available. Otherwise | |
raise the Empty exception. | |
""" | |
return self.get(False) | |
# Override these methods to implement other queue organizations | |
# (e.g. stack or priority queue). | |
# These will only be called with appropriate locks held | |
# Initialize the queue representation | |
def _init(self, maxsize): | |
self.queue = deque() | |
def _qsize(self, len=len): | |
return len(self.queue) | |
# Put a new item in the queue | |
def _put(self, item): | |
self.queue.append(item) | |
# Get an item from the queue | |
def _get(self): | |
return self.queue.popleft() | |
class PriorityQueue(Queue): | |
'''Variant of Queue that retrieves open entries in priority order (lowest first). | |
Entries are typically tuples of the form: (priority number, data). | |
''' | |
def _init(self, maxsize): | |
self.queue = [] | |
def _qsize(self, len=len): | |
return len(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): | |
'''Variant of Queue that retrieves most recently added entries first.''' | |
def _init(self, maxsize): | |
self.queue = [] | |
def _qsize(self, len=len): | |
return len(self.queue) | |
def _put(self, item): | |
self.queue.append(item) | |
def _get(self): | |
return self.queue.pop() |