a multiproducer multiconsumer queue import threading import types from collections import deque from heapq import heappush heappop from time import monotonic as time try from queue import simplequeue except importerror simplequeue none all empty full queue priorityqueue lifoqueue simplequeue try from queue import empty except importerror class emptyexception exception raised by queue getblock0getnowait pass class fullexception exception raised by queue putblock0putnowait pass class queue def initself maxsize0 self maxsize maxsize self initmaxsize 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 notempty whenever an item is added to the queue a thread waiting to get is notified then self notempty threading conditionself mutex notify notfull whenever an item is removed from the queue a thread waiting to put is notified then self notfull threading conditionself mutex notify alltasksdone whenever the number of unfinished tasks drops to zero thread waiting to join is notified to resume self alltasksdone threading conditionself mutex self unfinishedtasks 0 def taskdoneself with self alltasksdone unfinished self unfinishedtasks 1 if unfinished 0 if unfinished 0 raise valueerror taskdone called too many times self alltasksdone notifyall self unfinishedtasks unfinished def joinself with self alltasksdone while self unfinishedtasks self alltasksdone wait def qsizeself return true if the queue is empty false otherwise not reliable this method is likely to be removed at some point use qsize 0 as a direct substitute but be aware that either approach risks a race condition where a queue can grow before the result of empty or qsize can be used to create code that needs to wait for all queued tasks to be completed the preferred technique is to use the join method return true if the queue is full false otherwise not reliable this method is likely to be removed at some point use qsize n as a direct substitute but be aware that either approach risks a race condition where a queue can shrink before the result of full or qsize can be used 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 nonnegative 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 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 nonnegative 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 put an item into the queue without blocking only enqueue the item if a free slot is immediately available otherwise raise the full exception remove and return an item from the queue without blocking only get an item if one is immediately available otherwise raise the empty exception 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 put a new item in the queue get an item from the queue variant of queue that retrieves open entries in priority order lowest first entries are typically tuples of the form priority number data variant of queue that retrieves most recently added entries first def initself maxsize self queue def qsizeself return lenself queue def putself item self queue appenditem def getself return self queue pop class pysimplequeue note while this pure python version provides fairness by using a threading semaphore which is itself fair being based on threading condition fairness is not part of the api contract this allows the c version to use a different implementation def initself self queue deque self count threading semaphore0 def putself item blocktrue timeoutnone self queue appenditem self count release def getself blocktrue timeoutnone if timeout is not none and timeout 0 raise valueerror timeout must be a nonnegative number if not self count acquireblock timeout raise empty return self queue popleft def putnowaitself item return self putitem blockfalse def getnowaitself return self getblockfalse def emptyself return the approximate size of the queue not reliable return lenself queue classgetitem classmethodtypes genericalias if simplequeue is none simplequeue pysimplequeue create a queue object with a given maximum size if maxsize is 0 the queue size is infinite 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 notify not_empty whenever an item is added to the queue a thread waiting to get is notified then notify not_full whenever an item is removed from the queue a thread waiting to put is notified then notify all_tasks_done whenever the number of unfinished tasks drops to zero thread waiting to join is notified to resume 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 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 return the approximate size of the queue not reliable return true if the queue is empty false otherwise not reliable this method is likely to be removed at some point use qsize 0 as a direct substitute but be aware that either approach risks a race condition where a queue can grow before the result of empty or qsize can be used to create code that needs to wait for all queued tasks to be completed the preferred technique is to use the join method return true if the queue is full false otherwise not reliable this method is likely to be removed at some point use qsize n as a direct substitute but be aware that either approach risks a race condition where a queue can shrink before the result of full or qsize can be used 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 non negative 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 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 non negative 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 put an item into the queue without blocking only enqueue the item if a free slot is immediately available otherwise raise the full exception remove and return an item from the queue without blocking only get an item if one is immediately available otherwise raise the empty exception 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 put a new item in the queue get an item from the queue variant of queue that retrieves open entries in priority order lowest first entries are typically tuples of the form priority number data variant of queue that retrieves most recently added entries first simple unbounded fifo queue this pure python implementation is not reentrant note while this pure python version provides fairness by using a threading semaphore which is itself fair being based on threading condition fairness is not part of the api contract this allows the c version to use a different implementation put the item on the queue the optional block and timeout arguments are ignored as this method never blocks they are provided for compatibility with the queue class 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 non negative 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 put an item into the queue without blocking this is exactly equivalent to put item block false and is only provided for compatibility with the queue class 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 true if the queue is empty false otherwise not reliable return the approximate size of the queue not reliable
import threading
import types
from collections import deque
from heapq import heappush, heappop
from time import monotonic as time
try:
    from _queue import SimpleQueue
