Lib/Queue.py

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