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
|
#!/usr/bin/env python
from futures._base import (PENDING, RUNNING, CANCELLED,
CANCELLED_AND_NOTIFIED, FINISHED,
ALL_COMPLETED,
set_future_exception, set_future_result,
Executor, Future, FutureList, ThreadEventSink)
import queue
import multiprocessing
import threading
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 BaseException as e:
result_queue.put(_ResultItem(call_item.work_id,
exception=e))
else:
result_queue.put(_ResultItem(call_item.work_id,
result=r))
class ProcessPoolExecutor(Executor):
def __init__(self, max_processes=None):
if max_processes is None:
try:
max_processes = multiprocessing.cpu_count()
except NotImplementedError:
max_processes = 16
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 _add_call_item_to_queue(self):
while True:
try:
work_id = self._work_ids.get(block=False)
except queue.Empty:
return
else:
work_item = self._pending_work_items[work_id]
if work_item.future.cancelled():
with work_item.future._condition:
work_item.future._condition.notify_all()
work_item.completion_tracker.add_cancelled()
continue
else:
with work_item.future._condition:
work_item.future._state = RUNNING
self._call_queue.put(_CallItem(work_id, work_item.call),
block=True)
if self._call_queue.full():
return
def _result(self):
while True:
self._add_call_item_to_queue()
try:
result_item = self._result_queue.get(block=True,
timeout=0.1)
except queue.Empty:
if self._shutdown_thread and not self._pending_work_items:
self._shutdown_process_event.set()
return
else:
work_item = self._pending_work_items[result_item.work_id]
del self._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)
def _adjust_process_count(self):
if self._queue_management_thread is None:
self._queue_management_thread = threading.Thread(
target=self._result)
self._queue_management_thread.daemon = True
self._queue_management_thread.start()
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.daemon = True
p.start()
self._processes.add(p)
def run_to_futures(self, calls, timeout=None, return_when=ALL_COMPLETED):
with self._shutdown_lock:
if self._shutdown_thread:
raise RuntimeError('cannot run new futures after shutdown')
futures = []
event_sink = ThreadEventSink()
self._queue_count
for call in calls:
f = Future()
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
def shutdown(self):
with self._shutdown_lock:
self._shutdown_thread = True
|