queue.py 6.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199
  1. # util/queue.py
  2. # Copyright (C) 2005-2017 the SQLAlchemy authors and contributors
  3. # <see AUTHORS file>
  4. #
  5. # This module is part of SQLAlchemy and is released under
  6. # the MIT License: http://www.opensource.org/licenses/mit-license.php
  7. """An adaptation of Py2.3/2.4's Queue module which supports reentrant
  8. behavior, using RLock instead of Lock for its mutex object. The
  9. Queue object is used exclusively by the sqlalchemy.pool.QueuePool
  10. class.
  11. This is to support the connection pool's usage of weakref callbacks to return
  12. connections to the underlying Queue, which can in extremely
  13. rare cases be invoked within the ``get()`` method of the Queue itself,
  14. producing a ``put()`` inside the ``get()`` and therefore a reentrant
  15. condition.
  16. """
  17. from collections import deque
  18. from time import time as _time
  19. from .compat import threading
  20. __all__ = ['Empty', 'Full', 'Queue']
  21. class Empty(Exception):
  22. "Exception raised by Queue.get(block=0)/get_nowait()."
  23. pass
  24. class Full(Exception):
  25. "Exception raised by Queue.put(block=0)/put_nowait()."
  26. pass
  27. class Queue:
  28. def __init__(self, maxsize=0):
  29. """Initialize a queue object with a given maximum size.
  30. If `maxsize` is <= 0, the queue size is infinite.
  31. """
  32. self._init(maxsize)
  33. # mutex must be held whenever the queue is mutating. All methods
  34. # that acquire mutex must release it before returning. mutex
  35. # is shared between the two conditions, so acquiring and
  36. # releasing the conditions also acquires and releases mutex.
  37. self.mutex = threading.RLock()
  38. # Notify not_empty whenever an item is added to the queue; a
  39. # thread waiting to get is notified then.
  40. self.not_empty = threading.Condition(self.mutex)
  41. # Notify not_full whenever an item is removed from the queue;
  42. # a thread waiting to put is notified then.
  43. self.not_full = threading.Condition(self.mutex)
  44. def qsize(self):
  45. """Return the approximate size of the queue (not reliable!)."""
  46. self.mutex.acquire()
  47. n = self._qsize()
  48. self.mutex.release()
  49. return n
  50. def empty(self):
  51. """Return True if the queue is empty, False otherwise (not
  52. reliable!)."""
  53. self.mutex.acquire()
  54. n = self._empty()
  55. self.mutex.release()
  56. return n
  57. def full(self):
  58. """Return True if the queue is full, False otherwise (not
  59. reliable!)."""
  60. self.mutex.acquire()
  61. n = self._full()
  62. self.mutex.release()
  63. return n
  64. def put(self, item, block=True, timeout=None):
  65. """Put an item into the queue.
  66. If optional args `block` is True and `timeout` is None (the
  67. default), block if necessary until a free slot is
  68. available. If `timeout` is a positive number, it blocks at
  69. most `timeout` seconds and raises the ``Full`` exception if no
  70. free slot was available within that time. Otherwise (`block`
  71. is false), put an item on the queue if a free slot is
  72. immediately available, else raise the ``Full`` exception
  73. (`timeout` is ignored in that case).
  74. """
  75. self.not_full.acquire()
  76. try:
  77. if not block:
  78. if self._full():
  79. raise Full
  80. elif timeout is None:
  81. while self._full():
  82. self.not_full.wait()
  83. else:
  84. if timeout < 0:
  85. raise ValueError("'timeout' must be a positive number")
  86. endtime = _time() + timeout
  87. while self._full():
  88. remaining = endtime - _time()
  89. if remaining <= 0.0:
  90. raise Full
  91. self.not_full.wait(remaining)
  92. self._put(item)
  93. self.not_empty.notify()
  94. finally:
  95. self.not_full.release()
  96. def put_nowait(self, item):
  97. """Put an item into the queue without blocking.
  98. Only enqueue the item if a free slot is immediately available.
  99. Otherwise raise the ``Full`` exception.
  100. """
  101. return self.put(item, False)
  102. def get(self, block=True, timeout=None):
  103. """Remove and return an item from the queue.
  104. If optional args `block` is True and `timeout` is None (the
  105. default), block if necessary until an item is available. If
  106. `timeout` is a positive number, it blocks at most `timeout`
  107. seconds and raises the ``Empty`` exception if no item was
  108. available within that time. Otherwise (`block` is false),
  109. return an item if one is immediately available, else raise the
  110. ``Empty`` exception (`timeout` is ignored in that case).
  111. """
  112. self.not_empty.acquire()
  113. try:
  114. if not block:
  115. if self._empty():
  116. raise Empty
  117. elif timeout is None:
  118. while self._empty():
  119. self.not_empty.wait()
  120. else:
  121. if timeout < 0:
  122. raise ValueError("'timeout' must be a positive number")
  123. endtime = _time() + timeout
  124. while self._empty():
  125. remaining = endtime - _time()
  126. if remaining <= 0.0:
  127. raise Empty
  128. self.not_empty.wait(remaining)
  129. item = self._get()
  130. self.not_full.notify()
  131. return item
  132. finally:
  133. self.not_empty.release()
  134. def get_nowait(self):
  135. """Remove and return an item from the queue without blocking.
  136. Only get an item if one is immediately available. Otherwise
  137. raise the ``Empty`` exception.
  138. """
  139. return self.get(False)
  140. # Override these methods to implement other queue organizations
  141. # (e.g. stack or priority queue).
  142. # These will only be called with appropriate locks held
  143. # Initialize the queue representation
  144. def _init(self, maxsize):
  145. self.maxsize = maxsize
  146. self.queue = deque()
  147. def _qsize(self):
  148. return len(self.queue)
  149. # Check whether the queue is empty
  150. def _empty(self):
  151. return not self.queue
  152. # Check whether the queue is full
  153. def _full(self):
  154. return self.maxsize > 0 and len(self.queue) == self.maxsize
  155. # Put a new item in the queue
  156. def _put(self, item):
  157. self.queue.append(item)
  158. # Get an item from the queue
  159. def _get(self):
  160. return self.queue.popleft()