Mercurial > cpython-lru_cache-weakref

annotate Lib/multiprocessing/pool.py @ 78069:a8aa77458e12 3.2

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Add c_size_t and c_ssize_t to the ctypes summary table.
author Antoine Pitrou <solipsis@pitrou.net>
date Thu, 12 Jul 2012 20:31:50 +0200
parents e1cd1f430ff1
children 59e0a51c5fc3
Ignore whitespace changes - Everywhere: Within whitespace: At end of lines:
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
71fb03274a89 Merged revisions 87225 via svnmerge from
R. David Murray <rdmurray@bitdance.com>
parents: 60063
diff changeset
7 # All rights reserved.
71fb03274a89 Merged revisions 87225 via svnmerge from
R. David Murray <rdmurray@bitdance.com>
parents: 60063
diff changeset
8 #
71fb03274a89 Merged revisions 87225 via svnmerge from
R. David Murray <rdmurray@bitdance.com>
parents: 60063
diff changeset
9 # Redistribution and use in source and binary forms, with or without
71fb03274a89 Merged revisions 87225 via svnmerge from
R. David Murray <rdmurray@bitdance.com>
parents: 60063
diff changeset
10 # modification, are permitted provided that the following conditions
71fb03274a89 Merged revisions 87225 via svnmerge from
R. David Murray <rdmurray@bitdance.com>
parents: 60063
diff changeset
11 # are met:
71fb03274a89 Merged revisions 87225 via svnmerge from
R. David Murray <rdmurray@bitdance.com>
parents: 60063
diff changeset
12 #
71fb03274a89 Merged revisions 87225 via svnmerge from
R. David Murray <rdmurray@bitdance.com>
parents: 60063
diff changeset
13 # 1. Redistributions of source code must retain the above copyright
71fb03274a89 Merged revisions 87225 via svnmerge from
R. David Murray <rdmurray@bitdance.com>
parents: 60063
diff changeset
14 # notice, this list of conditions and the following disclaimer.
71fb03274a89 Merged revisions 87225 via svnmerge from
R. David Murray <rdmurray@bitdance.com>
parents: 60063
diff changeset
15 # 2. Redistributions in binary form must reproduce the above copyright
71fb03274a89 Merged revisions 87225 via svnmerge from
R. David Murray <rdmurray@bitdance.com>
parents: 60063
diff changeset
16 # notice, this list of conditions and the following disclaimer in the
71fb03274a89 Merged revisions 87225 via svnmerge from
R. David Murray <rdmurray@bitdance.com>
parents: 60063
diff changeset
17 # documentation and/or other materials provided with the distribution.
71fb03274a89 Merged revisions 87225 via svnmerge from
R. David Murray <rdmurray@bitdance.com>
parents: 60063
diff changeset
18 # 3. Neither the name of author nor the names of any contributors may be
71fb03274a89 Merged revisions 87225 via svnmerge from
R. David Murray <rdmurray@bitdance.com>
parents: 60063
diff changeset
19 # used to endorse or promote products derived from this software
71fb03274a89 Merged revisions 87225 via svnmerge from
R. David Murray <rdmurray@bitdance.com>
parents: 60063
diff changeset
20 # without specific prior written permission.
71fb03274a89 Merged revisions 87225 via svnmerge from
R. David Murray <rdmurray@bitdance.com>
parents: 60063
diff changeset
21 #
71fb03274a89 Merged revisions 87225 via svnmerge from
R. David Murray <rdmurray@bitdance.com>
parents: 60063
diff changeset
22 # THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS "AS IS" AND
71fb03274a89 Merged revisions 87225 via svnmerge from
R. David Murray <rdmurray@bitdance.com>
parents: 60063
diff changeset
23 # ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
71fb03274a89 Merged revisions 87225 via svnmerge from
R. David Murray <rdmurray@bitdance.com>
parents: 60063
diff changeset
24 # IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
71fb03274a89 Merged revisions 87225 via svnmerge from
R. David Murray <rdmurray@bitdance.com>
parents: 60063
diff changeset
25 # ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
71fb03274a89 Merged revisions 87225 via svnmerge from
R. David Murray <rdmurray@bitdance.com>
parents: 60063
diff changeset
26 # FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
71fb03274a89 Merged revisions 87225 via svnmerge from
R. David Murray <rdmurray@bitdance.com>
parents: 60063
diff changeset
27 # DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
71fb03274a89 Merged revisions 87225 via svnmerge from
R. David Murray <rdmurray@bitdance.com>
parents: 60063
diff changeset
28 # OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
71fb03274a89 Merged revisions 87225 via svnmerge from
R. David Murray <rdmurray@bitdance.com>
parents: 60063
diff changeset
29 # HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
71fb03274a89 Merged revisions 87225 via svnmerge from
R. David Murray <rdmurray@bitdance.com>
parents: 60063
diff changeset
30 # LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
71fb03274a89 Merged revisions 87225 via svnmerge from
R. David Murray <rdmurray@bitdance.com>
parents: 60063
diff changeset
31 # OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
71fb03274a89 Merged revisions 87225 via svnmerge from
R. David Murray <rdmurray@bitdance.com>
parents: 60063
diff changeset
32 # SUCH DAMAGE.
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
33 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
34
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
35 __all__ = ['Pool']
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 # Imports
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 import threading
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
42 import queue
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
43 import itertools
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
44 import collections
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
45 import time
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
46
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
47 from multiprocessing import Process, cpu_count, TimeoutError
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
48 from multiprocessing.util import Finalize, debug
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
49
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
50 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
51 # Constants representing the state of a pool
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
52 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
53
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
54 RUN = 0
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
55 CLOSE = 1
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
56 TERMINATE = 2
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
57
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
58 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
59 # Miscellaneous
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
60 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
61
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
62 job_counter = itertools.count()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
63
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
64 def mapstar(args):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
65 return list(map(*args))
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
66
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
67 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
68 # Code run by worker processes
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
69 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
70
66115
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded
Ask Solem <askh@opera.com>
parents: 60598
diff changeset
71 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
72 """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
73 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
74
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded
Ask Solem <askh@opera.com>
parents: 60598
diff changeset
75 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
76 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
77 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
78 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
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 __str__(self):
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded
Ask Solem <askh@opera.com>
parents: 60598
diff changeset
81 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
82 self.exc)
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded
Ask Solem <askh@opera.com>
parents: 60598
diff changeset
83
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded
Ask Solem <askh@opera.com>
parents: 60598
diff changeset
84 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
85 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
86
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded
Ask Solem <askh@opera.com>
parents: 60598
diff changeset
87
58537
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
88 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
89 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
90 put = outqueue.put
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
91 get = inqueue.get
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
92 if hasattr(inqueue, '_writer'):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
93 inqueue._writer.close()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
94 outqueue._reader.close()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
95
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
96 if initializer is not None:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
97 initializer(*initargs)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
98
58537
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
99 completed = 0
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
100 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
101 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
102 task = get()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
103 except (EOFError, IOError):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
104 debug('worker got EOFError or IOError -- exiting')
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
105 break
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
106
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
107 if task is None:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
108 debug('worker got sentinel -- exiting')
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
109 break
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
110
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
111 job, i, func, args, kwds = task
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
112 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
113 result = (True, func(*args, **kwds))
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
114 except Exception as e:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
115 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
116 try:
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded
Ask Solem <askh@opera.com>
parents: 60598
diff changeset
117 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
118 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
119 wrapped = MaybeEncodingError(e, result[1])
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded
Ask Solem <askh@opera.com>
parents: 60598
diff changeset
120 debug("Possible encoding error while sending result: %s" % (
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded
Ask Solem <askh@opera.com>
parents: 60598
diff changeset
121 wrapped))
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded
Ask Solem <askh@opera.com>
parents: 60598
diff changeset
122 put((job, i, (False, wrapped)))
58537
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
123 completed += 1
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
124 debug('worker exiting after %d tasks' % completed)
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
125
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
126 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
127 # Class representing a process pool
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
128 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
129
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
130 class Pool(object):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
131 '''
49809
a4220c92f7cc #4206: fix 2.xisms in multiprocessing docs and docstrings.
Georg Brandl <georg@python.org>
parents: 48773
diff changeset
132 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
133 '''
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
134 Process = Process
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
135
58537
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
136 def __init__(self, processes=None, initializer=None, initargs=(),
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
137 maxtasksperchild=None):
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
138 self._setup_queues()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
139 self._taskqueue = queue.Queue()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
140 self._cache = {}
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
141 self._state = RUN
58537
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
142 self._maxtasksperchild = maxtasksperchild
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
143 self._initializer = initializer
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
144 self._initargs = initargs
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
145
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
146 if processes is None:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
147 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
148 processes = cpu_count()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
149 except NotImplementedError:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
150 processes = 1
70909
1f171dd21bdb Issue #12285: multiprocessing.Pool() raises a ValueError if the number of
Victor Stinner <victor.stinner@haypocalc.com>
parents: 69237
diff changeset
151 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
152 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
153
73165
8e57b5d8f58f Closes #13258: Use callable() built-in in the standard library.
Florent Xicluna <florent.xicluna@gmail.com>
parents: 73100
diff changeset
154 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
155 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
156
58537
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
157 self._processes = processes
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
158 self._pool = []
58537
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
159 self._repopulate_pool()
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
160
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
161 self._worker_handler = threading.Thread(
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
162 target=Pool._handle_workers,
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
163 args=(self, )
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
164 )
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
165 self._worker_handler.daemon = True
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
166 self._worker_handler._state = RUN
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
167 self._worker_handler.start()
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
168
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
169
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
170 self._task_handler = threading.Thread(
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
171 target=Pool._handle_tasks,
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
172 args=(self._taskqueue, self._quick_put, self._outqueue, self._pool)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
173 )
48749
6dda01643106 Merged revisions 65828 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents: 47615
diff changeset
174 self._task_handler.daemon = True
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
175 self._task_handler._state = RUN
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
176 self._task_handler.start()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
177
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
178 self._result_handler = threading.Thread(
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
179 target=Pool._handle_results,
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
180 args=(self._outqueue, self._quick_get, self._cache)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
181 )
48749
6dda01643106 Merged revisions 65828 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents: 47615
diff changeset
182 self._result_handler.daemon = True
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
183 self._result_handler._state = RUN
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
184 self._result_handler.start()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
185
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
186 self._terminate = Finalize(
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
187 self, self._terminate_pool,
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
188 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
189 self._worker_handler, self._task_handler,
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
190 self._result_handler, self._cache),
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
191 exitpriority=15
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
192 )
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
193
58537
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
194 def _join_exited_workers(self):
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
195 """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
196 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
197 """
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
198 cleaned = False
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
199 for i in reversed(range(len(self._pool))):
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
200 worker = self._pool[i]
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
201 if worker.exitcode is not None:
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
202 # worker exited
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
203 debug('cleaning up worker %d' % i)
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
204 worker.join()
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
205 cleaned = True
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
206 del self._pool[i]
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
207 return cleaned
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
208
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
209 def _repopulate_pool(self):
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
210 """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
211 for use after reaping workers which have exited.
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
212 """
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
213 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
214 w = self.Process(target=worker,
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
215 args=(self._inqueue, self._outqueue,
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
216 self._initializer,
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
217 self._initargs, self._maxtasksperchild)
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 self._pool.append(w)
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
220 w.name = w.name.replace('Process', 'PoolWorker')
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
221 w.daemon = True
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
222 w.start()
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
223 debug('added worker')
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 def _maintain_pool(self):
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
226 """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
227 """
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
228 if self._join_exited_workers():
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
229 self._repopulate_pool()
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
230
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
231 def _setup_queues(self):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
232 from .queues import SimpleQueue
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
233 self._inqueue = SimpleQueue()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
234 self._outqueue = SimpleQueue()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
235 self._quick_put = self._inqueue._writer.send
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
236 self._quick_get = self._outqueue._reader.recv
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
237
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
238 def apply(self, func, args=(), kwds={}):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
239 '''
49809
a4220c92f7cc #4206: fix 2.xisms in multiprocessing docs and docstrings.
Georg Brandl <georg@python.org>
parents: 48773
diff changeset
240 Equivalent of `func(*args, **kwds)`.
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
241 '''
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
242 assert self._state == RUN
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
243 return self.apply_async(func, args, kwds).get()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
244
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
245 def map(self, func, iterable, chunksize=None):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
246 '''
49809
a4220c92f7cc #4206: fix 2.xisms in multiprocessing docs and docstrings.
Georg Brandl <georg@python.org>
parents: 48773
diff changeset
247 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
248 in a list that is returned.
47595
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 assert self._state == RUN
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
251 return self.map_async(func, iterable, chunksize).get()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
252
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
253 def imap(self, func, iterable, chunksize=1):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
254 '''
49809
a4220c92f7cc #4206: fix 2.xisms in multiprocessing docs and docstrings.
Georg Brandl <georg@python.org>
parents: 48773
diff changeset
255 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
256 '''
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
257 assert self._state == RUN
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
258 if chunksize == 1:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
259 result = IMapIterator(self._cache)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
260 self._taskqueue.put((((result._job, i, func, (x,), {})
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
261 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
262 return result
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
263 else:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
264 assert chunksize > 1
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
265 task_batches = Pool._get_tasks(func, iterable, chunksize)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
266 result = IMapIterator(self._cache)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
267 self._taskqueue.put((((result._job, i, mapstar, (x,), {})
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
268 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
269 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
270
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
271 def imap_unordered(self, func, iterable, chunksize=1):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
272 '''
49809
a4220c92f7cc #4206: fix 2.xisms in multiprocessing docs and docstrings.
Georg Brandl <georg@python.org>
parents: 48773
diff changeset
273 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
274 '''
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
275 assert self._state == RUN
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
276 if chunksize == 1:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
277 result = IMapUnorderedIterator(self._cache)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
278 self._taskqueue.put((((result._job, i, func, (x,), {})
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
279 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
280 return result
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
281 else:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
282 assert chunksize > 1
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
283 task_batches = Pool._get_tasks(func, iterable, chunksize)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
284 result = IMapUnorderedIterator(self._cache)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
285 self._taskqueue.put((((result._job, i, mapstar, (x,), {})
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
286 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
287 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
288
66115
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded
Ask Solem <askh@opera.com>
parents: 60598
diff changeset
289 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
290 error_callback=None):
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
291 '''
49809
a4220c92f7cc #4206: fix 2.xisms in multiprocessing docs and docstrings.
Georg Brandl <georg@python.org>
parents: 48773
diff changeset
292 Asynchronous version of `apply()` method.
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
293 '''
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
294 assert self._state == RUN
66115
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded
Ask Solem <askh@opera.com>
parents: 60598
diff changeset
295 result = ApplyResult(self._cache, callback, error_callback)
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
296 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
297 return result
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
298
66115
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded
Ask Solem <askh@opera.com>
parents: 60598
diff changeset
299 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
300 error_callback=None):
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
301 '''
49809
a4220c92f7cc #4206: fix 2.xisms in multiprocessing docs and docstrings.
Georg Brandl <georg@python.org>
parents: 48773
diff changeset
302 Asynchronous version of `map()` method.
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
303 '''
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
304 assert self._state == RUN
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
305 if not hasattr(iterable, '__len__'):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
306 iterable = list(iterable)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
307
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
308 if chunksize is None:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
309 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
310 if extra:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
311 chunksize += 1
55618
0148417e1301 Merged revisions 74055 via svnmerge from
Georg Brandl <georg@python.org>
parents: 53077
diff changeset
312 if len(iterable) == 0:
0148417e1301 Merged revisions 74055 via svnmerge from
Georg Brandl <georg@python.org>
parents: 53077
diff changeset
313 chunksize = 0
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
314
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
315 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
316 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
317 error_callback=error_callback)
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
318 self._taskqueue.put((((result._job, i, mapstar, (x,), {})
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
319 for i, x in enumerate(task_batches)), 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
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
322 @staticmethod
58537
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
323 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
324 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
325
52c98a729a71 Issue #10332: multiprocessing: fix a race condition when a Pool is closed
Charles-François Natali <neologix@free.fr>
parents: 70909
diff changeset
326 # 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
327 # 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
328 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
329 pool._maintain_pool()
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
330 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
331 # 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
332 pool._taskqueue.put(None)
58537
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
333 debug('worker handler exiting')
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
334
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
335 @staticmethod
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
336 def _handle_tasks(taskqueue, put, outqueue, pool):
47615
045f83a1b864 Merged revisions 64125 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents: 47595
diff changeset
337 thread = threading.current_thread()
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
338
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
339 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
340 i = -1
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
341 for i, task in enumerate(taskseq):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
342 if thread._state:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
343 debug('task handler found thread._state != RUN')
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
344 break
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
345 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
346 put(task)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
347 except IOError:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
348 debug('could not put task on queue')
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
349 break
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
350 else:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
351 if set_length:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
352 debug('doing set_length()')
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
353 set_length(i+1)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
354 continue
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
355 break
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
356 else:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
357 debug('task handler got sentinel')
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
358
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
359
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
360 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
361 # tell result handler to finish when cache is empty
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
362 debug('task handler sending sentinel to result handler')
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
363 outqueue.put(None)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
364
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
365 # tell workers there is no more work
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
366 debug('task handler sending sentinel to workers')
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
367 for p in pool:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
368 put(None)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
369 except IOError:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
370 debug('task handler got IOError when sending sentinels')
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
371
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
372 debug('task handler exiting')
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
373
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
374 @staticmethod
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
375 def _handle_results(outqueue, get, cache):
47615
045f83a1b864 Merged revisions 64125 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents: 47595
diff changeset
376 thread = threading.current_thread()
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
377
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
378 while 1:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
379 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
380 task = get()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
381 except (IOError, EOFError):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
382 debug('result handler got EOFError/IOError -- exiting')
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
383 return
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
384
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
385 if thread._state:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
386 assert thread._state == TERMINATE
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
387 debug('result handler found thread._state=TERMINATE')
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
388 break
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
389
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
390 if task is None:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
391 debug('result handler got sentinel')
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
392 break
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 job, i, obj = task
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 cache[job]._set(i, obj)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
397 except KeyError:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
398 pass
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 while cache and thread._state != TERMINATE:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
401 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
402 task = get()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
403 except (IOError, EOFError):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
404 debug('result handler got EOFError/IOError -- exiting')
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
405 return
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
406
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
407 if task is None:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
408 debug('result handler ignoring extra sentinel')
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
409 continue
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
410 job, i, obj = task
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
411 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
412 cache[job]._set(i, obj)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
413 except KeyError:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
414 pass
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
415
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
416 if hasattr(outqueue, '_reader'):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
417 debug('ensuring that outqueue is not full')
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
418 # 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
419 # 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
420 # 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
421 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
422 for i in range(10):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
423 if not outqueue._reader.poll():
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
424 break
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
425 get()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
426 except (IOError, EOFError):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
427 pass
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 debug('result handler exiting: len(cache)=%s, thread._state=%s',
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
430 len(cache), thread._state)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
431
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
432 @staticmethod
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
433 def _get_tasks(func, it, size):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
434 it = iter(it)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
435 while 1:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
436 x = tuple(itertools.islice(it, size))
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
437 if not x:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
438 return
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
439 yield (func, x)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
440
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
441 def __reduce__(self):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
442 raise NotImplementedError(
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
443 '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
444 )
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
445
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
446 def close(self):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
447 debug('closing pool')
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
448 if self._state == RUN:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
449 self._state = CLOSE
58537
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
450 self._worker_handler._state = CLOSE
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
451
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
452 def terminate(self):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
453 debug('terminating pool')
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
454 self._state = TERMINATE
58537
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
455 self._worker_handler._state = TERMINATE
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
456 self._terminate()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
457
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
458 def join(self):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
459 debug('joining pool')
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
460 assert self._state in (CLOSE, TERMINATE)
58537
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
461 self._worker_handler.join()
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
462 self._task_handler.join()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
463 self._result_handler.join()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
464 for p in self._pool:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
465 p.join()
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 _help_stuff_finish(inqueue, task_handler, size):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
469 # task_handler may be blocked trying to put items on inqueue
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
470 debug('removing tasks from inqueue until task handler finished')
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
471 inqueue._rlock.acquire()
47615
045f83a1b864 Merged revisions 64125 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents: 47595
diff changeset
472 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
473 inqueue._reader.recv()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
474 time.sleep(0)
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 @classmethod
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
477 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
478 worker_handler, task_handler, result_handler, cache):
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
479 # this is guaranteed to only be called once
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
480 debug('finalizing pool')
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
481
58537
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
482 worker_handler._state = TERMINATE
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
483 task_handler._state = TERMINATE
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
484
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
485 debug('helping task handler/workers to finish')
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
486 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
487
47615
045f83a1b864 Merged revisions 64125 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents: 47595
diff changeset
488 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
489
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
490 result_handler._state = TERMINATE
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
491 outqueue.put(None) # sentinel
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
492
69236
d5e43afeede6 Issue #8428: Fix a race condition in multiprocessing.Pool when terminating
Antoine Pitrou <solipsis@pitrou.net>
parents: 68257
diff changeset
493 # 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
494 # workers because we don't want workers to be restarted behind our back.
d5e43afeede6 Issue #8428: Fix a race condition in multiprocessing.Pool when terminating
Antoine Pitrou <solipsis@pitrou.net>
parents: 68257
diff changeset
495 debug('joining worker handler')
77507
e1cd1f430ff1 Issue #15101: Make pool finalizer avoid joining current thread.
Richard Oudkerk <shibturn@gmail.com>
parents: 77374
diff changeset
496 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
497 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
498
58537
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
499 # Terminate workers which haven't already finished.
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
500 if pool and hasattr(pool[0], 'terminate'):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
501 debug('terminating workers')
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
502 for p in pool:
58537
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
503 if p.exitcode is None:
7a36ca04742d Merged revisions 77794 via svnmerge from
Jesse Noller <jnoller@gmail.com>
parents: 55618
diff changeset
504 p.terminate()
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
505
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
506 debug('joining task handler')
77507
e1cd1f430ff1 Issue #15101: Make pool finalizer avoid joining current thread.
Richard Oudkerk <shibturn@gmail.com>
parents: 77374
diff changeset
507 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
508 task_handler.join()
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
509
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
510 debug('joining result handler')
77507
e1cd1f430ff1 Issue #15101: Make pool finalizer avoid joining current thread.
Richard Oudkerk <shibturn@gmail.com>
parents: 77374
diff changeset
511 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
512 result_handler.join()
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
513
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
514 if pool and hasattr(pool[0], 'terminate'):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
515 debug('joining pool workers')
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
516 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
517 if p.is_alive():
e63d4678fca7 Note: only the relevant parts of r79474 are merged.
Florent Xicluna <florent.xicluna@gmail.com>
parents: 55618
diff changeset
518 # worker has not yet exited
e63d4678fca7 Note: only the relevant parts of r79474 are merged.
Florent Xicluna <florent.xicluna@gmail.com>
parents: 55618
diff changeset
519 debug('cleaning up worker %d' % p.pid)
e63d4678fca7 Note: only the relevant parts of r79474 are merged.
Florent Xicluna <florent.xicluna@gmail.com>
parents: 55618
diff changeset
520 p.join()
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
521
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
522 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
523 # 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
524 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
525
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
526 class ApplyResult(object):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
527
66115
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded
Ask Solem <askh@opera.com>
parents: 60598
diff changeset
528 def __init__(self, cache, callback, error_callback):
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
529 self._cond = threading.Condition(threading.Lock())
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
530 self._job = next(job_counter)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
531 self._cache = cache
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
532 self._ready = False
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
533 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
534 self._error_callback = error_callback
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
535 cache[self._job] = self
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
536
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
537 def ready(self):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
538 return self._ready
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
539
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
540 def successful(self):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
541 assert self._ready
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
542 return self._success
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
543
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
544 def wait(self, timeout=None):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
545 self._cond.acquire()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
546 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
547 if not self._ready:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
548 self._cond.wait(timeout)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
549 finally:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
550 self._cond.release()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
551
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
552 def get(self, timeout=None):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
553 self.wait(timeout)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
554 if not self._ready:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
555 raise TimeoutError
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
556 if self._success:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
557 return self._value
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
558 else:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
559 raise self._value
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
560
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
561 def _set(self, i, obj):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
562 self._success, self._value = obj
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
563 if self._callback and self._success:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
564 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
565 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
566 self._error_callback(self._value)
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
567 self._cond.acquire()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
568 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
569 self._ready = True
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
570 self._cond.notify()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
571 finally:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
572 self._cond.release()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
573 del self._cache[self._job]
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
574
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
575 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
576 # 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
577 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
578
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
579 class MapResult(ApplyResult):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
580
66115
0aa8af79359d Issue #9244: multiprocessing.pool: Worker crashes if result can't be encoded
Ask Solem <askh@opera.com>
parents: 60598
diff changeset
581 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
582 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
583 error_callback=error_callback)
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
584 self._success = True
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
585 self._value = [None] * length
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
586 self._chunksize = chunksize
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
587 if chunksize <= 0:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
588 self._number_left = 0
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
589 self._ready = True
77374
1b3d4ffcb4d1 Issue #12157: pool.map() does not handle empty iterable correctly
Richard Oudkerk <shibturn@gmail.com>
parents: 73165
diff changeset
590 del cache[self._job]
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
591 else:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
592 self._number_left = length//chunksize + bool(length % chunksize)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
593
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
594 def _set(self, i, success_result):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
595 success, result = success_result
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
596 if success:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
597 self._value[i*self._chunksize:(i+1)*self._chunksize] = result
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
598 self._number_left -= 1
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
599 if self._number_left == 0:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
600 if self._callback:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
601 self._callback(self._value)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
602 del self._cache[self._job]
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
603 self._cond.acquire()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
604 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
605 self._ready = True
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
606 self._cond.notify()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
607 finally:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
608 self._cond.release()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
609 else:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
610 self._success = False
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
611 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
612 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
613 self._error_callback(self._value)
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
614 del self._cache[self._job]
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
615 self._cond.acquire()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
616 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
617 self._ready = True
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
618 self._cond.notify()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
619 finally:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
620 self._cond.release()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
621
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
622 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
623 # Class whose instances are returned by `Pool.imap()`
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
624 #
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 class IMapIterator(object):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
627
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
628 def __init__(self, cache):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
629 self._cond = threading.Condition(threading.Lock())
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
630 self._job = next(job_counter)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
631 self._cache = cache
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
632 self._items = collections.deque()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
633 self._index = 0
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
634 self._length = None
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
635 self._unsorted = {}
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
636 cache[self._job] = self
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
637
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
638 def __iter__(self):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
639 return self
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
640
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
641 def next(self, timeout=None):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
642 self._cond.acquire()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
643 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
644 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
645 item = self._items.popleft()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
646 except IndexError:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
647 if self._index == self._length:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
648 raise StopIteration
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
649 self._cond.wait(timeout)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
650 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
651 item = self._items.popleft()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
652 except IndexError:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
653 if self._index == self._length:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
654 raise StopIteration
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
655 raise TimeoutError
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
656 finally:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
657 self._cond.release()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
658
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
659 success, value = item
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
660 if success:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
661 return value
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
662 raise value
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
663
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
664 __next__ = next # XXX
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
665
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
666 def _set(self, i, obj):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
667 self._cond.acquire()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
668 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
669 if self._index == i:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
670 self._items.append(obj)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
671 self._index += 1
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
672 while self._index in self._unsorted:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
673 obj = self._unsorted.pop(self._index)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
674 self._items.append(obj)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
675 self._index += 1
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
676 self._cond.notify()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
677 else:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
678 self._unsorted[i] = obj
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
679
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
680 if self._index == self._length:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
681 del self._cache[self._job]
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
682 finally:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
683 self._cond.release()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
684
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
685 def _set_length(self, length):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
686 self._cond.acquire()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
687 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
688 self._length = length
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
689 if self._index == self._length:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
690 self._cond.notify()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
691 del self._cache[self._job]
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
692 finally:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
693 self._cond.release()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
694
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
695 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
696 # 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
697 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
698
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
699 class IMapUnorderedIterator(IMapIterator):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
700
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
701 def _set(self, i, obj):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
702 self._cond.acquire()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
703 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
704 self._items.append(obj)
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
705 self._index += 1
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
706 self._cond.notify()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
707 if self._index == self._length:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
708 del self._cache[self._job]
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
709 finally:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
710 self._cond.release()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
711
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
712 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
713 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
714 #
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
715
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
716 class ThreadPool(Pool):
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 from .dummy import Process
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 def __init__(self, processes=None, initializer=None, initargs=()):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
721 Pool.__init__(self, processes, initializer, initargs)
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 _setup_queues(self):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
724 self._inqueue = queue.Queue()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
725 self._outqueue = queue.Queue()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
726 self._quick_put = self._inqueue.put
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
727 self._quick_get = self._outqueue.get
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
728
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
729 @staticmethod
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
730 def _help_stuff_finish(inqueue, task_handler, size):
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
731 # 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
732 inqueue.not_empty.acquire()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
733 try:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
734 inqueue.queue.clear()
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
735 inqueue.queue.extend([None] * size)
47615
045f83a1b864 Merged revisions 64125 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents: 47595
diff changeset
736 inqueue.not_empty.notify_all()
47595
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
737 finally:
5c6cdfcd6eae Merged revisions 64104,64117 via svnmerge from
Benjamin Peterson <benjamin@python.org>
parents:
diff changeset
738 inqueue.not_empty.release()