except ImportError:
    SimpleQueue = None

__all__ = ['Empty', 'Full', 'Queue', 'PriorityQueue', 'LifoQueue', 'SimpleQueue']


try:
    from _queue import Empty
except ImportError:
    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:
    

    def __init__(self, maxsize=0):
        self.maxsize = maxsize
        self._init(maxsize)

        
        
        
        
        self.mutex = threading.Lock()

        
        
        self.not_empty = threading.Condition(self.mutex)

        
        
        self.not_full = threading.Condition(self.mutex)

        
        
        self.all_tasks_done = threading.Condition(self.mutex)
        self.unfinished_tasks = 0

    def task_done(self):
        
        with self.all_tasks_done:
            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

    def join(self):
        
        with self.all_tasks_done:
            while self.unfinished_tasks:
                self.all_tasks_done.wait()

    def qsize(self):
        
        with self.mutex:
            return self._qsize()

    def empty(self):
        
        with self.mutex:
            return not self._qsize()

    def full(self):
        
        with self.mutex:
            return 0 < self.maxsize <= self._qsize()

    def put(self, item, block=True, timeout=None):
        
        with self.not_full:
            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 non-negative 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()

    def get(self, block=True, timeout=None):
        
        with self.not_empty:
            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 non-negative 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

    def put_nowait(self, item):
        
        return self.put(item, block=False)

    def get_nowait(self):
        
        return self.get(block=False)

    
    
    

    
    def _init(self, maxsize):
        self.queue = deque()

    def _qsize(self):
        return len(self.queue)

    
    def _put(self, item):
        self.queue.append(item)

    
    def _get(self):
        return self.queue.popleft()

    __class_getitem__ = classmethod(types.GenericAlias)


class PriorityQueue(Queue):
    

    def _init(self, maxsize):
        self.queue = []

    def _qsize(self):
        return len(self.queue)

    def _put(self, item):
        heappush(self.queue, item)

    def _get(self):
        return heappop(self.queue)


class LifoQueue(Queue):
    

    def _init(self, maxsize):
        self.queue = []

    def _qsize(self):
        return len(self.queue)

    def _put(self, item):
        self.queue.append(item)

    def _get(self):
        return self.queue.pop()


class _PySimpleQueue:
    
    
    
    
    

    def __init__(self):
        self._queue = deque()
        self._count = threading.Semaphore(0)

    def put(self, item, block=True, timeout=None):
        
        self._queue.append(item)
        self._count.release()

    def get(self, block=True, timeout=None):
        
        if timeout is not None and timeout < 0:
            raise ValueError("'timeout' must be a non-negative number")
        if not self._count.acquire(block, timeout):
            raise Empty
        return self._queue.popleft()

    def put_nowait(self, item):
        
        return self.put(item, block=False)

    def get_nowait(self):
        
        return self.get(block=False)

    def empty(self):
        
        return len(self._queue) == 0

    def qsize(self):
        
        return len(self._queue)

    __class_getitem__ = classmethod(types.GenericAlias)


if SimpleQueue is None:
    SimpleQueue = _PySimpleQueue