| 120 | |
| 121 | |
| 122 | class ThreadPoolExecutor(_base.Executor): |
| 123 | |
| 124 | # Used to assign unique thread names when thread_name_prefix is not supplied. |
| 125 | _counter = itertools.count().__next__ |
| 126 | |
| 127 | def __init__(self, max_workers=None, thread_name_prefix='', |
| 128 | initializer=None, initargs=()): |
| 129 | """Initializes a new ThreadPoolExecutor instance. |
| 130 | |
| 131 | Args: |
| 132 | max_workers: The maximum number of threads that can be used to |
| 133 | execute the given calls. |
| 134 | thread_name_prefix: An optional name prefix to give our threads. |
| 135 | initializer: A callable used to initialize worker threads. |
| 136 | initargs: A tuple of arguments to pass to the initializer. |
| 137 | """ |
| 138 | if max_workers is None: |
| 139 | # ThreadPoolExecutor is often used to: |
| 140 | # * CPU bound task which releases GIL |
| 141 | # * I/O bound task (which releases GIL, of course) |
| 142 | # |
| 143 | # We use process_cpu_count + 4 for both types of tasks. |
| 144 | # But we limit it to 32 to avoid consuming surprisingly large resource |
| 145 | # on many core machine. |
| 146 | max_workers = min(32, (os.process_cpu_count() or 1) + 4) |
| 147 | if max_workers <= 0: |
| 148 | raise ValueError("max_workers must be greater than 0") |
| 149 | |
| 150 | if initializer is not None and not callable(initializer): |
| 151 | raise TypeError("initializer must be a callable") |
| 152 | |
| 153 | self._max_workers = max_workers |
| 154 | self._work_queue = queue.SimpleQueue() |
| 155 | self._idle_semaphore = threading.Semaphore(0) |
| 156 | self._threads = set() |
| 157 | self._broken = False |
| 158 | self._shutdown = False |
| 159 | self._shutdown_lock = threading.Lock() |
| 160 | self._thread_name_prefix = (thread_name_prefix or |
| 161 | ("ThreadPoolExecutor-%d" % self._counter())) |
| 162 | self._initializer = initializer |
| 163 | self._initargs = initargs |
| 164 | |
| 165 | def submit(self, fn, /, *args, **kwargs): |
| 166 | with self._shutdown_lock, _global_shutdown_lock: |
| 167 | if self._broken: |
| 168 | raise BrokenThreadPool(self._broken) |
| 169 | |
| 170 | if self._shutdown: |
| 171 | raise RuntimeError('cannot schedule new futures after shutdown') |
| 172 | if _shutdown: |
| 173 | raise RuntimeError('cannot schedule new futures after ' |
| 174 | 'interpreter shutdown') |
| 175 | |
| 176 | f = _base.Future() |
| 177 | w = _WorkItem(f, fn, args, kwargs) |
| 178 | |
| 179 | self._work_queue.put(w) |