pools.py 6.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182
  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. self.channel.put(item)
  99. else:
  100. if self.order_as_stack:
  101. self.free_items.appendleft(item)
  102. else:
  103. self.free_items.append(item)
  104. def resize(self, new_size):
  105. """Resize the pool to *new_size*.
  106. Adjusting this number does not affect existing items checked out of
  107. the pool, nor on any greenthreads who are waiting for an item to free
  108. up. Some indeterminate number of :meth:`get`/:meth:`put`
  109. cycles will be necessary before the new maximum size truly matches
  110. the actual operation of the pool.
  111. """
  112. self.max_size = new_size
  113. def free(self):
  114. """Return the number of free items in the pool. This corresponds
  115. to the number of :meth:`get` calls needed to empty the pool.
  116. """
  117. return len(self.free_items) + self.max_size - self.current_size
  118. def waiting(self):
  119. """Return the number of routines waiting for a pool item.
  120. """
  121. return max(0, self.channel.getting() - self.channel.putting())
  122. def create(self):
  123. """Generate a new pool item. In order for the pool to
  124. function, either this method must be overriden in a subclass
  125. or the pool must be constructed with the `create` argument.
  126. It accepts no arguments and returns a single instance of
  127. whatever thing the pool is supposed to contain.
  128. In general, :meth:`create` is called whenever the pool exceeds its
  129. previous high-water mark of concurrently-checked-out-items. In other
  130. words, in a new pool with *min_size* of 0, the very first call
  131. to :meth:`get` will result in a call to :meth:`create`. If the first
  132. caller calls :meth:`put` before some other caller calls :meth:`get`,
  133. then the first item will be returned, and :meth:`create` will not be
  134. called a second time.
  135. """
  136. raise NotImplementedError("Implement in subclass")
  137. class Token(object):
  138. pass
  139. class TokenPool(Pool):
  140. """A pool which gives out tokens (opaque unique objects), which indicate
  141. that the coroutine which holds the token has a right to consume some
  142. limited resource.
  143. """
  144. def create(self):
  145. return Token()