| 148 | |
| 149 | |
| 150 | class ThreadPoolExecutor(_base.Executor): |
| 151 | |
| 152 | BROKEN = BrokenThreadPool |
| 153 | |
| 154 | # Used to assign unique thread names when thread_name_prefix is not supplied. |
| 155 | _counter = itertools.count().__next__ |
| 156 | |
| 157 | @classmethod |
| 158 | def prepare_context(cls, initializer, initargs): |
| 159 | return WorkerContext.prepare(initializer, initargs) |
| 160 | |
| 161 | def __init__(self, max_workers=None, thread_name_prefix='', |
| 162 | initializer=None, initargs=(), **ctxkwargs): |
| 163 | """Initializes a new ThreadPoolExecutor instance. |
| 164 | |
| 165 | Args: |
| 166 | max_workers: The maximum number of threads that can be used to |
| 167 | execute the given calls. |
| 168 | thread_name_prefix: An optional name prefix to give our threads. |
| 169 | initializer: A callable used to initialize worker threads. |
| 170 | initargs: A tuple of arguments to pass to the initializer. |
| 171 | ctxkwargs: Additional arguments to cls.prepare_context(). |
| 172 | """ |
| 173 | if max_workers is None: |
| 174 | # ThreadPoolExecutor is often used to: |
| 175 | # * CPU bound task which releases GIL |
| 176 | # * I/O bound task (which releases GIL, of course) |
| 177 | # |
| 178 | # We use process_cpu_count + 4 for both types of tasks. |
| 179 | # But we limit it to 32 to avoid consuming surprisingly large resource |
| 180 | # on many core machine. |
| 181 | max_workers = min(32, (os.process_cpu_count() or 1) + 4) |
| 182 | if max_workers <= 0: |
| 183 | raise ValueError("max_workers must be greater than 0") |
| 184 | |
| 185 | (self._create_worker_context, |
| 186 | self._resolve_work_item_task, |
| 187 | ) = type(self).prepare_context(initializer, initargs, **ctxkwargs) |
| 188 | |
| 189 | self._max_workers = max_workers |
| 190 | self._work_queue = queue.SimpleQueue() |
| 191 | self._idle_semaphore = threading.Semaphore(0) |
| 192 | self._threads = set() |
| 193 | self._broken = False |
| 194 | self._shutdown = False |
| 195 | self._shutdown_lock = threading.Lock() |
| 196 | self._thread_name_prefix = (thread_name_prefix or |
| 197 | ("ThreadPoolExecutor-%d" % self._counter())) |
| 198 | |
| 199 | def submit(self, fn, /, *args, **kwargs): |
| 200 | with self._shutdown_lock, _global_shutdown_lock: |
| 201 | if self._broken: |
| 202 | raise self.BROKEN(self._broken) |
| 203 | |
| 204 | if self._shutdown: |
| 205 | raise RuntimeError('cannot schedule new futures after shutdown') |
| 206 | if _shutdown: |
| 207 | raise RuntimeError('cannot schedule new futures after ' |
searching dependent graphs…