1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
|
# Copyright 2009 Brian Quinlan. All Rights Reserved. See LICENSE file.
"""Implements ProcessPoolExecutor."""
__author__ = 'Brian Quinlan (brian@sweetapp.com)'
from futures._base import (PENDING, RUNNING, CANCELLED,
CANCELLED_AND_NOTIFIED, FINISHED,
ALL_COMPLETED,
set_future_exception, set_future_result,
Executor, Future, FutureList, ThreadEventSink)
import atexit
import Queue
import multiprocessing
import threading
import weakref
_thread_references = set()
_shutdown = False
def _python_exit():
global _shutdown
_shutdown = True
for thread_reference in _thread_references:
thread = thread_reference()
if thread is not None:
thread.join()
class _WorkItem(object):
def __init__(self, call, future, completion_tracker):
self.call = call
self.future = future
self.completion_tracker = completion_tracker
class _ResultItem(object):
def __init__(self, work_id, exception=None, result=None):
self.work_id = work_id
self.exception = exception
self.result = result
class _CallItem(object):
def __init__(self, work_id, call):
self.work_id = work_id
self.call = call
def _process_worker(call_queue, result_queue, shutdown):
while True:
try:
call_item = call_queue.get(block=True, timeout=0.1)
except Queue.Empty:
if shutdown.is_set():
return
else:
try:
r = call_item.call()
except Exception, e:
result_queue.put(_ResultItem(call_item.work_id,
exception=e))
else:
result_queue.put(_ResultItem(call_item.work_id,
result=r))
def _add_call_item_to_queue(pending_work_items,
work_ids,
call_queue):
while True:
if call_queue.full():
return
try:
work_id = work_ids.get(block=False)
except Queue.Empty:
return
else:
work_item = pending_work_items[work_id]
if work_item.future.cancelled():
work_item.future._condition.acquire()
work_item.future._condition.notify_all()
work_item.future._condition.release()
work_item.completion_tracker.add_cancelled()
del pending_work_items[work_id]
continue
else:
work_item.future._condition.acquire()
work_item.future._state = RUNNING
work_item.future._condition.release()
call_queue.put(_CallItem(work_id, work_item.call), block=True)
def _result(executor_reference,
processes,
pending_work_items,
work_ids_queue,
call_queue,
result_queue,
shutdown_process_event):
while True:
_add_call_item_to_queue(pending_work_items,
work_ids_queue,
call_queue)
try:
result_item = result_queue.get(block=True, timeout=0.1)
except Queue.Empty:
executor = executor_reference()
# No more work items can be added if:
# - The interpreter is shutting down OR
# - The executor that owns this worker has been collected OR
# - The executor that owns this worker has been shutdown.
if _shutdown or executor is None or executor._shutdown_thread:
# Since no new work items can be added, it is safe to shutdown
# this thread if there are no pending work items.
if not pending_work_items:
shutdown_process_event.set()
# If .join() is not called on the created processes then
# some multiprocessing.Queue methods may deadlock on Mac OS
# X.
for p in processes:
p.join()
return
del executor
else:
work_item = pending_work_items[result_item.work_id]
del pending_work_items[result_item.work_id]
if result_item.exception:
set_future_exception(work_item.future,
work_item.completion_tracker,
result_item.exception)
else:
set_future_result(work_item.future,
work_item.completion_tracker,
result_item.result)
class ProcessPoolExecutor(Executor):
def __init__(self, max_processes=None):
import warnings
warnings.warn('ProcessPoolExecutor has known deadlocking behavior')
if max_processes is None:
max_processes = multiprocessing.cpu_count()
self._max_processes = max_processes
# Make the call queue slightly larger than the number of processes to
# prevent the worker processes from starving but to make future.cancel()
# responsive.
self._call_queue = multiprocessing.Queue(self._max_processes + 1)
self._result_queue = multiprocessing.Queue()
self._work_ids = Queue.Queue()
self._queue_management_thread = None
self._processes = set()
# Shutdown is a two-step process.
self._shutdown_thread = False
self._shutdown_process_event = multiprocessing.Event()
self._shutdown_lock = threading.Lock()
self._queue_count = 0
self._pending_work_items = {}
def _adjust_process_count(self):
if self._queue_management_thread is None:
self._queue_management_thread = threading.Thread(
target=_result,
args=(weakref.ref(self),
self._processes,
self._pending_work_items,
self._work_ids,
self._call_queue,
self._result_queue,
self._shutdown_process_event))
self._queue_management_thread.setDaemon(True)
self._queue_management_thread.start()
_thread_references.add(weakref.ref(self._queue_management_thread))
for _ in range(len(self._processes), self._max_processes):
p = multiprocessing.Process(
target=_process_worker,
args=(self._call_queue,
self._result_queue,
self._shutdown_process_event))
p.start()
self._processes.add(p)
def run_to_futures(self, calls, timeout=None, return_when=ALL_COMPLETED):
self._shutdown_lock.acquire()
try:
if self._shutdown_thread:
raise RuntimeError('cannot run new futures after shutdown')
futures = []
event_sink = ThreadEventSink()
for index, call in enumerate(calls):
f = Future(index)
self._pending_work_items[self._queue_count] = _WorkItem(
call, f, event_sink)
self._work_ids.put(self._queue_count)
futures.append(f)
self._queue_count += 1
self._adjust_process_count()
fl = FutureList(futures, event_sink)
fl.wait(timeout=timeout, return_when=return_when)
return fl
finally:
self._shutdown_lock.release()
def shutdown(self):
self._shutdown_lock.acquire()
try:
self._shutdown_thread = True
finally:
self._shutdown_lock.release()
atexit.register(_python_exit)
|
