Lib/Queue.py

   1 """A multi-producer, multi-consumer queue."""
   2
   3 class Empty(Exception):
   4     "Exception raised by Queue.get(block=0)/get_nowait()."
   5     pass
   6
   7 class Full(Exception):
   8     "Exception raised by Queue.put(block=0)/put_nowait()."
   9     pass
  10
  11 class Queue:
  12     def __init__(self, maxsize=0):
  13         """Initialize a queue object with a given maximum size.
  14
  15         If maxsize is <= 0, the queue size is infinite.
  16         """
  17         import thread
  18         self._init(maxsize)
  19         self.mutex = thread.allocate_lock()
  20         self.esema = thread.allocate_lock()
  21         self.esema.acquire()
  22         self.fsema = thread.allocate_lock()
  23
  24     def qsize(self):
  25         """Return the approximate size of the queue (not reliable!)."""
  26         self.mutex.acquire()
  27         n = self._qsize()
  28         self.mutex.release()
  29         return n
  30
  31     def empty(self):
  32         """Return 1 if the queue is empty, 0 otherwise (not reliable!)."""
  33         self.mutex.acquire()
  34         n = self._empty()
  35         self.mutex.release()
  36         return n
  37
  38     def full(self):
  39         """Return 1 if the queue is full, 0 otherwise (not reliable!)."""
  40         self.mutex.acquire()
  41         n = self._full()
  42         self.mutex.release()
  43         return n
  44
  45     def put(self, item, block=1):
  46         """Put an item into the queue.
  47
  48         If optional arg 'block' is 1 (the default), block if
  49         necessary until a free slot is available.  Otherwise (block
  50         is 0), put an item on the queue if a free slot is immediately
  51         available, else raise the Full exception.
  52         """
  53         if block:
  54             self.fsema.acquire()
  55         elif not self.fsema.acquire(0):
  56             raise Full
  57         self.mutex.acquire()
  58         was_empty = self._empty()
  59         self._put(item)
  60         if was_empty:
  61             self.esema.release()
  62         if not self._full():
  63             self.fsema.release()
  64         self.mutex.release()
  65
  66     def put_nowait(self, item):
  67         """Put an item into the queue without blocking.
  68
  69         Only enqueue the item if a free slot is immediately available.
  70         Otherwise raise the Full exception.
  71         """
  72         return self.put(item, 0)
  73
  74     def get(self, block=1):
  75         """Remove and return an item from the queue.
  76
  77         If optional arg 'block' is 1 (the default), block if
  78         necessary until an item is available.  Otherwise (block is 0),
  79         return an item if one is immediately available, else raise the
  80         Empty exception.
  81         """
  82         if block:
  83             self.esema.acquire()
  84         elif not self.esema.acquire(0):
  85             raise Empty
  86         self.mutex.acquire()
  87         was_full = self._full()
  88         item = self._get()
  89         if was_full:
  90             self.fsema.release()
  91         if not self._empty():
  92             self.esema.release()
  93         self.mutex.release()
  94         return item
  95
  96     def get_nowait(self):
  97         """Remove and return an item from the queue without blocking.
  98
  99         Only get an item if one is immediately available.  Otherwise
 100         raise the Empty exception.
 101         """
 102         return self.get(0)
 103
 104     # Override these methods to implement other queue organizations
 105     # (e.g. stack or priority queue).
 106     # These will only be called with appropriate locks held
 107
 108     # Initialize the queue representation
 109     def _init(self, maxsize):
 110         self.maxsize = maxsize
 111         self.queue = []
 112
 113     def _qsize(self):
 114         return len(self.queue)
 115
 116     # Check whether the queue is empty
 117     def _empty(self):
 118         return not self.queue
 119
 120     # Check whether the queue is full
 121     def _full(self):
 122         return self.maxsize > 0 and len(self.queue) == self.maxsize
 123
 124     # Put a new item in the queue
 125     def _put(self, item):
 126         self.queue.append(item)
 127
 128     # Get an item from the queue
 129     def _get(self):
 130         item = self.queue[0]
 131         del self.queue[0]
 132         return item