forked from apache/beam
-
Notifications
You must be signed in to change notification settings - Fork 0
/
thread_pool_executor.py
136 lines (113 loc) · 4.2 KB
/
thread_pool_executor.py
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
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http:https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
# pytype: skip-file
from __future__ import absolute_import
import sys
import threading
import weakref
from concurrent.futures import _base
try: # Python3
import queue
except Exception: # Python2
import Queue as queue # type: ignore[no-redef]
class _WorkItem(object):
def __init__(self, future, fn, args, kwargs):
self._future = future
self._fn = fn
self._fn_args = args
self._fn_kwargs = kwargs
def run(self):
if self._future.set_running_or_notify_cancel():
# If the future wasn't cancelled, then attempt to execute it.
try:
self._future.set_result(self._fn(*self._fn_args, **self._fn_kwargs))
except BaseException as exc:
# Even though Python 2 futures library has #set_exection(),
# the way it generates the traceback doesn't align with
# the way in which Python 3 does it so we provide alternative
# implementations that match our test expectations.
if sys.version_info.major >= 3:
self._future.set_exception(exc)
else:
e, tb = sys.exc_info()[1:]
self._future.set_exception_info(e, tb)
class _Worker(threading.Thread):
def __init__(self, idle_worker_queue, work_item):
super(_Worker, self).__init__()
self._idle_worker_queue = idle_worker_queue
self._work_item = work_item
self._wake_semaphore = threading.Semaphore(0)
self._lock = threading.Lock()
self._shutdown = False
def run(self):
while True:
self._work_item.run()
self._work_item = None
self._idle_worker_queue.put(self)
self._wake_semaphore.acquire()
if self._work_item is None:
return
def assign_work(self, work_item):
"""Assigns the work item and wakes up the thread.
This method must only be called while the worker is idle.
"""
self._work_item = work_item
self._wake_semaphore.release()
def shutdown(self):
"""Wakes up this thread with a 'None' work item signalling to shutdown."""
self._wake_semaphore.release()
class UnboundedThreadPoolExecutor(_base.Executor):
def __init__(self):
self._idle_worker_queue = queue.Queue()
self._max_idle_threads = 16
self._workers = weakref.WeakSet()
self._shutdown = False
self._lock = threading.Lock() # Guards access to _workers and _shutdown
def submit(self, fn, *args, **kwargs):
"""Attempts to submit the work item.
A runtime error is raised if the pool has been shutdown.
"""
future = _base.Future()
work_item = _WorkItem(future, fn, args, kwargs)
with self._lock:
if self._shutdown:
raise RuntimeError(
'Cannot schedule new tasks after thread pool has been shutdown.')
try:
self._idle_worker_queue.get(block=False).assign_work(work_item)
# If we have more idle threads then the max allowed, shutdown a thread.
if self._idle_worker_queue.qsize() > self._max_idle_threads:
try:
self._idle_worker_queue.get(block=False).shutdown()
except queue.Empty:
pass
except queue.Empty:
worker = _Worker(self._idle_worker_queue, work_item)
worker.daemon = True
worker.start()
self._workers.add(worker)
return future
def shutdown(self, wait=True):
with self._lock:
if self._shutdown:
return
self._shutdown = True
for worker in self._workers:
worker.shutdown()
if wait:
for worker in self._workers:
worker.join()