pool.py 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316
  1. from eventlet import coros, proc, api
  2. from eventlet.semaphore import Semaphore
  3. import warnings
  4. warnings.warn("The pool module is deprecated. Please use the "
  5. "eventlet.GreenPool and eventlet.GreenPile classes instead.",
  6. DeprecationWarning, stacklevel=2)
  7. class Pool(object):
  8. def __init__(self, min_size=0, max_size=4, track_events=False):
  9. if min_size > max_size:
  10. raise ValueError('min_size cannot be bigger than max_size')
  11. self.max_size = max_size
  12. self.sem = Semaphore(max_size)
  13. self.procs = proc.RunningProcSet()
  14. if track_events:
  15. self.results = coros.queue()
  16. else:
  17. self.results = None
  18. def resize(self, new_max_size):
  19. """ Change the :attr:`max_size` of the pool.
  20. If the pool gets resized when there are more than *new_max_size*
  21. coroutines checked out, when they are returned to the pool they will be
  22. discarded. The return value of :meth:`free` will be negative in this
  23. situation.
  24. """
  25. max_size_delta = new_max_size - self.max_size
  26. self.sem.counter += max_size_delta
  27. self.max_size = new_max_size
  28. @property
  29. def current_size(self):
  30. """ The number of coroutines that are currently executing jobs. """
  31. return len(self.procs)
  32. def free(self):
  33. """ Returns the number of coroutines that are available for doing
  34. work."""
  35. return self.sem.counter
  36. def execute(self, func, *args, **kwargs):
  37. """Execute func in one of the coroutines maintained
  38. by the pool, when one is free.
  39. Immediately returns a :class:`~eventlet.proc.Proc` object which can be
  40. queried for the func's result.
  41. >>> pool = Pool()
  42. >>> task = pool.execute(lambda a: ('foo', a), 1)
  43. >>> task.wait()
  44. ('foo', 1)
  45. """
  46. # if reentering an empty pool, don't try to wait on a coroutine freeing
  47. # itself -- instead, just execute in the current coroutine
  48. if self.sem.locked() and api.getcurrent() in self.procs:
  49. p = proc.spawn(func, *args, **kwargs)
  50. try:
  51. p.wait()
  52. except:
  53. pass
  54. else:
  55. self.sem.acquire()
  56. p = self.procs.spawn(func, *args, **kwargs)
  57. # assuming the above line cannot raise
  58. p.link(lambda p: self.sem.release())
  59. if self.results is not None:
  60. p.link(self.results)
  61. return p
  62. execute_async = execute
  63. def _execute(self, evt, func, args, kw):
  64. p = self.execute(func, *args, **kw)
  65. p.link(evt)
  66. return p
  67. def waitall(self):
  68. """ Calling this function blocks until every coroutine
  69. completes its work (i.e. there are 0 running coroutines)."""
  70. return self.procs.waitall()
  71. wait_all = waitall
  72. def wait(self):
  73. """Wait for the next execute in the pool to complete,
  74. and return the result."""
  75. return self.results.wait()
  76. def waiting(self):
  77. """Return the number of coroutines waiting to execute.
  78. """
  79. if self.sem.balance < 0:
  80. return -self.sem.balance
  81. else:
  82. return 0
  83. def killall(self):
  84. """ Kill every running coroutine as immediately as possible."""
  85. return self.procs.killall()
  86. def launch_all(self, function, iterable):
  87. """For each tuple (sequence) in *iterable*, launch ``function(*tuple)``
  88. in its own coroutine -- like ``itertools.starmap()``, but in parallel.
  89. Discard values returned by ``function()``. You should call
  90. ``wait_all()`` to wait for all coroutines, newly-launched plus any
  91. previously-submitted :meth:`execute` or :meth:`execute_async` calls, to
  92. complete.
  93. >>> pool = Pool()
  94. >>> def saw(x):
  95. ... print "I saw %s!" % x
  96. ...
  97. >>> pool.launch_all(saw, "ABC")
  98. >>> pool.wait_all()
  99. I saw A!
  100. I saw B!
  101. I saw C!
  102. """
  103. for tup in iterable:
  104. self.execute(function, *tup)
  105. def process_all(self, function, iterable):
  106. """For each tuple (sequence) in *iterable*, launch ``function(*tuple)``
  107. in its own coroutine -- like ``itertools.starmap()``, but in parallel.
  108. Discard values returned by ``function()``. Don't return until all
  109. coroutines, newly-launched plus any previously-submitted :meth:`execute()`
  110. or :meth:`execute_async` calls, have completed.
  111. >>> from eventlet import coros
  112. >>> pool = coros.CoroutinePool()
  113. >>> def saw(x): print "I saw %s!" % x
  114. ...
  115. >>> pool.process_all(saw, "DEF")
  116. I saw D!
  117. I saw E!
  118. I saw F!
  119. """
  120. self.launch_all(function, iterable)
  121. self.wait_all()
  122. def generate_results(self, function, iterable, qsize=None):
  123. """For each tuple (sequence) in *iterable*, launch ``function(*tuple)``
  124. in its own coroutine -- like ``itertools.starmap()``, but in parallel.
  125. Yield each of the values returned by ``function()``, in the order
  126. they're completed rather than the order the coroutines were launched.
  127. Iteration stops when we've yielded results for each arguments tuple in
  128. *iterable*. Unlike :meth:`wait_all` and :meth:`process_all`, this
  129. function does not wait for any previously-submitted :meth:`execute` or
  130. :meth:`execute_async` calls.
  131. Results are temporarily buffered in a queue. If you pass *qsize=*, this
  132. value is used to limit the max size of the queue: an attempt to buffer
  133. too many results will suspend the completed :class:`CoroutinePool`
  134. coroutine until the requesting coroutine (the caller of
  135. :meth:`generate_results`) has retrieved one or more results by calling
  136. this generator-iterator's ``next()``.
  137. If any coroutine raises an uncaught exception, that exception will
  138. propagate to the requesting coroutine via the corresponding ``next()``
  139. call.
  140. What I particularly want these tests to illustrate is that using this
  141. generator function::
  142. for result in generate_results(function, iterable):
  143. # ... do something with result ...
  144. pass
  145. executes coroutines at least as aggressively as the classic eventlet
  146. idiom::
  147. events = [pool.execute(function, *args) for args in iterable]
  148. for event in events:
  149. result = event.wait()
  150. # ... do something with result ...
  151. even without a distinct event object for every arg tuple in *iterable*,
  152. and despite the funny flow control from interleaving launches of new
  153. coroutines with yields of completed coroutines' results.
  154. (The use case that makes this function preferable to the classic idiom
  155. above is when the *iterable*, which may itself be a generator, produces
  156. millions of items.)
  157. >>> from eventlet import coros
  158. >>> import string
  159. >>> pool = coros.CoroutinePool(max_size=5)
  160. >>> pausers = [coros.Event() for x in xrange(2)]
  161. >>> def longtask(evt, desc):
  162. ... print "%s woke up with %s" % (desc, evt.wait())
  163. ...
  164. >>> pool.launch_all(longtask, zip(pausers, "AB"))
  165. >>> def quicktask(desc):
  166. ... print "returning %s" % desc
  167. ... return desc
  168. ...
  169. (Instead of using a ``for`` loop, step through :meth:`generate_results`
  170. items individually to illustrate timing)
  171. >>> step = iter(pool.generate_results(quicktask, string.ascii_lowercase))
  172. >>> print step.next()
  173. returning a
  174. returning b
  175. returning c
  176. a
  177. >>> print step.next()
  178. b
  179. >>> print step.next()
  180. c
  181. >>> print step.next()
  182. returning d
  183. returning e
  184. returning f
  185. d
  186. >>> pausers[0].send("A")
  187. >>> print step.next()
  188. e
  189. >>> print step.next()
  190. f
  191. >>> print step.next()
  192. A woke up with A
  193. returning g
  194. returning h
  195. returning i
  196. g
  197. >>> print "".join([step.next() for x in xrange(3)])
  198. returning j
  199. returning k
  200. returning l
  201. returning m
  202. hij
  203. >>> pausers[1].send("B")
  204. >>> print "".join([step.next() for x in xrange(4)])
  205. B woke up with B
  206. returning n
  207. returning o
  208. returning p
  209. returning q
  210. klmn
  211. """
  212. # Get an iterator because of our funny nested loop below. Wrap the
  213. # iterable in enumerate() so we count items that come through.
  214. tuples = iter(enumerate(iterable))
  215. # If the iterable is empty, this whole function is a no-op, and we can
  216. # save ourselves some grief by just quitting out. In particular, once
  217. # we enter the outer loop below, we're going to wait on the queue --
  218. # but if we launched no coroutines with that queue as the destination,
  219. # we could end up waiting a very long time.
  220. try:
  221. index, args = tuples.next()
  222. except StopIteration:
  223. return
  224. # From this point forward, 'args' is the current arguments tuple and
  225. # 'index+1' counts how many such tuples we've seen.
  226. # This implementation relies on the fact that _execute() accepts an
  227. # event-like object, and -- unless it's None -- the completed
  228. # coroutine calls send(result). We slyly pass a queue rather than an
  229. # event -- the same queue instance for all coroutines. This is why our
  230. # queue interface intentionally resembles the event interface.
  231. q = coros.queue(max_size=qsize)
  232. # How many results have we yielded so far?
  233. finished = 0
  234. # This first loop is only until we've launched all the coroutines. Its
  235. # complexity is because if iterable contains more args tuples than the
  236. # size of our pool, attempting to _execute() the (poolsize+1)th
  237. # coroutine would suspend until something completes and send()s its
  238. # result to our queue. But to keep down queue overhead and to maximize
  239. # responsiveness to our caller, we'd rather suspend on reading the
  240. # queue. So we stuff the pool as full as we can, then wait for
  241. # something to finish, then stuff more coroutines into the pool.
  242. try:
  243. while True:
  244. # Before each yield, start as many new coroutines as we can fit.
  245. # (The self.free() test isn't 100% accurate: if we happen to be
  246. # executing in one of the pool's coroutines, we could _execute()
  247. # without waiting even if self.free() reports 0. See _execute().)
  248. # The point is that we don't want to wait in the _execute() call,
  249. # we want to wait in the q.wait() call.
  250. # IMPORTANT: at start, and whenever we've caught up with all
  251. # coroutines we've launched so far, we MUST iterate this inner
  252. # loop at least once, regardless of self.free() -- otherwise the
  253. # q.wait() call below will deadlock!
  254. # Recall that index is the index of the NEXT args tuple that we
  255. # haven't yet launched. Therefore it counts how many args tuples
  256. # we've launched so far.
  257. while self.free() > 0 or finished == index:
  258. # Just like the implementation of execute_async(), save that
  259. # we're passing our queue instead of None as the "event" to
  260. # which to send() the result.
  261. self._execute(q, function, args, {})
  262. # We've consumed that args tuple, advance to next.
  263. index, args = tuples.next()
  264. # Okay, we've filled up the pool again, yield a result -- which
  265. # will probably wait for a coroutine to complete. Although we do
  266. # have q.ready(), so we could iterate without waiting, we avoid
  267. # that because every yield could involve considerable real time.
  268. # We don't know how long it takes to return from yield, so every
  269. # time we do, take the opportunity to stuff more requests into the
  270. # pool before yielding again.
  271. yield q.wait()
  272. # Be sure to count results so we know when to stop!
  273. finished += 1
  274. except StopIteration:
  275. pass
  276. # Here we've exhausted the input iterable. index+1 is the total number
  277. # of coroutines we've launched. We probably haven't yielded that many
  278. # results yet. Wait for the rest of the results, yielding them as they
  279. # arrive.
  280. while finished < index + 1:
  281. yield q.wait()
  282. finished += 1