Lib/Queue.py

   1 """A multi-producer, multi-consumer queue."""
   2
   3 from time import time as _time, sleep as _sleep
   4
   5 class Empty(Exception):
   6     "Exception raised by Queue.get(block=0)/get_nowait()."
   7     pass
   8
   9 class Full(Exception):
  10     "Exception raised by Queue.put(block=0)/put_nowait()."
  11     pass
  12
  13 class Queue:
  14     def __init__(self, maxsize=0):
  15         """Initialize a queue object with a given maximum size.
  16
  17         If maxsize is <= 0, the queue size is infinite.
  18         """
  19         try:
  20             import thread
  21         except ImportError:
  22             import dummy_thread as thread
  23         self._init(maxsize)
  24         self.mutex = thread.allocate_lock()
  25         self.esema = thread.allocate_lock()
  26         self.esema.acquire()
  27         self.fsema = thread.allocate_lock()
  28
  29     def qsize(self):
  30         """Return the approximate size of the queue (not reliable!)."""
  31         self.mutex.acquire()
  32         n = self._qsize()
  33         self.mutex.release()
  34         return n
  35
  36     def empty(self):
  37         """Return True if the queue is empty, False otherwise (not reliable!)."""
  38         self.mutex.acquire()
  39         n = self._empty()
  40         self.mutex.release()
  41         return n
  42
  43     def full(self):
  44         """Return True if the queue is full, False otherwise (not reliable!)."""
  45         self.mutex.acquire()
  46         n = self._full()
  47         self.mutex.release()
  48         return n
  49
  50     def put(self, item, block=True, timeout=None):
  51         """Put an item into the queue.
  52
  53         If optional args 'block' is true and 'timeout' is None (the default),
  54         block if necessary until a free slot is available. If 'timeout' is
  55         a positive number, it blocks at most 'timeout' seconds and raises
  56         the Full exception if no free slot was available within that time.
  57         Otherwise ('block' is false), put an item on the queue if a free slot
  58         is immediately available, else raise the Full exception ('timeout'
  59         is ignored in that case).
  60         """
  61         if block:
  62             if timeout is None:
  63                 # blocking, w/o timeout, i.e. forever
  64                 self.fsema.acquire()
  65             elif timeout >= 0:
  66                 # waiting max. 'timeout' seconds.
  67                 # this code snipped is from threading.py: _Event.wait():
  68                 # Balancing act:  We can't afford a pure busy loop, so we
  69                 # have to sleep; but if we sleep the whole timeout time,
  70                 # we'll be unresponsive.  The scheme here sleeps very
  71                 # little at first, longer as time goes on, but never longer
  72                 # than 20 times per second (or the timeout time remaining).
  73                 delay = 0.0005 # 500 us -> initial delay of 1 ms
  74                 endtime = _time() + timeout
  75                 while True:
  76                     if self.fsema.acquire(0):
  77                         break
  78                     remaining = endtime - _time()
  79                     if remaining <= 0:  #time is over and no slot was free
  80                         raise Full
  81                     delay = min(delay * 2, remaining, .05)
  82                     _sleep(delay)       #reduce CPU usage by using a sleep
  83             else:
  84                 raise ValueError("'timeout' must be a positive number")
  85         elif not self.fsema.acquire(0):
  86             raise Full
  87         self.mutex.acquire()
  88         release_fsema = True
  89         try:
  90             was_empty = self._empty()
  91             self._put(item)
  92             # If we fail before here, the empty state has
  93             # not changed, so we can skip the release of esema
  94             if was_empty:
  95                 self.esema.release()
  96             # If we fail before here, the queue can not be full, so
  97             # release_full_sema remains True
  98             release_fsema = not self._full()
  99         finally:
 100             # Catching system level exceptions here (RecursionDepth,
 101             # OutOfMemory, etc) - so do as little as possible in terms
 102             # of Python calls.
 103             if release_fsema:
 104                 self.fsema.release()
 105             self.mutex.release()
 106
 107     def put_nowait(self, item):
 108         """Put an item into the queue without blocking.
 109
 110         Only enqueue the item if a free slot is immediately available.
 111         Otherwise raise the Full exception.
 112         """
 113         return self.put(item, False)
 114
 115     def get(self, block=True, timeout=None):
 116         """Remove and return an item from the queue.
 117
 118         If optional args 'block' is true and 'timeout' is None (the default),
 119         block if necessary until an item is available. If 'timeout' is
 120         a positive number, it blocks at most 'timeout' seconds and raises
 121         the Empty exception if no item was available within that time.
 122         Otherwise ('block' is false), return an item if one is immediately
 123         available, else raise the Empty exception ('timeout' is ignored
 124         in that case).
 125         """
 126         if block:
 127             if timeout is None:
 128                 # blocking, w/o timeout, i.e. forever
 129                 self.esema.acquire()
 130             elif timeout >= 0:
 131                 # waiting max. 'timeout' seconds.
 132                 # this code snipped is from threading.py: _Event.wait():
 133                 # Balancing act:  We can't afford a pure busy loop, so we
 134                 # have to sleep; but if we sleep the whole timeout time,
 135                 # we'll be unresponsive.  The scheme here sleeps very
 136                 # little at first, longer as time goes on, but never longer
 137                 # than 20 times per second (or the timeout time remaining).
 138                 delay = 0.0005 # 500 us -> initial delay of 1 ms
 139                 endtime = _time() + timeout
 140                 while 1:
 141                     if self.esema.acquire(0):
 142                         break
 143                     remaining = endtime - _time()
 144                     if remaining <= 0:  #time is over and no element arrived
 145                         raise Empty
 146                     delay = min(delay * 2, remaining, .05)
 147                     _sleep(delay)       #reduce CPU usage by using a sleep
 148             else:
 149                 raise ValueError("'timeout' must be a positive number")
 150         elif not self.esema.acquire(0):
 151             raise Empty
 152         self.mutex.acquire()
 153         release_esema = True
 154         try:
 155             was_full = self._full()
 156             item = self._get()
 157             # If we fail before here, the full state has
 158             # not changed, so we can skip the release of fsema
 159             if was_full:
 160                 self.fsema.release()
 161             # Failure means empty state also unchanged - release_esema
 162             # remains True.
 163             release_esema = not self._empty()
 164         finally:
 165             if release_esema:
 166                 self.esema.release()
 167             self.mutex.release()
 168         return item
 169
 170     def get_nowait(self):
 171         """Remove and return an item from the queue without blocking.
 172
 173         Only get an item if one is immediately available. Otherwise
 174         raise the Empty exception.
 175         """
 176         return self.get(False)
 177
 178     # Override these methods to implement other queue organizations
 179     # (e.g. stack or priority queue).
 180     # These will only be called with appropriate locks held
 181
 182     # Initialize the queue representation
 183     def _init(self, maxsize):
 184         self.maxsize = maxsize
 185         self.queue = []
 186
 187     def _qsize(self):
 188         return len(self.queue)
 189
 190     # Check whether the queue is empty
 191     def _empty(self):
 192         return not self.queue
 193
 194     # Check whether the queue is full
 195     def _full(self):
 196         return self.maxsize > 0 and len(self.queue) == self.maxsize
 197
 198     # Put a new item in the queue
 199     def _put(self, item):
 200         self.queue.append(item)
 201
 202     # Get an item from the queue
 203     def _get(self):
 204         return self.queue.pop(0)