pools.py 6.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186
  1. from __future__ import print_function
  2. import collections
  3. from contextlib import contextmanager
  4. from eventlet import queue
  5. __all__ = ['Pool', 'TokenPool']
  6. class Pool(object):
  7. """
  8. Pool class implements resource limitation and construction.
  9. There are two ways of using Pool: passing a `create` argument or
  10. subclassing. In either case you must provide a way to create
  11. the resource.
  12. When using `create` argument, pass a function with no arguments::
  13. http_pool = pools.Pool(create=httplib2.Http)
  14. If you need to pass arguments, build a nullary function with either
  15. `lambda` expression::
  16. http_pool = pools.Pool(create=lambda: httplib2.Http(timeout=90))
  17. or :func:`functools.partial`::
  18. from functools import partial
  19. http_pool = pools.Pool(create=partial(httplib2.Http, timeout=90))
  20. When subclassing, define only the :meth:`create` method
  21. to implement the desired resource::
  22. class MyPool(pools.Pool):
  23. def create(self):
  24. return MyObject()
  25. If using 2.5 or greater, the :meth:`item` method acts as a context manager;
  26. that's the best way to use it::
  27. with mypool.item() as thing:
  28. thing.dostuff()
  29. The maximum size of the pool can be modified at runtime via
  30. the :meth:`resize` method.
  31. Specifying a non-zero *min-size* argument pre-populates the pool with
  32. *min_size* items. *max-size* sets a hard limit to the size of the pool --
  33. it cannot contain any more items than *max_size*, and if there are already
  34. *max_size* items 'checked out' of the pool, the pool will cause any
  35. greenthread calling :meth:`get` to cooperatively yield until an item
  36. is :meth:`put` in.
  37. """
  38. def __init__(self, min_size=0, max_size=4, order_as_stack=False, create=None):
  39. """*order_as_stack* governs the ordering of the items in the free pool.
  40. If ``False`` (the default), the free items collection (of items that
  41. were created and were put back in the pool) acts as a round-robin,
  42. giving each item approximately equal utilization. If ``True``, the
  43. free pool acts as a FILO stack, which preferentially re-uses items that
  44. have most recently been used.
  45. """
  46. self.min_size = min_size
  47. self.max_size = max_size
  48. self.order_as_stack = order_as_stack
  49. self.current_size = 0
  50. self.channel = queue.LightQueue(0)
  51. self.free_items = collections.deque()
  52. if create is not None:
  53. self.create = create
  54. for x in range(min_size):
  55. self.current_size += 1
  56. self.free_items.append(self.create())
  57. def get(self):
  58. """Return an item from the pool, when one is available. This may
  59. cause the calling greenthread to block.
  60. """
  61. if self.free_items:
  62. return self.free_items.popleft()
  63. self.current_size += 1
  64. if self.current_size <= self.max_size:
  65. try:
  66. created = self.create()
  67. except:
  68. self.current_size -= 1
  69. raise
  70. return created
  71. self.current_size -= 1 # did not create
  72. return self.channel.get()
  73. @contextmanager
  74. def item(self):
  75. """ Get an object out of the pool, for use with with statement.
  76. >>> from eventlet import pools
  77. >>> pool = pools.TokenPool(max_size=4)
  78. >>> with pool.item() as obj:
  79. ... print("got token")
  80. ...
  81. got token
  82. >>> pool.free()
  83. 4
  84. """
  85. obj = self.get()
  86. try:
  87. yield obj
  88. finally:
  89. self.put(obj)
  90. def put(self, item):
  91. """Put an item back into the pool, when done. This may
  92. cause the putting greenthread to block.
  93. """
  94. if self.current_size > self.max_size:
  95. self.current_size -= 1
  96. return
  97. if self.waiting():
  98. try:
  99. self.channel.put(item, block=False)
  100. return
  101. except queue.Full:
  102. pass
  103. if self.order_as_stack:
  104. self.free_items.appendleft(item)
  105. else:
  106. self.free_items.append(item)
  107. def resize(self, new_size):
  108. """Resize the pool to *new_size*.
  109. Adjusting this number does not affect existing items checked out of
  110. the pool, nor on any greenthreads who are waiting for an item to free
  111. up. Some indeterminate number of :meth:`get`/:meth:`put`
  112. cycles will be necessary before the new maximum size truly matches
  113. the actual operation of the pool.
  114. """
  115. self.max_size = new_size
  116. def free(self):
  117. """Return the number of free items in the pool. This corresponds
  118. to the number of :meth:`get` calls needed to empty the pool.
  119. """
  120. return len(self.free_items) + self.max_size - self.current_size
  121. def waiting(self):
  122. """Return the number of routines waiting for a pool item.
  123. """
  124. return max(0, self.channel.getting() - self.channel.putting())
  125. def create(self):
  126. """Generate a new pool item. In order for the pool to
  127. function, either this method must be overriden in a subclass
  128. or the pool must be constructed with the `create` argument.
  129. It accepts no arguments and returns a single instance of
  130. whatever thing the pool is supposed to contain.
  131. In general, :meth:`create` is called whenever the pool exceeds its
  132. previous high-water mark of concurrently-checked-out-items. In other
  133. words, in a new pool with *min_size* of 0, the very first call
  134. to :meth:`get` will result in a call to :meth:`create`. If the first
  135. caller calls :meth:`put` before some other caller calls :meth:`get`,
  136. then the first item will be returned, and :meth:`create` will not be
  137. called a second time.
  138. """
  139. raise NotImplementedError("Implement in subclass")
  140. class Token(object):
  141. pass
  142. class TokenPool(Pool):
  143. """A pool which gives out tokens (opaque unique objects), which indicate
  144. that the coroutine which holds the token has a right to consume some
  145. limited resource.
  146. """
  147. def create(self):
  148. return Token()