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
|
#!/usr/bin/env python
from futures._base import (PENDING, RUNNING, CANCELLED,
CANCELLED_AND_NOTIFIED, FINISHED,
ALL_COMPLETED,
LOGGER,
set_future_exception, set_future_result,
Executor, Future, FutureList, ThreadEventSink)
import atexit
import Queue
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()
def _remove_dead_thread_references():
for thread_reference in set(_thread_references):
if thread_reference() is None:
_thread_references.discard(thread_reference)
class _WorkItem(object):
def __init__(self, call, future, completion_tracker):
self.call = call
self.future = future
self.completion_tracker = completion_tracker
def run(self):
self.future._condition.acquire()
try:
if self.future._state == PENDING:
self.future._state = RUNNING
elif self.future._state == CANCELLED:
self.completion_tracker._condition.acquire()
try:
self.future._state = CANCELLED_AND_NOTIFIED
self.completion_tracker.add_cancelled()
return
finally:
self.completion_tracker._condition.release()
else:
LOGGER.critical('Future %s in unexpected state: %d',
id(self.future),
self.future._state)
return
finally:
self.future._condition.release()
try:
result = self.call()
except Exception, e:
set_future_exception(self.future, self.completion_tracker, e)
else:
set_future_result(self.future, self.completion_tracker, result)
def _worker(executor_reference, work_queue):
try:
while True:
try:
work_item = work_queue.get(block=True, timeout=0.1)
except Queue.Empty:
executor = executor_reference()
if _shutdown or executor is None or executor._shutdown:
return
del executor
else:
work_item.run()
except Exception, e:
LOGGER.critical('Exception in worker', exc_info=True)
class ThreadPoolExecutor(Executor):
def __init__(self, max_threads):
_remove_dead_thread_references()
self._max_threads = max_threads
self._work_queue = Queue.Queue()
self._threads = set()
self._shutdown = False
self._shutdown_lock = threading.Lock()
def _adjust_thread_count(self):
for _ in range(len(self._threads),
min(self._max_threads, self._work_queue.qsize())):
t = threading.Thread(target=_worker,
args=(weakref.ref(self), self._work_queue))
t.setDaemon(True)
t.start()
self._threads.add(t)
_thread_references.add(weakref.ref(t))
def run_to_futures(self, calls, timeout=None, return_when=ALL_COMPLETED):
self._shutdown_lock.acquire()
try:
if self._shutdown:
raise RuntimeError('cannot run new futures after shutdown')
futures = []
event_sink = ThreadEventSink()
for index, call in enumerate(calls):
f = Future(index)
w = _WorkItem(call, f, event_sink)
self._work_queue.put(w)
futures.append(f)
self._adjust_thread_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 = True
finally:
self._shutdown_lock.release()
atexit.register(_python_exit)
|
