Lib/Queue.py

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