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
|
#!/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 Exception, 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:
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 _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():
work_item.future._condition.acquire()
work_item.future._condition.notify_all()
work_item.future._condition.release()
work_item.completion_tracker.add_cancelled()
continue
else:
work_item.future._condition.acquire()
work_item.future._state = RUNNING
work_item.future._condition.release()
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.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.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()
self._queue_count
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()
|
