1# 2# Module providing the `Pool` class for managing a process pool 3# 4# multiprocessing/pool.py 5# 6# Copyright (c) 2006-2008, R Oudkerk 7# Licensed to PSF under a Contributor Agreement. 8# 9 10__all__ = ['Pool', 'ThreadPool'] 11 12# 13# Imports 14# 15 16import collections 17import itertools 18import os 19import queue 20import threading 21import time 22import traceback 23import warnings 24from queue import Empty 25 26# If threading is available then ThreadPool should be provided. Therefore 27# we avoid top-level imports which are liable to fail on some systems. 28from . import util 29from . import get_context, TimeoutError 30from .connection import wait 31 32# 33# Constants representing the state of a pool 34# 35 36INIT = "INIT" 37RUN = "RUN" 38CLOSE = "CLOSE" 39TERMINATE = "TERMINATE" 40 41# 42# Miscellaneous 43# 44 45job_counter = itertools.count() 46 47def mapstar(args): 48 return list(map(*args)) 49 50def starmapstar(args): 51 return list(itertools.starmap(args[0], args[1])) 52 53# 54# Hack to embed stringification of remote traceback in local traceback 55# 56 57class RemoteTraceback(Exception): 58 def __init__(self, tb): 59 self.tb = tb 60 def __str__(self): 61 return self.tb 62 63class ExceptionWithTraceback: 64 def __init__(self, exc, tb): 65 tb = traceback.format_exception(type(exc), exc, tb) 66 tb = ''.join(tb) 67 self.exc = exc 68 self.tb = '\n"""\n%s"""' % tb 69 def __reduce__(self): 70 return rebuild_exc, (self.exc, self.tb) 71 72def rebuild_exc(exc, tb): 73 exc.__cause__ = RemoteTraceback(tb) 74 return exc 75 76# 77# Code run by worker processes 78# 79 80class MaybeEncodingError(Exception): 81 """Wraps possible unpickleable errors, so they can be 82 safely sent through the socket.""" 83 84 def __init__(self, exc, value): 85 self.exc = repr(exc) 86 self.value = repr(value) 87 super(MaybeEncodingError, self).__init__(self.exc, self.value) 88 89 def __str__(self): 90 return "Error sending result: '%s'. Reason: '%s'" % (self.value, 91 self.exc) 92 93 def __repr__(self): 94 return "<%s: %s>" % (self.__class__.__name__, self) 95 96 97def worker(inqueue, outqueue, initializer=None, initargs=(), maxtasks=None, 98 wrap_exception=False): 99 if (maxtasks is not None) and not (isinstance(maxtasks, int) 100 and maxtasks >= 1): 101 raise AssertionError("Maxtasks {!r} is not valid".format(maxtasks)) 102 put = outqueue.put 103 get = inqueue.get 104 if hasattr(inqueue, '_writer'): 105 inqueue._writer.close() 106 outqueue._reader.close() 107 108 if initializer is not None: 109 initializer(*initargs) 110 111 completed = 0 112 while maxtasks is None or (maxtasks and completed < maxtasks): 113 try: 114 task = get() 115 except (EOFError, OSError): 116 util.debug('worker got EOFError or OSError -- exiting') 117 break 118 119 if task is None: 120 util.debug('worker got sentinel -- exiting') 121 break 122 123 job, i, func, args, kwds = task 124 try: 125 result = (True, func(*args, **kwds)) 126 except Exception as e: 127 if wrap_exception and func is not _helper_reraises_exception: 128 e = ExceptionWithTraceback(e, e.__traceback__) 129 result = (False, e) 130 try: 131 put((job, i, result)) 132 except Exception as e: 133 wrapped = MaybeEncodingError(e, result[1]) 134 util.debug("Possible encoding error while sending result: %s" % ( 135 wrapped)) 136 put((job, i, (False, wrapped))) 137 138 task = job = result = func = args = kwds = None 139 completed += 1 140 util.debug('worker exiting after %d tasks' % completed) 141 142def _helper_reraises_exception(ex): 143 'Pickle-able helper function for use by _guarded_task_generation.' 144 raise ex 145 146# 147# Class representing a process pool 148# 149 150class _PoolCache(dict): 151 """ 152 Class that implements a cache for the Pool class that will notify 153 the pool management threads every time the cache is emptied. The 154 notification is done by the use of a queue that is provided when 155 instantiating the cache. 156 """ 157 def __init__(self, /, *args, notifier=None, **kwds): 158 self.notifier = notifier 159 super().__init__(*args, **kwds) 160 161 def __delitem__(self, item): 162 super().__delitem__(item) 163 164 # Notify that the cache is empty. This is important because the 165 # pool keeps maintaining workers until the cache gets drained. This 166 # eliminates a race condition in which a task is finished after the 167 # the pool's _handle_workers method has enter another iteration of the 168 # loop. In this situation, the only event that can wake up the pool 169 # is the cache to be emptied (no more tasks available). 170 if not self: 171 self.notifier.put(None) 172 173class Pool(object): 174 ''' 175 Class which supports an async version of applying functions to arguments. 176 ''' 177 _wrap_exception = True 178 179 @staticmethod 180 def Process(ctx, *args, **kwds): 181 return ctx.Process(*args, **kwds) 182 183 def __init__(self, processes=None, initializer=None, initargs=(), 184 maxtasksperchild=None, context=None): 185 # Attributes initialized early to make sure that they exist in 186 # __del__() if __init__() raises an exception 187 self._pool = [] 188 self._state = INIT 189 190 self._ctx = context or get_context() 191 self._setup_queues() 192 self._taskqueue = queue.SimpleQueue() 193 # The _change_notifier queue exist to wake up self._handle_workers() 194 # when the cache (self._cache) is empty or when there is a change in 195 # the _state variable of the thread that runs _handle_workers. 196 self._change_notifier = self._ctx.SimpleQueue() 197 self._cache = _PoolCache(notifier=self._change_notifier) 198 self._maxtasksperchild = maxtasksperchild 199 self._initializer = initializer 200 self._initargs = initargs 201 202 if processes is None: 203 processes = os.cpu_count() or 1 204 if processes < 1: 205 raise ValueError("Number of processes must be at least 1") 206 207 if initializer is not None and not callable(initializer): 208 raise TypeError('initializer must be a callable') 209 210 self._processes = processes 211 try: 212 self._repopulate_pool() 213 except Exception: 214 for p in self._pool: 215 if p.exitcode is None: 216 p.terminate() 217 for p in self._pool: 218 p.join() 219 raise 220 221 sentinels = self._get_sentinels() 222 223 self._worker_handler = threading.Thread( 224 target=Pool._handle_workers, 225 args=(self._cache, self._taskqueue, self._ctx, self.Process, 226 self._processes, self._pool, self._inqueue, self._outqueue, 227 self._initializer, self._initargs, self._maxtasksperchild, 228 self._wrap_exception, sentinels, self._change_notifier) 229 ) 230 self._worker_handler.daemon = True 231 self._worker_handler._state = RUN 232 self._worker_handler.start() 233 234 235 self._task_handler = threading.Thread( 236 target=Pool._handle_tasks, 237 args=(self._taskqueue, self._quick_put, self._outqueue, 238 self._pool, self._cache) 239 ) 240 self._task_handler.daemon = True 241 self._task_handler._state = RUN 242 self._task_handler.start() 243 244 self._result_handler = threading.Thread( 245 target=Pool._handle_results, 246 args=(self._outqueue, self._quick_get, self._cache) 247 ) 248 self._result_handler.daemon = True 249 self._result_handler._state = RUN 250 self._result_handler.start() 251 252 self._terminate = util.Finalize( 253 self, self._terminate_pool, 254 args=(self._taskqueue, self._inqueue, self._outqueue, self._pool, 255 self._change_notifier, self._worker_handler, self._task_handler, 256 self._result_handler, self._cache), 257 exitpriority=15 258 ) 259 self._state = RUN 260 261 # Copy globals as function locals to make sure that they are available 262 # during Python shutdown when the Pool is destroyed. 263 def __del__(self, _warn=warnings.warn, RUN=RUN): 264 if self._state == RUN: 265 _warn(f"unclosed running multiprocessing pool {self!r}", 266 ResourceWarning, source=self) 267 if getattr(self, '_change_notifier', None) is not None: 268 self._change_notifier.put(None) 269 270 def __repr__(self): 271 cls = self.__class__ 272 return (f'<{cls.__module__}.{cls.__qualname__} ' 273 f'state={self._state} ' 274 f'pool_size={len(self._pool)}>') 275 276 def _get_sentinels(self): 277 task_queue_sentinels = [self._outqueue._reader] 278 self_notifier_sentinels = [self._change_notifier._reader] 279 return [*task_queue_sentinels, *self_notifier_sentinels] 280 281 @staticmethod 282 def _get_worker_sentinels(workers): 283 return [worker.sentinel for worker in 284 workers if hasattr(worker, "sentinel")] 285 286 @staticmethod 287 def _join_exited_workers(pool): 288 """Cleanup after any worker processes which have exited due to reaching 289 their specified lifetime. Returns True if any workers were cleaned up. 290 """ 291 cleaned = False 292 for i in reversed(range(len(pool))): 293 worker = pool[i] 294 if worker.exitcode is not None: 295 # worker exited 296 util.debug('cleaning up worker %d' % i) 297 worker.join() 298 cleaned = True 299 del pool[i] 300 return cleaned 301 302 def _repopulate_pool(self): 303 return self._repopulate_pool_static(self._ctx, self.Process, 304 self._processes, 305 self._pool, self._inqueue, 306 self._outqueue, self._initializer, 307 self._initargs, 308 self._maxtasksperchild, 309 self._wrap_exception) 310 311 @staticmethod 312 def _repopulate_pool_static(ctx, Process, processes, pool, inqueue, 313 outqueue, initializer, initargs, 314 maxtasksperchild, wrap_exception): 315 """Bring the number of pool processes up to the specified number, 316 for use after reaping workers which have exited. 317 """ 318 for i in range(processes - len(pool)): 319 w = Process(ctx, target=worker, 320 args=(inqueue, outqueue, 321 initializer, 322 initargs, maxtasksperchild, 323 wrap_exception)) 324 w.name = w.name.replace('Process', 'PoolWorker') 325 w.daemon = True 326 w.start() 327 pool.append(w) 328 util.debug('added worker') 329 330 @staticmethod 331 def _maintain_pool(ctx, Process, processes, pool, inqueue, outqueue, 332 initializer, initargs, maxtasksperchild, 333 wrap_exception): 334 """Clean up any exited workers and start replacements for them. 335 """ 336 if Pool._join_exited_workers(pool): 337 Pool._repopulate_pool_static(ctx, Process, processes, pool, 338 inqueue, outqueue, initializer, 339 initargs, maxtasksperchild, 340 wrap_exception) 341 342 def _setup_queues(self): 343 self._inqueue = self._ctx.SimpleQueue() 344 self._outqueue = self._ctx.SimpleQueue() 345 self._quick_put = self._inqueue._writer.send 346 self._quick_get = self._outqueue._reader.recv 347 348 def _check_running(self): 349 if self._state != RUN: 350 raise ValueError("Pool not running") 351 352 def apply(self, func, args=(), kwds={}): 353 ''' 354 Equivalent of `func(*args, **kwds)`. 355 Pool must be running. 356 ''' 357 return self.apply_async(func, args, kwds).get() 358 359 def map(self, func, iterable, chunksize=None): 360 ''' 361 Apply `func` to each element in `iterable`, collecting the results 362 in a list that is returned. 363 ''' 364 return self._map_async(func, iterable, mapstar, chunksize).get() 365 366 def starmap(self, func, iterable, chunksize=None): 367 ''' 368 Like `map()` method but the elements of the `iterable` are expected to 369 be iterables as well and will be unpacked as arguments. Hence 370 `func` and (a, b) becomes func(a, b). 371 ''' 372 return self._map_async(func, iterable, starmapstar, chunksize).get() 373 374 def starmap_async(self, func, iterable, chunksize=None, callback=None, 375 error_callback=None): 376 ''' 377 Asynchronous version of `starmap()` method. 378 ''' 379 return self._map_async(func, iterable, starmapstar, chunksize, 380 callback, error_callback) 381 382 def _guarded_task_generation(self, result_job, func, iterable): 383 '''Provides a generator of tasks for imap and imap_unordered with 384 appropriate handling for iterables which throw exceptions during 385 iteration.''' 386 try: 387 i = -1 388 for i, x in enumerate(iterable): 389 yield (result_job, i, func, (x,), {}) 390 except Exception as e: 391 yield (result_job, i+1, _helper_reraises_exception, (e,), {}) 392 393 def imap(self, func, iterable, chunksize=1): 394 ''' 395 Equivalent of `map()` -- can be MUCH slower than `Pool.map()`. 396 ''' 397 self._check_running() 398 if chunksize == 1: 399 result = IMapIterator(self) 400 self._taskqueue.put( 401 ( 402 self._guarded_task_generation(result._job, func, iterable), 403 result._set_length 404 )) 405 return result 406 else: 407 if chunksize < 1: 408 raise ValueError( 409 "Chunksize must be 1+, not {0:n}".format( 410 chunksize)) 411 task_batches = Pool._get_tasks(func, iterable, chunksize) 412 result = IMapIterator(self) 413 self._taskqueue.put( 414 ( 415 self._guarded_task_generation(result._job, 416 mapstar, 417 task_batches), 418 result._set_length 419 )) 420 return (item for chunk in result for item in chunk) 421 422 def imap_unordered(self, func, iterable, chunksize=1): 423 ''' 424 Like `imap()` method but ordering of results is arbitrary. 425 ''' 426 self._check_running() 427 if chunksize == 1: 428 result = IMapUnorderedIterator(self) 429 self._taskqueue.put( 430 ( 431 self._guarded_task_generation(result._job, func, iterable), 432 result._set_length 433 )) 434 return result 435 else: 436 if chunksize < 1: 437 raise ValueError( 438 "Chunksize must be 1+, not {0!r}".format(chunksize)) 439 task_batches = Pool._get_tasks(func, iterable, chunksize) 440 result = IMapUnorderedIterator(self) 441 self._taskqueue.put( 442 ( 443 self._guarded_task_generation(result._job, 444 mapstar, 445 task_batches), 446 result._set_length 447 )) 448 return (item for chunk in result for item in chunk) 449 450 def apply_async(self, func, args=(), kwds={}, callback=None, 451 error_callback=None): 452 ''' 453 Asynchronous version of `apply()` method. 454 ''' 455 self._check_running() 456 result = ApplyResult(self, callback, error_callback) 457 self._taskqueue.put(([(result._job, 0, func, args, kwds)], None)) 458 return result 459 460 def map_async(self, func, iterable, chunksize=None, callback=None, 461 error_callback=None): 462 ''' 463 Asynchronous version of `map()` method. 464 ''' 465 return self._map_async(func, iterable, mapstar, chunksize, callback, 466 error_callback) 467 468 def _map_async(self, func, iterable, mapper, chunksize=None, callback=None, 469 error_callback=None): 470 ''' 471 Helper function to implement map, starmap and their async counterparts. 472 ''' 473 self._check_running() 474 if not hasattr(iterable, '__len__'): 475 iterable = list(iterable) 476 477 if chunksize is None: 478 chunksize, extra = divmod(len(iterable), len(self._pool) * 4) 479 if extra: 480 chunksize += 1 481 if len(iterable) == 0: 482 chunksize = 0 483 484 task_batches = Pool._get_tasks(func, iterable, chunksize) 485 result = MapResult(self, chunksize, len(iterable), callback, 486 error_callback=error_callback) 487 self._taskqueue.put( 488 ( 489 self._guarded_task_generation(result._job, 490 mapper, 491 task_batches), 492 None 493 ) 494 ) 495 return result 496 497 @staticmethod 498 def _wait_for_updates(sentinels, change_notifier, timeout=None): 499 wait(sentinels, timeout=timeout) 500 while not change_notifier.empty(): 501 change_notifier.get() 502 503 @classmethod 504 def _handle_workers(cls, cache, taskqueue, ctx, Process, processes, 505 pool, inqueue, outqueue, initializer, initargs, 506 maxtasksperchild, wrap_exception, sentinels, 507 change_notifier): 508 thread = threading.current_thread() 509 510 # Keep maintaining workers until the cache gets drained, unless the pool 511 # is terminated. 512 while thread._state == RUN or (cache and thread._state != TERMINATE): 513 cls._maintain_pool(ctx, Process, processes, pool, inqueue, 514 outqueue, initializer, initargs, 515 maxtasksperchild, wrap_exception) 516 517 current_sentinels = [*cls._get_worker_sentinels(pool), *sentinels] 518 519 cls._wait_for_updates(current_sentinels, change_notifier) 520 # send sentinel to stop workers 521 taskqueue.put(None) 522 util.debug('worker handler exiting') 523 524 @staticmethod 525 def _handle_tasks(taskqueue, put, outqueue, pool, cache): 526 thread = threading.current_thread() 527 528 for taskseq, set_length in iter(taskqueue.get, None): 529 task = None 530 try: 531 # iterating taskseq cannot fail 532 for task in taskseq: 533 if thread._state != RUN: 534 util.debug('task handler found thread._state != RUN') 535 break 536 try: 537 put(task) 538 except Exception as e: 539 job, idx = task[:2] 540 try: 541 cache[job]._set(idx, (False, e)) 542 except KeyError: 543 pass 544 else: 545 if set_length: 546 util.debug('doing set_length()') 547 idx = task[1] if task else -1 548 set_length(idx + 1) 549 continue 550 break 551 finally: 552 task = taskseq = job = None 553 else: 554 util.debug('task handler got sentinel') 555 556 try: 557 # tell result handler to finish when cache is empty 558 util.debug('task handler sending sentinel to result handler') 559 outqueue.put(None) 560 561 # tell workers there is no more work 562 util.debug('task handler sending sentinel to workers') 563 for p in pool: 564 put(None) 565 except OSError: 566 util.debug('task handler got OSError when sending sentinels') 567 568 util.debug('task handler exiting') 569 570 @staticmethod 571 def _handle_results(outqueue, get, cache): 572 thread = threading.current_thread() 573 574 while 1: 575 try: 576 task = get() 577 except (OSError, EOFError): 578 util.debug('result handler got EOFError/OSError -- exiting') 579 return 580 581 if thread._state != RUN: 582 assert thread._state == TERMINATE, "Thread not in TERMINATE" 583 util.debug('result handler found thread._state=TERMINATE') 584 break 585 586 if task is None: 587 util.debug('result handler got sentinel') 588 break 589 590 job, i, obj = task 591 try: 592 cache[job]._set(i, obj) 593 except KeyError: 594 pass 595 task = job = obj = None 596 597 while cache and thread._state != TERMINATE: 598 try: 599 task = get() 600 except (OSError, EOFError): 601 util.debug('result handler got EOFError/OSError -- exiting') 602 return 603 604 if task is None: 605 util.debug('result handler ignoring extra sentinel') 606 continue 607 job, i, obj = task 608 try: 609 cache[job]._set(i, obj) 610 except KeyError: 611 pass 612 task = job = obj = None 613 614 if hasattr(outqueue, '_reader'): 615 util.debug('ensuring that outqueue is not full') 616 # If we don't make room available in outqueue then 617 # attempts to add the sentinel (None) to outqueue may 618 # block. There is guaranteed to be no more than 2 sentinels. 619 try: 620 for i in range(10): 621 if not outqueue._reader.poll(): 622 break 623 get() 624 except (OSError, EOFError): 625 pass 626 627 util.debug('result handler exiting: len(cache)=%s, thread._state=%s', 628 len(cache), thread._state) 629 630 @staticmethod 631 def _get_tasks(func, it, size): 632 it = iter(it) 633 while 1: 634 x = tuple(itertools.islice(it, size)) 635 if not x: 636 return 637 yield (func, x) 638 639 def __reduce__(self): 640 raise NotImplementedError( 641 'pool objects cannot be passed between processes or pickled' 642 ) 643 644 def close(self): 645 util.debug('closing pool') 646 if self._state == RUN: 647 self._state = CLOSE 648 self._worker_handler._state = CLOSE 649 self._change_notifier.put(None) 650 651 def terminate(self): 652 util.debug('terminating pool') 653 self._state = TERMINATE 654 self._worker_handler._state = TERMINATE 655 self._change_notifier.put(None) 656 self._terminate() 657 658 def join(self): 659 util.debug('joining pool') 660 if self._state == RUN: 661 raise ValueError("Pool is still running") 662 elif self._state not in (CLOSE, TERMINATE): 663 raise ValueError("In unknown state") 664 self._worker_handler.join() 665 self._task_handler.join() 666 self._result_handler.join() 667 for p in self._pool: 668 p.join() 669 670 @staticmethod 671 def _help_stuff_finish(inqueue, task_handler, size): 672 # task_handler may be blocked trying to put items on inqueue 673 util.debug('removing tasks from inqueue until task handler finished') 674 inqueue._rlock.acquire() 675 while task_handler.is_alive() and inqueue._reader.poll(): 676 inqueue._reader.recv() 677 time.sleep(0) 678 679 @classmethod 680 def _terminate_pool(cls, taskqueue, inqueue, outqueue, pool, change_notifier, 681 worker_handler, task_handler, result_handler, cache): 682 # this is guaranteed to only be called once 683 util.debug('finalizing pool') 684 685 worker_handler._state = TERMINATE 686 task_handler._state = TERMINATE 687 688 util.debug('helping task handler/workers to finish') 689 cls._help_stuff_finish(inqueue, task_handler, len(pool)) 690 691 if (not result_handler.is_alive()) and (len(cache) != 0): 692 raise AssertionError( 693 "Cannot have cache with result_hander not alive") 694 695 result_handler._state = TERMINATE 696 change_notifier.put(None) 697 outqueue.put(None) # sentinel 698 699 # We must wait for the worker handler to exit before terminating 700 # workers because we don't want workers to be restarted behind our back. 701 util.debug('joining worker handler') 702 if threading.current_thread() is not worker_handler: 703 worker_handler.join() 704 705 # Terminate workers which haven't already finished. 706 if pool and hasattr(pool[0], 'terminate'): 707 util.debug('terminating workers') 708 for p in pool: 709 if p.exitcode is None: 710 p.terminate() 711 712 util.debug('joining task handler') 713 if threading.current_thread() is not task_handler: 714 task_handler.join() 715 716 util.debug('joining result handler') 717 if threading.current_thread() is not result_handler: 718 result_handler.join() 719 720 if pool and hasattr(pool[0], 'terminate'): 721 util.debug('joining pool workers') 722 for p in pool: 723 if p.is_alive(): 724 # worker has not yet exited 725 util.debug('cleaning up worker %d' % p.pid) 726 p.join() 727 728 def __enter__(self): 729 self._check_running() 730 return self 731 732 def __exit__(self, exc_type, exc_val, exc_tb): 733 self.terminate() 734 735# 736# Class whose instances are returned by `Pool.apply_async()` 737# 738 739class ApplyResult(object): 740 741 def __init__(self, pool, callback, error_callback): 742 self._pool = pool 743 self._event = threading.Event() 744 self._job = next(job_counter) 745 self._cache = pool._cache 746 self._callback = callback 747 self._error_callback = error_callback 748 self._cache[self._job] = self 749 750 def ready(self): 751 return self._event.is_set() 752 753 def successful(self): 754 if not self.ready(): 755 raise ValueError("{0!r} not ready".format(self)) 756 return self._success 757 758 def wait(self, timeout=None): 759 self._event.wait(timeout) 760 761 def get(self, timeout=None): 762 self.wait(timeout) 763 if not self.ready(): 764 raise TimeoutError 765 if self._success: 766 return self._value 767 else: 768 raise self._value 769 770 def _set(self, i, obj): 771 self._success, self._value = obj 772 if self._callback and self._success: 773 self._callback(self._value) 774 if self._error_callback and not self._success: 775 self._error_callback(self._value) 776 self._event.set() 777 del self._cache[self._job] 778 self._pool = None 779 780AsyncResult = ApplyResult # create alias -- see #17805 781 782# 783# Class whose instances are returned by `Pool.map_async()` 784# 785 786class MapResult(ApplyResult): 787 788 def __init__(self, pool, chunksize, length, callback, error_callback): 789 ApplyResult.__init__(self, pool, callback, 790 error_callback=error_callback) 791 self._success = True 792 self._value = [None] * length 793 self._chunksize = chunksize 794 if chunksize <= 0: 795 self._number_left = 0 796 self._event.set() 797 del self._cache[self._job] 798 else: 799 self._number_left = length//chunksize + bool(length % chunksize) 800 801 def _set(self, i, success_result): 802 self._number_left -= 1 803 success, result = success_result 804 if success and self._success: 805 self._value[i*self._chunksize:(i+1)*self._chunksize] = result 806 if self._number_left == 0: 807 if self._callback: 808 self._callback(self._value) 809 del self._cache[self._job] 810 self._event.set() 811 self._pool = None 812 else: 813 if not success and self._success: 814 # only store first exception 815 self._success = False 816 self._value = result 817 if self._number_left == 0: 818 # only consider the result ready once all jobs are done 819 if self._error_callback: 820 self._error_callback(self._value) 821 del self._cache[self._job] 822 self._event.set() 823 self._pool = None 824 825# 826# Class whose instances are returned by `Pool.imap()` 827# 828 829class IMapIterator(object): 830 831 def __init__(self, pool): 832 self._pool = pool 833 self._cond = threading.Condition(threading.Lock()) 834 self._job = next(job_counter) 835 self._cache = pool._cache 836 self._items = collections.deque() 837 self._index = 0 838 self._length = None 839 self._unsorted = {} 840 self._cache[self._job] = self 841 842 def __iter__(self): 843 return self 844 845 def next(self, timeout=None): 846 with self._cond: 847 try: 848 item = self._items.popleft() 849 except IndexError: 850 if self._index == self._length: 851 self._pool = None 852 raise StopIteration from None 853 self._cond.wait(timeout) 854 try: 855 item = self._items.popleft() 856 except IndexError: 857 if self._index == self._length: 858 self._pool = None 859 raise StopIteration from None 860 raise TimeoutError from None 861 862 success, value = item 863 if success: 864 return value 865 raise value 866 867 __next__ = next # XXX 868 869 def _set(self, i, obj): 870 with self._cond: 871 if self._index == i: 872 self._items.append(obj) 873 self._index += 1 874 while self._index in self._unsorted: 875 obj = self._unsorted.pop(self._index) 876 self._items.append(obj) 877 self._index += 1 878 self._cond.notify() 879 else: 880 self._unsorted[i] = obj 881 882 if self._index == self._length: 883 del self._cache[self._job] 884 self._pool = None 885 886 def _set_length(self, length): 887 with self._cond: 888 self._length = length 889 if self._index == self._length: 890 self._cond.notify() 891 del self._cache[self._job] 892 self._pool = None 893 894# 895# Class whose instances are returned by `Pool.imap_unordered()` 896# 897 898class IMapUnorderedIterator(IMapIterator): 899 900 def _set(self, i, obj): 901 with self._cond: 902 self._items.append(obj) 903 self._index += 1 904 self._cond.notify() 905 if self._index == self._length: 906 del self._cache[self._job] 907 self._pool = None 908 909# 910# 911# 912 913class ThreadPool(Pool): 914 _wrap_exception = False 915 916 @staticmethod 917 def Process(ctx, *args, **kwds): 918 from .dummy import Process 919 return Process(*args, **kwds) 920 921 def __init__(self, processes=None, initializer=None, initargs=()): 922 Pool.__init__(self, processes, initializer, initargs) 923 924 def _setup_queues(self): 925 self._inqueue = queue.SimpleQueue() 926 self._outqueue = queue.SimpleQueue() 927 self._quick_put = self._inqueue.put 928 self._quick_get = self._outqueue.get 929 930 def _get_sentinels(self): 931 return [self._change_notifier._reader] 932 933 @staticmethod 934 def _get_worker_sentinels(workers): 935 return [] 936 937 @staticmethod 938 def _help_stuff_finish(inqueue, task_handler, size): 939 # drain inqueue, and put sentinels at its head to make workers finish 940 try: 941 while True: 942 inqueue.get(block=False) 943 except queue.Empty: 944 pass 945 for i in range(size): 946 inqueue.put(None) 947 948 def _wait_for_updates(self, sentinels, change_notifier, timeout): 949 time.sleep(timeout) 950