cpython-lru_cache-weakref: Lib/multiprocessing/pool.py annotate

annotate Lib/multiprocessing/pool.py @ 87690:66c1c9f32567 default tip

Add use_weakref option to functools.lru_cache The use_weakref option makes lru_cache create weak references to the function arguments when creating the cache key, which prevents them from staying in the memory forever even after all other references to them are gone. This is a proof of concept for http://bugs.python.org/issue19859

author	Radomir Dopieralski <devel@sheep.art.pl>
date	Tue, 03 Dec 2013 16:02:48 +0100
parents	8c25b4515b3d
children

rev	line source
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	1 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	2 # Module providing the `Pool` class for managing a process pool
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	3 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	4 # multiprocessing/pool.py
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	5 #
66886 71fb03274a89 Merged revisions 87225 via svnmerge from R. David Murray <rdmurray@bitdance.com> parents: 60063 diff changeset	6 # Copyright (c) 2006-2008, R Oudkerk
76663 25cc0c8fd4e2 Mark multiprocessing files with "Licensed to PSF under a Contributor Agreement" Richard Oudkerk <shibturn@gmail.com> parents: 74107 diff changeset	7 # Licensed to PSF under a Contributor Agreement.
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	8 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	9
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	10 __all__ = ['Pool', 'ThreadPool']
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	11
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	12 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	13 # Imports
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	14 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	15
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	16 import threading
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	17 import queue
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	18 import itertools
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	19 import collections
84355 6a0437adafbd Issue #17914: Use os.cpu_count() instead of multiprocessing.cpu_count() where Charles-François Natali <cf.natali@gmail.com> parents: 83641 diff changeset	20 import os
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	21 import time
83638 c4f92b597074 Issue #13813: Embed stringification of remote traceback in local Richard Oudkerk <shibturn@gmail.com> parents: 81039 diff changeset	22 import traceback
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	23
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	24 # If threading is available then ThreadPool should be provided. Therefore
3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	25 # we avoid top-level imports which are liable to fail on some systems.
3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	26 from . import util
86387 72a5ac909c7a Issue #18999: Make multiprocessing use context objects. Richard Oudkerk <shibturn@gmail.com> parents: 85169 diff changeset	27 from . import get_context, cpu_count, TimeoutError
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	28
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	29 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	30 # Constants representing the state of a pool
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	31 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	32
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	33 RUN = 0
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	34 CLOSE = 1
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	35 TERMINATE = 2
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	36
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	37 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	38 # Miscellaneous
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	39 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	40
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	41 job_counter = itertools.count()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	42
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	43 def mapstar(args):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	44 return list(map(*args))
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	45
74107 b07b1e58582d Issue #12708: Add starmap() and starmap_async() methods (similar to itertools.starmap()) to multiprocessing.Pool. Antoine Pitrou <solipsis@pitrou.net> parents: 73165 diff changeset	46 def starmapstar(args):
b07b1e58582d Issue #12708: Add starmap() and starmap_async() methods (similar to itertools.starmap()) to multiprocessing.Pool. Antoine Pitrou <solipsis@pitrou.net> parents: 73165 diff changeset	47 return list(itertools.starmap(args[0], args[1]))
b07b1e58582d Issue #12708: Add starmap() and starmap_async() methods (similar to itertools.starmap()) to multiprocessing.Pool. Antoine Pitrou <solipsis@pitrou.net> parents: 73165 diff changeset	48
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	49 #
83638 c4f92b597074 Issue #13813: Embed stringification of remote traceback in local Richard Oudkerk <shibturn@gmail.com> parents: 81039 diff changeset	50 # Hack to embed stringification of remote traceback in local traceback
c4f92b597074 Issue #13813: Embed stringification of remote traceback in local Richard Oudkerk <shibturn@gmail.com> parents: 81039 diff changeset	51 #
c4f92b597074 Issue #13813: Embed stringification of remote traceback in local Richard Oudkerk <shibturn@gmail.com> parents: 81039 diff changeset	52
c4f92b597074 Issue #13813: Embed stringification of remote traceback in local Richard Oudkerk <shibturn@gmail.com> parents: 81039 diff changeset	53 class RemoteTraceback(Exception):
c4f92b597074 Issue #13813: Embed stringification of remote traceback in local Richard Oudkerk <shibturn@gmail.com> parents: 81039 diff changeset	54 def __init__(self, tb):
c4f92b597074 Issue #13813: Embed stringification of remote traceback in local Richard Oudkerk <shibturn@gmail.com> parents: 81039 diff changeset	55 self.tb = tb
c4f92b597074 Issue #13813: Embed stringification of remote traceback in local Richard Oudkerk <shibturn@gmail.com> parents: 81039 diff changeset	56 def __str__(self):
c4f92b597074 Issue #13813: Embed stringification of remote traceback in local Richard Oudkerk <shibturn@gmail.com> parents: 81039 diff changeset	57 return self.tb
c4f92b597074 Issue #13813: Embed stringification of remote traceback in local Richard Oudkerk <shibturn@gmail.com> parents: 81039 diff changeset	58
c4f92b597074 Issue #13813: Embed stringification of remote traceback in local Richard Oudkerk <shibturn@gmail.com> parents: 81039 diff changeset	59 class ExceptionWithTraceback:
c4f92b597074 Issue #13813: Embed stringification of remote traceback in local Richard Oudkerk <shibturn@gmail.com> parents: 81039 diff changeset	60 def __init__(self, exc, tb):
c4f92b597074 Issue #13813: Embed stringification of remote traceback in local Richard Oudkerk <shibturn@gmail.com> parents: 81039 diff changeset	61 tb = traceback.format_exception(type(exc), exc, tb)
c4f92b597074 Issue #13813: Embed stringification of remote traceback in local Richard Oudkerk <shibturn@gmail.com> parents: 81039 diff changeset	62 tb = ''.join(tb)
c4f92b597074 Issue #13813: Embed stringification of remote traceback in local Richard Oudkerk <shibturn@gmail.com> parents: 81039 diff changeset	63 self.exc = exc
c4f92b597074 Issue #13813: Embed stringification of remote traceback in local Richard Oudkerk <shibturn@gmail.com> parents: 81039 diff changeset	64 self.tb = '\n"""\n%s"""' % tb
c4f92b597074 Issue #13813: Embed stringification of remote traceback in local Richard Oudkerk <shibturn@gmail.com> parents: 81039 diff changeset	65 def __reduce__(self):
c4f92b597074 Issue #13813: Embed stringification of remote traceback in local Richard Oudkerk <shibturn@gmail.com> parents: 81039 diff changeset	66 return rebuild_exc, (self.exc, self.tb)
c4f92b597074 Issue #13813: Embed stringification of remote traceback in local Richard Oudkerk <shibturn@gmail.com> parents: 81039 diff changeset	67
c4f92b597074 Issue #13813: Embed stringification of remote traceback in local Richard Oudkerk <shibturn@gmail.com> parents: 81039 diff changeset	68 def rebuild_exc(exc, tb):
c4f92b597074 Issue #13813: Embed stringification of remote traceback in local Richard Oudkerk <shibturn@gmail.com> parents: 81039 diff changeset	69 exc.__cause__ = RemoteTraceback(tb)
c4f92b597074 Issue #13813: Embed stringification of remote traceback in local Richard Oudkerk <shibturn@gmail.com> parents: 81039 diff changeset	70 return exc
c4f92b597074 Issue #13813: Embed stringification of remote traceback in local Richard Oudkerk <shibturn@gmail.com> parents: 81039 diff changeset	71
c4f92b597074 Issue #13813: Embed stringification of remote traceback in local Richard Oudkerk <shibturn@gmail.com> parents: 81039 diff changeset	72 #
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	73 # Code run by worker processes
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	74 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	75
66115 0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	76 class MaybeEncodingError(Exception):
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	77 """Wraps possible unpickleable errors, so they can be
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	78 safely sent through the socket."""
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	79
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	80 def __init__(self, exc, value):
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	81 self.exc = repr(exc)
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	82 self.value = repr(value)
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	83 super(MaybeEncodingError, self).__init__(self.exc, self.value)
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	84
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	85 def __str__(self):
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	86 return "Error sending result: '%s'. Reason: '%s'" % (self.value,
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	87 self.exc)
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	88
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	89 def __repr__(self):
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	90 return "<MaybeEncodingError: %s>" % str(self)
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	91
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	92
58537 7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	93 def worker(inqueue, outqueue, initializer=None, initargs=(), maxtasks=None):
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	94 assert maxtasks is None or (type(maxtasks) == int and maxtasks > 0)
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	95 put = outqueue.put
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	96 get = inqueue.get
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	97 if hasattr(inqueue, '_writer'):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	98 inqueue._writer.close()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	99 outqueue._reader.close()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	100
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	101 if initializer is not None:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	102 initializer(*initargs)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	103
58537 7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	104 completed = 0
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	105 while maxtasks is None or (maxtasks and completed < maxtasks):
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	106 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	107 task = get()
81039 7d69d04522e3 Replace IOError with OSError (#16715) Andrew Svetlov <andrew.svetlov@gmail.com> parents: 79961 diff changeset	108 except (EOFError, OSError):
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	109 util.debug('worker got EOFError or OSError -- exiting')
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	110 break
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	111
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	112 if task is None:
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	113 util.debug('worker got sentinel -- exiting')
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	114 break
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	115
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	116 job, i, func, args, kwds = task
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	117 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	118 result = (True, func(args, *kwds))
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	119 except Exception as e:
83638 c4f92b597074 Issue #13813: Embed stringification of remote traceback in local Richard Oudkerk <shibturn@gmail.com> parents: 81039 diff changeset	120 e = ExceptionWithTraceback(e, e.__traceback__)
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	121 result = (False, e)
66115 0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	122 try:
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	123 put((job, i, result))
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	124 except Exception as e:
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	125 wrapped = MaybeEncodingError(e, result[1])
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	126 util.debug("Possible encoding error while sending result: %s" % (
66115 0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	127 wrapped))
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	128 put((job, i, (False, wrapped)))
58537 7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	129 completed += 1
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	130 util.debug('worker exiting after %d tasks' % completed)
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	131
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	132 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	133 # Class representing a process pool
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	134 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	135
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	136 class Pool(object):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	137 '''
49809 a4220c92f7cc #4206: fix 2.xisms in multiprocessing docs and docstrings. Georg Brandl <georg@python.org> parents: 48773 diff changeset	138 Class which supports an async version of applying functions to arguments.
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	139 '''
86387 72a5ac909c7a Issue #18999: Make multiprocessing use context objects. Richard Oudkerk <shibturn@gmail.com> parents: 85169 diff changeset	140 def Process(self, args, *kwds):
72a5ac909c7a Issue #18999: Make multiprocessing use context objects. Richard Oudkerk <shibturn@gmail.com> parents: 85169 diff changeset	141 return self._ctx.Process(args, *kwds)
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	142
58537 7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	143 def __init__(self, processes=None, initializer=None, initargs=(),
86387 72a5ac909c7a Issue #18999: Make multiprocessing use context objects. Richard Oudkerk <shibturn@gmail.com> parents: 85169 diff changeset	144 maxtasksperchild=None, context=None):
72a5ac909c7a Issue #18999: Make multiprocessing use context objects. Richard Oudkerk <shibturn@gmail.com> parents: 85169 diff changeset	145 self._ctx = context or get_context()
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	146 self._setup_queues()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	147 self._taskqueue = queue.Queue()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	148 self._cache = {}
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	149 self._state = RUN
58537 7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	150 self._maxtasksperchild = maxtasksperchild
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	151 self._initializer = initializer
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	152 self._initargs = initargs
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	153
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	154 if processes is None:
84355 6a0437adafbd Issue #17914: Use os.cpu_count() instead of multiprocessing.cpu_count() where Charles-François Natali <cf.natali@gmail.com> parents: 83641 diff changeset	155 processes = os.cpu_count() or 1
70909 1f171dd21bdb Issue #12285: multiprocessing.Pool() raises a ValueError if the number of Victor Stinner <victor.stinner@haypocalc.com> parents: 69237 diff changeset	156 if processes < 1:
1f171dd21bdb Issue #12285: multiprocessing.Pool() raises a ValueError if the number of Victor Stinner <victor.stinner@haypocalc.com> parents: 69237 diff changeset	157 raise ValueError("Number of processes must be at least 1")
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	158
73165 8e57b5d8f58f Closes #13258: Use callable() built-in in the standard library. Florent Xicluna <florent.xicluna@gmail.com> parents: 73100 diff changeset	159 if initializer is not None and not callable(initializer):
53077 48485604ed48 Merged revisions 70912,70944,70968,71033,71041,71208,71263,71286,71395-71396,71405-71406,71485,71492,71494 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: 49809 diff changeset	160 raise TypeError('initializer must be a callable')
48485604ed48 Merged revisions 70912,70944,70968,71033,71041,71208,71263,71286,71395-71396,71405-71406,71485,71492,71494 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: 49809 diff changeset	161
58537 7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	162 self._processes = processes
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	163 self._pool = []
58537 7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	164 self._repopulate_pool()
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	165
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	166 self._worker_handler = threading.Thread(
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	167 target=Pool._handle_workers,
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	168 args=(self, )
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	169 )
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	170 self._worker_handler.daemon = True
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	171 self._worker_handler._state = RUN
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	172 self._worker_handler.start()
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	173
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	174
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	175 self._task_handler = threading.Thread(
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	176 target=Pool._handle_tasks,
86707 a2230a8420a5 Issue #19425 -- a pickling error should not cause pool to hang. Richard Oudkerk <shibturn@gmail.com> parents: 83640 diff changeset	177 args=(self._taskqueue, self._quick_put, self._outqueue,
a2230a8420a5 Issue #19425 -- a pickling error should not cause pool to hang. Richard Oudkerk <shibturn@gmail.com> parents: 83640 diff changeset	178 self._pool, self._cache)
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	179 )
48749 6dda01643106 Merged revisions 65828 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: 47615 diff changeset	180 self._task_handler.daemon = True
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	181 self._task_handler._state = RUN
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	182 self._task_handler.start()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	183
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	184 self._result_handler = threading.Thread(
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	185 target=Pool._handle_results,
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	186 args=(self._outqueue, self._quick_get, self._cache)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	187 )
48749 6dda01643106 Merged revisions 65828 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: 47615 diff changeset	188 self._result_handler.daemon = True
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	189 self._result_handler._state = RUN
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	190 self._result_handler.start()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	191
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	192 self._terminate = util.Finalize(
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	193 self, self._terminate_pool,
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	194 args=(self._taskqueue, self._inqueue, self._outqueue, self._pool,
58537 7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	195 self._worker_handler, self._task_handler,
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	196 self._result_handler, self._cache),
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	197 exitpriority=15
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	198 )
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	199
58537 7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	200 def _join_exited_workers(self):
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	201 """Cleanup after any worker processes which have exited due to reaching
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	202 their specified lifetime. Returns True if any workers were cleaned up.
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	203 """
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	204 cleaned = False
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	205 for i in reversed(range(len(self._pool))):
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	206 worker = self._pool[i]
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	207 if worker.exitcode is not None:
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	208 # worker exited
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	209 util.debug('cleaning up worker %d' % i)
58537 7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	210 worker.join()
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	211 cleaned = True
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	212 del self._pool[i]
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	213 return cleaned
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	214
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	215 def _repopulate_pool(self):
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	216 """Bring the number of pool processes up to the specified number,
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	217 for use after reaping workers which have exited.
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	218 """
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	219 for i in range(self._processes - len(self._pool)):
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	220 w = self.Process(target=worker,
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	221 args=(self._inqueue, self._outqueue,
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	222 self._initializer,
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	223 self._initargs, self._maxtasksperchild)
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	224 )
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	225 self._pool.append(w)
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	226 w.name = w.name.replace('Process', 'PoolWorker')
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	227 w.daemon = True
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	228 w.start()
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	229 util.debug('added worker')
58537 7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	230
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	231 def _maintain_pool(self):
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	232 """Clean up any exited workers and start replacements for them.
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	233 """
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	234 if self._join_exited_workers():
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	235 self._repopulate_pool()
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	236
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	237 def _setup_queues(self):
86387 72a5ac909c7a Issue #18999: Make multiprocessing use context objects. Richard Oudkerk <shibturn@gmail.com> parents: 85169 diff changeset	238 self._inqueue = self._ctx.SimpleQueue()
72a5ac909c7a Issue #18999: Make multiprocessing use context objects. Richard Oudkerk <shibturn@gmail.com> parents: 85169 diff changeset	239 self._outqueue = self._ctx.SimpleQueue()
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	240 self._quick_put = self._inqueue._writer.send
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	241 self._quick_get = self._outqueue._reader.recv
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	242
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	243 def apply(self, func, args=(), kwds={}):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	244 '''
49809 a4220c92f7cc #4206: fix 2.xisms in multiprocessing docs and docstrings. Georg Brandl <georg@python.org> parents: 48773 diff changeset	245 Equivalent of `func(args, *kwds)`.
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	246 '''
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	247 assert self._state == RUN
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	248 return self.apply_async(func, args, kwds).get()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	249
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	250 def map(self, func, iterable, chunksize=None):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	251 '''
49809 a4220c92f7cc #4206: fix 2.xisms in multiprocessing docs and docstrings. Georg Brandl <georg@python.org> parents: 48773 diff changeset	252 Apply `func` to each element in `iterable`, collecting the results
a4220c92f7cc #4206: fix 2.xisms in multiprocessing docs and docstrings. Georg Brandl <georg@python.org> parents: 48773 diff changeset	253 in a list that is returned.
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	254 '''
74107 b07b1e58582d Issue #12708: Add starmap() and starmap_async() methods (similar to itertools.starmap()) to multiprocessing.Pool. Antoine Pitrou <solipsis@pitrou.net> parents: 73165 diff changeset	255 return self._map_async(func, iterable, mapstar, chunksize).get()
b07b1e58582d Issue #12708: Add starmap() and starmap_async() methods (similar to itertools.starmap()) to multiprocessing.Pool. Antoine Pitrou <solipsis@pitrou.net> parents: 73165 diff changeset	256
b07b1e58582d Issue #12708: Add starmap() and starmap_async() methods (similar to itertools.starmap()) to multiprocessing.Pool. Antoine Pitrou <solipsis@pitrou.net> parents: 73165 diff changeset	257 def starmap(self, func, iterable, chunksize=None):
b07b1e58582d Issue #12708: Add starmap() and starmap_async() methods (similar to itertools.starmap()) to multiprocessing.Pool. Antoine Pitrou <solipsis@pitrou.net> parents: 73165 diff changeset	258 '''
b07b1e58582d Issue #12708: Add starmap() and starmap_async() methods (similar to itertools.starmap()) to multiprocessing.Pool. Antoine Pitrou <solipsis@pitrou.net> parents: 73165 diff changeset	259 Like `map()` method but the elements of the `iterable` are expected to
b07b1e58582d Issue #12708: Add starmap() and starmap_async() methods (similar to itertools.starmap()) to multiprocessing.Pool. Antoine Pitrou <solipsis@pitrou.net> parents: 73165 diff changeset	260 be iterables as well and will be unpacked as arguments. Hence
b07b1e58582d Issue #12708: Add starmap() and starmap_async() methods (similar to itertools.starmap()) to multiprocessing.Pool. Antoine Pitrou <solipsis@pitrou.net> parents: 73165 diff changeset	261 `func` and (a, b) becomes func(a, b).
b07b1e58582d Issue #12708: Add starmap() and starmap_async() methods (similar to itertools.starmap()) to multiprocessing.Pool. Antoine Pitrou <solipsis@pitrou.net> parents: 73165 diff changeset	262 '''
b07b1e58582d Issue #12708: Add starmap() and starmap_async() methods (similar to itertools.starmap()) to multiprocessing.Pool. Antoine Pitrou <solipsis@pitrou.net> parents: 73165 diff changeset	263 return self._map_async(func, iterable, starmapstar, chunksize).get()
b07b1e58582d Issue #12708: Add starmap() and starmap_async() methods (similar to itertools.starmap()) to multiprocessing.Pool. Antoine Pitrou <solipsis@pitrou.net> parents: 73165 diff changeset	264
b07b1e58582d Issue #12708: Add starmap() and starmap_async() methods (similar to itertools.starmap()) to multiprocessing.Pool. Antoine Pitrou <solipsis@pitrou.net> parents: 73165 diff changeset	265 def starmap_async(self, func, iterable, chunksize=None, callback=None,
b07b1e58582d Issue #12708: Add starmap() and starmap_async() methods (similar to itertools.starmap()) to multiprocessing.Pool. Antoine Pitrou <solipsis@pitrou.net> parents: 73165 diff changeset	266 error_callback=None):
b07b1e58582d Issue #12708: Add starmap() and starmap_async() methods (similar to itertools.starmap()) to multiprocessing.Pool. Antoine Pitrou <solipsis@pitrou.net> parents: 73165 diff changeset	267 '''
b07b1e58582d Issue #12708: Add starmap() and starmap_async() methods (similar to itertools.starmap()) to multiprocessing.Pool. Antoine Pitrou <solipsis@pitrou.net> parents: 73165 diff changeset	268 Asynchronous version of `starmap()` method.
b07b1e58582d Issue #12708: Add starmap() and starmap_async() methods (similar to itertools.starmap()) to multiprocessing.Pool. Antoine Pitrou <solipsis@pitrou.net> parents: 73165 diff changeset	269 '''
b07b1e58582d Issue #12708: Add starmap() and starmap_async() methods (similar to itertools.starmap()) to multiprocessing.Pool. Antoine Pitrou <solipsis@pitrou.net> parents: 73165 diff changeset	270 return self._map_async(func, iterable, starmapstar, chunksize,
b07b1e58582d Issue #12708: Add starmap() and starmap_async() methods (similar to itertools.starmap()) to multiprocessing.Pool. Antoine Pitrou <solipsis@pitrou.net> parents: 73165 diff changeset	271 callback, error_callback)
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	272
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	273 def imap(self, func, iterable, chunksize=1):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	274 '''
49809 a4220c92f7cc #4206: fix 2.xisms in multiprocessing docs and docstrings. Georg Brandl <georg@python.org> parents: 48773 diff changeset	275 Equivalent of `map()` -- can be MUCH slower than `Pool.map()`.
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	276 '''
79179 d24b996d136e raise a ValueError instead of an AssertionError when pool is an invalid state Benjamin Peterson <benjamin@python.org> parents: 77509 diff changeset	277 if self._state != RUN:
d24b996d136e raise a ValueError instead of an AssertionError when pool is an invalid state Benjamin Peterson <benjamin@python.org> parents: 77509 diff changeset	278 raise ValueError("Pool not running")
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	279 if chunksize == 1:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	280 result = IMapIterator(self._cache)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	281 self._taskqueue.put((((result._job, i, func, (x,), {})
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	282 for i, x in enumerate(iterable)), result._set_length))
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	283 return result
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	284 else:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	285 assert chunksize > 1
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	286 task_batches = Pool._get_tasks(func, iterable, chunksize)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	287 result = IMapIterator(self._cache)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	288 self._taskqueue.put((((result._job, i, mapstar, (x,), {})
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	289 for i, x in enumerate(task_batches)), result._set_length))
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	290 return (item for chunk in result for item in chunk)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	291
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	292 def imap_unordered(self, func, iterable, chunksize=1):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	293 '''
49809 a4220c92f7cc #4206: fix 2.xisms in multiprocessing docs and docstrings. Georg Brandl <georg@python.org> parents: 48773 diff changeset	294 Like `imap()` method but ordering of results is arbitrary.
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	295 '''
79179 d24b996d136e raise a ValueError instead of an AssertionError when pool is an invalid state Benjamin Peterson <benjamin@python.org> parents: 77509 diff changeset	296 if self._state != RUN:
d24b996d136e raise a ValueError instead of an AssertionError when pool is an invalid state Benjamin Peterson <benjamin@python.org> parents: 77509 diff changeset	297 raise ValueError("Pool not running")
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	298 if chunksize == 1:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	299 result = IMapUnorderedIterator(self._cache)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	300 self._taskqueue.put((((result._job, i, func, (x,), {})
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	301 for i, x in enumerate(iterable)), result._set_length))
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	302 return result
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	303 else:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	304 assert chunksize > 1
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	305 task_batches = Pool._get_tasks(func, iterable, chunksize)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	306 result = IMapUnorderedIterator(self._cache)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	307 self._taskqueue.put((((result._job, i, mapstar, (x,), {})
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	308 for i, x in enumerate(task_batches)), result._set_length))
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	309 return (item for chunk in result for item in chunk)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	310
66115 0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	311 def apply_async(self, func, args=(), kwds={}, callback=None,
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	312 error_callback=None):
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	313 '''
49809 a4220c92f7cc #4206: fix 2.xisms in multiprocessing docs and docstrings. Georg Brandl <georg@python.org> parents: 48773 diff changeset	314 Asynchronous version of `apply()` method.
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	315 '''
79179 d24b996d136e raise a ValueError instead of an AssertionError when pool is an invalid state Benjamin Peterson <benjamin@python.org> parents: 77509 diff changeset	316 if self._state != RUN:
d24b996d136e raise a ValueError instead of an AssertionError when pool is an invalid state Benjamin Peterson <benjamin@python.org> parents: 77509 diff changeset	317 raise ValueError("Pool not running")
66115 0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	318 result = ApplyResult(self._cache, callback, error_callback)
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	319 self._taskqueue.put(([(result._job, None, func, args, kwds)], None))
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	320 return result
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	321
66115 0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	322 def map_async(self, func, iterable, chunksize=None, callback=None,
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	323 error_callback=None):
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	324 '''
49809 a4220c92f7cc #4206: fix 2.xisms in multiprocessing docs and docstrings. Georg Brandl <georg@python.org> parents: 48773 diff changeset	325 Asynchronous version of `map()` method.
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	326 '''
79961 1bccec4ff980 #16307: Fix multiprocessing.Pool.map_async not calling its callbacks Hynek Schlawack <hs@ox.cx> parents: 79179 diff changeset	327 return self._map_async(func, iterable, mapstar, chunksize, callback,
1bccec4ff980 #16307: Fix multiprocessing.Pool.map_async not calling its callbacks Hynek Schlawack <hs@ox.cx> parents: 79179 diff changeset	328 error_callback)
74107 b07b1e58582d Issue #12708: Add starmap() and starmap_async() methods (similar to itertools.starmap()) to multiprocessing.Pool. Antoine Pitrou <solipsis@pitrou.net> parents: 73165 diff changeset	329
b07b1e58582d Issue #12708: Add starmap() and starmap_async() methods (similar to itertools.starmap()) to multiprocessing.Pool. Antoine Pitrou <solipsis@pitrou.net> parents: 73165 diff changeset	330 def _map_async(self, func, iterable, mapper, chunksize=None, callback=None,
b07b1e58582d Issue #12708: Add starmap() and starmap_async() methods (similar to itertools.starmap()) to multiprocessing.Pool. Antoine Pitrou <solipsis@pitrou.net> parents: 73165 diff changeset	331 error_callback=None):
b07b1e58582d Issue #12708: Add starmap() and starmap_async() methods (similar to itertools.starmap()) to multiprocessing.Pool. Antoine Pitrou <solipsis@pitrou.net> parents: 73165 diff changeset	332 '''
b07b1e58582d Issue #12708: Add starmap() and starmap_async() methods (similar to itertools.starmap()) to multiprocessing.Pool. Antoine Pitrou <solipsis@pitrou.net> parents: 73165 diff changeset	333 Helper function to implement map, starmap and their async counterparts.
b07b1e58582d Issue #12708: Add starmap() and starmap_async() methods (similar to itertools.starmap()) to multiprocessing.Pool. Antoine Pitrou <solipsis@pitrou.net> parents: 73165 diff changeset	334 '''
79179 d24b996d136e raise a ValueError instead of an AssertionError when pool is an invalid state Benjamin Peterson <benjamin@python.org> parents: 77509 diff changeset	335 if self._state != RUN:
d24b996d136e raise a ValueError instead of an AssertionError when pool is an invalid state Benjamin Peterson <benjamin@python.org> parents: 77509 diff changeset	336 raise ValueError("Pool not running")
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	337 if not hasattr(iterable, '__len__'):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	338 iterable = list(iterable)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	339
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	340 if chunksize is None:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	341 chunksize, extra = divmod(len(iterable), len(self._pool) * 4)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	342 if extra:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	343 chunksize += 1
55618 0148417e1301 Merged revisions 74055 via svnmerge from Georg Brandl <georg@python.org> parents: 53077 diff changeset	344 if len(iterable) == 0:
0148417e1301 Merged revisions 74055 via svnmerge from Georg Brandl <georg@python.org> parents: 53077 diff changeset	345 chunksize = 0
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	346
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	347 task_batches = Pool._get_tasks(func, iterable, chunksize)
66115 0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	348 result = MapResult(self._cache, chunksize, len(iterable), callback,
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	349 error_callback=error_callback)
74107 b07b1e58582d Issue #12708: Add starmap() and starmap_async() methods (similar to itertools.starmap()) to multiprocessing.Pool. Antoine Pitrou <solipsis@pitrou.net> parents: 73165 diff changeset	350 self._taskqueue.put((((result._job, i, mapper, (x,), {})
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	351 for i, x in enumerate(task_batches)), None))
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	352 return result
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	353
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	354 @staticmethod
58537 7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	355 def _handle_workers(pool):
73100 52c98a729a71 Issue #10332: multiprocessing: fix a race condition when a Pool is closed Charles-François Natali <neologix@free.fr> parents: 70909 diff changeset	356 thread = threading.current_thread()
52c98a729a71 Issue #10332: multiprocessing: fix a race condition when a Pool is closed Charles-François Natali <neologix@free.fr> parents: 70909 diff changeset	357
52c98a729a71 Issue #10332: multiprocessing: fix a race condition when a Pool is closed Charles-François Natali <neologix@free.fr> parents: 70909 diff changeset	358 # Keep maintaining workers until the cache gets drained, unless the pool
52c98a729a71 Issue #10332: multiprocessing: fix a race condition when a Pool is closed Charles-François Natali <neologix@free.fr> parents: 70909 diff changeset	359 # is terminated.
52c98a729a71 Issue #10332: multiprocessing: fix a race condition when a Pool is closed Charles-François Natali <neologix@free.fr> parents: 70909 diff changeset	360 while thread._state == RUN or (pool._cache and thread._state != TERMINATE):
58537 7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	361 pool._maintain_pool()
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	362 time.sleep(0.1)
69236 d5e43afeede6 Issue #8428: Fix a race condition in multiprocessing.Pool when terminating Antoine Pitrou <solipsis@pitrou.net> parents: 68257 diff changeset	363 # send sentinel to stop workers
d5e43afeede6 Issue #8428: Fix a race condition in multiprocessing.Pool when terminating Antoine Pitrou <solipsis@pitrou.net> parents: 68257 diff changeset	364 pool._taskqueue.put(None)
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	365 util.debug('worker handler exiting')
58537 7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	366
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	367 @staticmethod
86707 a2230a8420a5 Issue #19425 -- a pickling error should not cause pool to hang. Richard Oudkerk <shibturn@gmail.com> parents: 83640 diff changeset	368 def _handle_tasks(taskqueue, put, outqueue, pool, cache):
47615 045f83a1b864 Merged revisions 64125 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: 47595 diff changeset	369 thread = threading.current_thread()
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	370
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	371 for taskseq, set_length in iter(taskqueue.get, None):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	372 i = -1
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	373 for i, task in enumerate(taskseq):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	374 if thread._state:
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	375 util.debug('task handler found thread._state != RUN')
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	376 break
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	377 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	378 put(task)
86707 a2230a8420a5 Issue #19425 -- a pickling error should not cause pool to hang. Richard Oudkerk <shibturn@gmail.com> parents: 83640 diff changeset	379 except Exception as e:
a2230a8420a5 Issue #19425 -- a pickling error should not cause pool to hang. Richard Oudkerk <shibturn@gmail.com> parents: 83640 diff changeset	380 job, ind = task[:2]
a2230a8420a5 Issue #19425 -- a pickling error should not cause pool to hang. Richard Oudkerk <shibturn@gmail.com> parents: 83640 diff changeset	381 try:
a2230a8420a5 Issue #19425 -- a pickling error should not cause pool to hang. Richard Oudkerk <shibturn@gmail.com> parents: 83640 diff changeset	382 cache[job]._set(ind, (False, e))
a2230a8420a5 Issue #19425 -- a pickling error should not cause pool to hang. Richard Oudkerk <shibturn@gmail.com> parents: 83640 diff changeset	383 except KeyError:
a2230a8420a5 Issue #19425 -- a pickling error should not cause pool to hang. Richard Oudkerk <shibturn@gmail.com> parents: 83640 diff changeset	384 pass
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	385 else:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	386 if set_length:
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	387 util.debug('doing set_length()')
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	388 set_length(i+1)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	389 continue
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	390 break
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	391 else:
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	392 util.debug('task handler got sentinel')
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	393
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	394
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	395 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	396 # tell result handler to finish when cache is empty
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	397 util.debug('task handler sending sentinel to result handler')
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	398 outqueue.put(None)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	399
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	400 # tell workers there is no more work
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	401 util.debug('task handler sending sentinel to workers')
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	402 for p in pool:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	403 put(None)
81039 7d69d04522e3 Replace IOError with OSError (#16715) Andrew Svetlov <andrew.svetlov@gmail.com> parents: 79961 diff changeset	404 except OSError:
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	405 util.debug('task handler got OSError when sending sentinels')
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	406
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	407 util.debug('task handler exiting')
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	408
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	409 @staticmethod
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	410 def _handle_results(outqueue, get, cache):
47615 045f83a1b864 Merged revisions 64125 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: 47595 diff changeset	411 thread = threading.current_thread()
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	412
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	413 while 1:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	414 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	415 task = get()
81039 7d69d04522e3 Replace IOError with OSError (#16715) Andrew Svetlov <andrew.svetlov@gmail.com> parents: 79961 diff changeset	416 except (OSError, EOFError):
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	417 util.debug('result handler got EOFError/OSError -- exiting')
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	418 return
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	419
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	420 if thread._state:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	421 assert thread._state == TERMINATE
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	422 util.debug('result handler found thread._state=TERMINATE')
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	423 break
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	424
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	425 if task is None:
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	426 util.debug('result handler got sentinel')
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	427 break
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	428
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	429 job, i, obj = task
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	430 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	431 cache[job]._set(i, obj)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	432 except KeyError:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	433 pass
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	434
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	435 while cache and thread._state != TERMINATE:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	436 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	437 task = get()
81039 7d69d04522e3 Replace IOError with OSError (#16715) Andrew Svetlov <andrew.svetlov@gmail.com> parents: 79961 diff changeset	438 except (OSError, EOFError):
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	439 util.debug('result handler got EOFError/OSError -- exiting')
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	440 return
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	441
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	442 if task is None:
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	443 util.debug('result handler ignoring extra sentinel')
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	444 continue
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	445 job, i, obj = task
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	446 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	447 cache[job]._set(i, obj)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	448 except KeyError:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	449 pass
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	450
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	451 if hasattr(outqueue, '_reader'):
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	452 util.debug('ensuring that outqueue is not full')
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	453 # If we don't make room available in outqueue then
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	454 # attempts to add the sentinel (None) to outqueue may
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	455 # block. There is guaranteed to be no more than 2 sentinels.
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	456 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	457 for i in range(10):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	458 if not outqueue._reader.poll():
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	459 break
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	460 get()
81039 7d69d04522e3 Replace IOError with OSError (#16715) Andrew Svetlov <andrew.svetlov@gmail.com> parents: 79961 diff changeset	461 except (OSError, EOFError):
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	462 pass
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	463
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	464 util.debug('result handler exiting: len(cache)=%s, thread._state=%s',
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	465 len(cache), thread._state)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	466
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	467 @staticmethod
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	468 def _get_tasks(func, it, size):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	469 it = iter(it)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	470 while 1:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	471 x = tuple(itertools.islice(it, size))
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	472 if not x:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	473 return
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	474 yield (func, x)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	475
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	476 def __reduce__(self):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	477 raise NotImplementedError(
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	478 'pool objects cannot be passed between processes or pickled'
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	479 )
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	480
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	481 def close(self):
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	482 util.debug('closing pool')
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	483 if self._state == RUN:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	484 self._state = CLOSE
58537 7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	485 self._worker_handler._state = CLOSE
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	486
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	487 def terminate(self):
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	488 util.debug('terminating pool')
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	489 self._state = TERMINATE
58537 7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	490 self._worker_handler._state = TERMINATE
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	491 self._terminate()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	492
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	493 def join(self):
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	494 util.debug('joining pool')
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	495 assert self._state in (CLOSE, TERMINATE)
58537 7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	496 self._worker_handler.join()
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	497 self._task_handler.join()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	498 self._result_handler.join()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	499 for p in self._pool:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	500 p.join()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	501
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	502 @staticmethod
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	503 def _help_stuff_finish(inqueue, task_handler, size):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	504 # task_handler may be blocked trying to put items on inqueue
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	505 util.debug('removing tasks from inqueue until task handler finished')
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	506 inqueue._rlock.acquire()
47615 045f83a1b864 Merged revisions 64125 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: 47595 diff changeset	507 while task_handler.is_alive() and inqueue._reader.poll():
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	508 inqueue._reader.recv()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	509 time.sleep(0)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	510
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	511 @classmethod
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	512 def _terminate_pool(cls, taskqueue, inqueue, outqueue, pool,
58537 7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	513 worker_handler, task_handler, result_handler, cache):
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	514 # this is guaranteed to only be called once
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	515 util.debug('finalizing pool')
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	516
58537 7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	517 worker_handler._state = TERMINATE
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	518 task_handler._state = TERMINATE
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	519
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	520 util.debug('helping task handler/workers to finish')
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	521 cls._help_stuff_finish(inqueue, task_handler, len(pool))
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	522
47615 045f83a1b864 Merged revisions 64125 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: 47595 diff changeset	523 assert result_handler.is_alive() or len(cache) == 0
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	524
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	525 result_handler._state = TERMINATE
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	526 outqueue.put(None) # sentinel
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	527
69236 d5e43afeede6 Issue #8428: Fix a race condition in multiprocessing.Pool when terminating Antoine Pitrou <solipsis@pitrou.net> parents: 68257 diff changeset	528 # We must wait for the worker handler to exit before terminating
d5e43afeede6 Issue #8428: Fix a race condition in multiprocessing.Pool when terminating Antoine Pitrou <solipsis@pitrou.net> parents: 68257 diff changeset	529 # workers because we don't want workers to be restarted behind our back.
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	530 util.debug('joining worker handler')
77507 e1cd1f430ff1 Issue #15101: Make pool finalizer avoid joining current thread. Richard Oudkerk <shibturn@gmail.com> parents: 77374 diff changeset	531 if threading.current_thread() is not worker_handler:
e1cd1f430ff1 Issue #15101: Make pool finalizer avoid joining current thread. Richard Oudkerk <shibturn@gmail.com> parents: 77374 diff changeset	532 worker_handler.join()
69236 d5e43afeede6 Issue #8428: Fix a race condition in multiprocessing.Pool when terminating Antoine Pitrou <solipsis@pitrou.net> parents: 68257 diff changeset	533
58537 7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	534 # Terminate workers which haven't already finished.
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	535 if pool and hasattr(pool[0], 'terminate'):
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	536 util.debug('terminating workers')
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	537 for p in pool:
58537 7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	538 if p.exitcode is None:
7a36ca04742d Merged revisions 77794 via svnmerge from Jesse Noller <jnoller@gmail.com> parents: 55618 diff changeset	539 p.terminate()
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	540
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	541 util.debug('joining task handler')
77507 e1cd1f430ff1 Issue #15101: Make pool finalizer avoid joining current thread. Richard Oudkerk <shibturn@gmail.com> parents: 77374 diff changeset	542 if threading.current_thread() is not task_handler:
e1cd1f430ff1 Issue #15101: Make pool finalizer avoid joining current thread. Richard Oudkerk <shibturn@gmail.com> parents: 77374 diff changeset	543 task_handler.join()
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	544
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	545 util.debug('joining result handler')
77507 e1cd1f430ff1 Issue #15101: Make pool finalizer avoid joining current thread. Richard Oudkerk <shibturn@gmail.com> parents: 77374 diff changeset	546 if threading.current_thread() is not result_handler:
e1cd1f430ff1 Issue #15101: Make pool finalizer avoid joining current thread. Richard Oudkerk <shibturn@gmail.com> parents: 77374 diff changeset	547 result_handler.join()
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	548
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	549 if pool and hasattr(pool[0], 'terminate'):
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	550 util.debug('joining pool workers')
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	551 for p in pool:
60063 e63d4678fca7 Note: only the relevant parts of r79474 are merged. Florent Xicluna <florent.xicluna@gmail.com> parents: 55618 diff changeset	552 if p.is_alive():
e63d4678fca7 Note: only the relevant parts of r79474 are merged. Florent Xicluna <florent.xicluna@gmail.com> parents: 55618 diff changeset	553 # worker has not yet exited
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	554 util.debug('cleaning up worker %d' % p.pid)
60063 e63d4678fca7 Note: only the relevant parts of r79474 are merged. Florent Xicluna <florent.xicluna@gmail.com> parents: 55618 diff changeset	555 p.join()
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	556
77509 6d2a773d8e00 Issue #15064: Implement context manager protocol for multiprocessing types Richard Oudkerk <shibturn@gmail.com> parents: 77508 diff changeset	557 def __enter__(self):
6d2a773d8e00 Issue #15064: Implement context manager protocol for multiprocessing types Richard Oudkerk <shibturn@gmail.com> parents: 77508 diff changeset	558 return self
6d2a773d8e00 Issue #15064: Implement context manager protocol for multiprocessing types Richard Oudkerk <shibturn@gmail.com> parents: 77508 diff changeset	559
6d2a773d8e00 Issue #15064: Implement context manager protocol for multiprocessing types Richard Oudkerk <shibturn@gmail.com> parents: 77508 diff changeset	560 def __exit__(self, exc_type, exc_val, exc_tb):
6d2a773d8e00 Issue #15064: Implement context manager protocol for multiprocessing types Richard Oudkerk <shibturn@gmail.com> parents: 77508 diff changeset	561 self.terminate()
6d2a773d8e00 Issue #15064: Implement context manager protocol for multiprocessing types Richard Oudkerk <shibturn@gmail.com> parents: 77508 diff changeset	562
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	563 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	564 # Class whose instances are returned by `Pool.apply_async()`
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	565 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	566
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	567 class ApplyResult(object):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	568
66115 0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	569 def __init__(self, cache, callback, error_callback):
77140 57d6265beaaa Issue #12091: simplify ApplyResult and MapResult with threading.Event Richard Oudkerk <shibturn@gmail.com> parents: 76663 diff changeset	570 self._event = threading.Event()
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	571 self._job = next(job_counter)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	572 self._cache = cache
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	573 self._callback = callback
66115 0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	574 self._error_callback = error_callback
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	575 cache[self._job] = self
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	576
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	577 def ready(self):
77140 57d6265beaaa Issue #12091: simplify ApplyResult and MapResult with threading.Event Richard Oudkerk <shibturn@gmail.com> parents: 76663 diff changeset	578 return self._event.is_set()
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	579
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	580 def successful(self):
77140 57d6265beaaa Issue #12091: simplify ApplyResult and MapResult with threading.Event Richard Oudkerk <shibturn@gmail.com> parents: 76663 diff changeset	581 assert self.ready()
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	582 return self._success
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	583
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	584 def wait(self, timeout=None):
77140 57d6265beaaa Issue #12091: simplify ApplyResult and MapResult with threading.Event Richard Oudkerk <shibturn@gmail.com> parents: 76663 diff changeset	585 self._event.wait(timeout)
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	586
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	587 def get(self, timeout=None):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	588 self.wait(timeout)
77140 57d6265beaaa Issue #12091: simplify ApplyResult and MapResult with threading.Event Richard Oudkerk <shibturn@gmail.com> parents: 76663 diff changeset	589 if not self.ready():
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	590 raise TimeoutError
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	591 if self._success:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	592 return self._value
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	593 else:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	594 raise self._value
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	596 def _set(self, i, obj):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	597 self._success, self._value = obj
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	598 if self._callback and self._success:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	599 self._callback(self._value)
66115 0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	600 if self._error_callback and not self._success:
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	601 self._error_callback(self._value)
77140 57d6265beaaa Issue #12091: simplify ApplyResult and MapResult with threading.Event Richard Oudkerk <shibturn@gmail.com> parents: 76663 diff changeset	602 self._event.set()
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	603 del self._cache[self._job]
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	604
83640 bb4bb2db6106 Issue #17805: Add AsyncResult alias for ApplyResult Richard Oudkerk <shibturn@gmail.com> parents: 79961 diff changeset	605 AsyncResult = ApplyResult # create alias -- see #17805
bb4bb2db6106 Issue #17805: Add AsyncResult alias for ApplyResult Richard Oudkerk <shibturn@gmail.com> parents: 79961 diff changeset	606
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	607 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	608 # Class whose instances are returned by `Pool.map_async()`
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	609 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	610
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	611 class MapResult(ApplyResult):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	612
66115 0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	613 def __init__(self, cache, chunksize, length, callback, error_callback):
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	614 ApplyResult.__init__(self, cache, callback,
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	615 error_callback=error_callback)
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	616 self._success = True
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	617 self._value = [None] * length
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	618 self._chunksize = chunksize
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	619 if chunksize <= 0:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	620 self._number_left = 0
77140 57d6265beaaa Issue #12091: simplify ApplyResult and MapResult with threading.Event Richard Oudkerk <shibturn@gmail.com> parents: 76663 diff changeset	621 self._event.set()
77374 1b3d4ffcb4d1 Issue #12157: pool.map() does not handle empty iterable correctly Richard Oudkerk <shibturn@gmail.com> parents: 73165 diff changeset	622 del cache[self._job]
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	623 else:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	624 self._number_left = length//chunksize + bool(length % chunksize)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	625
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	626 def _set(self, i, success_result):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	627 success, result = success_result
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	628 if success:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	629 self._value[iself._chunksize:(i+1)self._chunksize] = result
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	630 self._number_left -= 1
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	631 if self._number_left == 0:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	632 if self._callback:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	633 self._callback(self._value)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	634 del self._cache[self._job]
77140 57d6265beaaa Issue #12091: simplify ApplyResult and MapResult with threading.Event Richard Oudkerk <shibturn@gmail.com> parents: 76663 diff changeset	635 self._event.set()
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	636 else:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	637 self._success = False
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	638 self._value = result
66115 0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	639 if self._error_callback:
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded Ask Solem <askh@opera.com> parents: 60598 diff changeset	640 self._error_callback(self._value)
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	641 del self._cache[self._job]
77140 57d6265beaaa Issue #12091: simplify ApplyResult and MapResult with threading.Event Richard Oudkerk <shibturn@gmail.com> parents: 76663 diff changeset	642 self._event.set()
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	643
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	644 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	645 # Class whose instances are returned by `Pool.imap()`
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	646 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	647
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	648 class IMapIterator(object):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	649
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	650 def __init__(self, cache):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	651 self._cond = threading.Condition(threading.Lock())
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	652 self._job = next(job_counter)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	653 self._cache = cache
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	654 self._items = collections.deque()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	655 self._index = 0
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	656 self._length = None
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	657 self._unsorted = {}
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	658 cache[self._job] = self
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	659
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	660 def __iter__(self):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	661 return self
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	662
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	663 def next(self, timeout=None):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	664 self._cond.acquire()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	665 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	666 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	667 item = self._items.popleft()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	668 except IndexError:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	669 if self._index == self._length:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	670 raise StopIteration
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	671 self._cond.wait(timeout)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	672 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	673 item = self._items.popleft()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	674 except IndexError:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	675 if self._index == self._length:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	676 raise StopIteration
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	677 raise TimeoutError
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	678 finally:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	679 self._cond.release()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	680
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	681 success, value = item
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	682 if success:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	683 return value
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	684 raise value
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	685
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	686 __next__ = next # XXX
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	687
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	688 def _set(self, i, obj):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	689 self._cond.acquire()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	690 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	691 if self._index == i:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	692 self._items.append(obj)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	693 self._index += 1
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	694 while self._index in self._unsorted:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	695 obj = self._unsorted.pop(self._index)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	696 self._items.append(obj)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	697 self._index += 1
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	698 self._cond.notify()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	699 else:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	700 self._unsorted[i] = obj
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	701
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	702 if self._index == self._length:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	703 del self._cache[self._job]
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	704 finally:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	705 self._cond.release()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	706
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	707 def _set_length(self, length):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	708 self._cond.acquire()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	709 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	710 self._length = length
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	711 if self._index == self._length:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	712 self._cond.notify()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	713 del self._cache[self._job]
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	714 finally:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	715 self._cond.release()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	716
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	717 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	718 # Class whose instances are returned by `Pool.imap_unordered()`
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	719 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	720
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	721 class IMapUnorderedIterator(IMapIterator):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	722
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	723 def _set(self, i, obj):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	724 self._cond.acquire()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	725 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	726 self._items.append(obj)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	727 self._index += 1
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	728 self._cond.notify()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	729 if self._index == self._length:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	730 del self._cache[self._job]
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	731 finally:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	732 self._cond.release()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	733
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	734 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	735 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	736 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	737
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	738 class ThreadPool(Pool):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	739
85169 3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	740 @staticmethod
3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	741 def Process(args, *kwds):
3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	742 from .dummy import Process
3b82e0d83bf9 Issue #8713: Support alternative start methods in multiprocessing on Unix. Richard Oudkerk <shibturn@gmail.com> parents: 84355 diff changeset	743 return Process(args, *kwds)
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	744
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	745 def __init__(self, processes=None, initializer=None, initargs=()):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	746 Pool.__init__(self, processes, initializer, initargs)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	747
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	748 def _setup_queues(self):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	749 self._inqueue = queue.Queue()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	750 self._outqueue = queue.Queue()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	751 self._quick_put = self._inqueue.put
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	752 self._quick_get = self._outqueue.get
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	753
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	754 @staticmethod
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	755 def _help_stuff_finish(inqueue, task_handler, size):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	756 # put sentinels at head of inqueue to make workers finish
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	757 inqueue.not_empty.acquire()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	758 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	759 inqueue.queue.clear()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	760 inqueue.queue.extend([None] * size)
47615 045f83a1b864 Merged revisions 64125 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: 47595 diff changeset	761 inqueue.not_empty.notify_all()
47595 5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	762 finally:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from Benjamin Peterson <benjamin@python.org> parents: diff changeset	763 inqueue.not_empty.release()

Mercurial > cpython-lru_cache-weakref

annotate Lib/multiprocessing/pool.py @ 87690:66c1c9f32567 default tip