1:mod:`concurrent.futures` --- Launching parallel tasks 2====================================================== 3 4.. module:: concurrent.futures 5 :synopsis: Execute computations concurrently using threads or processes. 6 7.. versionadded:: 3.2 8 9**Source code:** :source:`Lib/concurrent/futures/thread.py` 10and :source:`Lib/concurrent/futures/process.py` 11 12-------------- 13 14The :mod:`concurrent.futures` module provides a high-level interface for 15asynchronously executing callables. 16 17The asynchronous execution can be performed with threads, using 18:class:`ThreadPoolExecutor`, or separate processes, using 19:class:`ProcessPoolExecutor`. Both implement the same interface, which is 20defined by the abstract :class:`Executor` class. 21 22 23Executor Objects 24---------------- 25 26.. class:: Executor 27 28 An abstract class that provides methods to execute calls asynchronously. It 29 should not be used directly, but through its concrete subclasses. 30 31 .. method:: submit(fn, *args, **kwargs) 32 33 Schedules the callable, *fn*, to be executed as ``fn(*args **kwargs)`` 34 and returns a :class:`Future` object representing the execution of the 35 callable. :: 36 37 with ThreadPoolExecutor(max_workers=1) as executor: 38 future = executor.submit(pow, 323, 1235) 39 print(future.result()) 40 41 .. method:: map(func, *iterables, timeout=None, chunksize=1) 42 43 Similar to :func:`map(func, *iterables) <map>` except: 44 45 * the *iterables* are collected immediately rather than lazily; 46 47 * *func* is executed asynchronously and several calls to 48 *func* may be made concurrently. 49 50 The returned iterator raises a :exc:`concurrent.futures.TimeoutError` 51 if :meth:`~iterator.__next__` is called and the result isn't available 52 after *timeout* seconds from the original call to :meth:`Executor.map`. 53 *timeout* can be an int or a float. If *timeout* is not specified or 54 ``None``, there is no limit to the wait time. 55 56 If a *func* call raises an exception, then that exception will be 57 raised when its value is retrieved from the iterator. 58 59 When using :class:`ProcessPoolExecutor`, this method chops *iterables* 60 into a number of chunks which it submits to the pool as separate 61 tasks. The (approximate) size of these chunks can be specified by 62 setting *chunksize* to a positive integer. For very long iterables, 63 using a large value for *chunksize* can significantly improve 64 performance compared to the default size of 1. With 65 :class:`ThreadPoolExecutor`, *chunksize* has no effect. 66 67 .. versionchanged:: 3.5 68 Added the *chunksize* argument. 69 70 .. method:: shutdown(wait=True) 71 72 Signal the executor that it should free any resources that it is using 73 when the currently pending futures are done executing. Calls to 74 :meth:`Executor.submit` and :meth:`Executor.map` made after shutdown will 75 raise :exc:`RuntimeError`. 76 77 If *wait* is ``True`` then this method will not return until all the 78 pending futures are done executing and the resources associated with the 79 executor have been freed. If *wait* is ``False`` then this method will 80 return immediately and the resources associated with the executor will be 81 freed when all pending futures are done executing. Regardless of the 82 value of *wait*, the entire Python program will not exit until all 83 pending futures are done executing. 84 85 You can avoid having to call this method explicitly if you use the 86 :keyword:`with` statement, which will shutdown the :class:`Executor` 87 (waiting as if :meth:`Executor.shutdown` were called with *wait* set to 88 ``True``):: 89 90 import shutil 91 with ThreadPoolExecutor(max_workers=4) as e: 92 e.submit(shutil.copy, 'src1.txt', 'dest1.txt') 93 e.submit(shutil.copy, 'src2.txt', 'dest2.txt') 94 e.submit(shutil.copy, 'src3.txt', 'dest3.txt') 95 e.submit(shutil.copy, 'src4.txt', 'dest4.txt') 96 97 98ThreadPoolExecutor 99------------------ 100 101:class:`ThreadPoolExecutor` is an :class:`Executor` subclass that uses a pool of 102threads to execute calls asynchronously. 103 104Deadlocks can occur when the callable associated with a :class:`Future` waits on 105the results of another :class:`Future`. For example:: 106 107 import time 108 def wait_on_b(): 109 time.sleep(5) 110 print(b.result()) # b will never complete because it is waiting on a. 111 return 5 112 113 def wait_on_a(): 114 time.sleep(5) 115 print(a.result()) # a will never complete because it is waiting on b. 116 return 6 117 118 119 executor = ThreadPoolExecutor(max_workers=2) 120 a = executor.submit(wait_on_b) 121 b = executor.submit(wait_on_a) 122 123And:: 124 125 def wait_on_future(): 126 f = executor.submit(pow, 5, 2) 127 # This will never complete because there is only one worker thread and 128 # it is executing this function. 129 print(f.result()) 130 131 executor = ThreadPoolExecutor(max_workers=1) 132 executor.submit(wait_on_future) 133 134 135.. class:: ThreadPoolExecutor(max_workers=None, thread_name_prefix='', initializer=None, initargs=()) 136 137 An :class:`Executor` subclass that uses a pool of at most *max_workers* 138 threads to execute calls asynchronously. 139 140 *initializer* is an optional callable that is called at the start of 141 each worker thread; *initargs* is a tuple of arguments passed to the 142 initializer. Should *initializer* raise an exception, all currently 143 pending jobs will raise a :exc:`~concurrent.futures.thread.BrokenThreadPool`, 144 as well as any attempt to submit more jobs to the pool. 145 146 .. versionchanged:: 3.5 147 If *max_workers* is ``None`` or 148 not given, it will default to the number of processors on the machine, 149 multiplied by ``5``, assuming that :class:`ThreadPoolExecutor` is often 150 used to overlap I/O instead of CPU work and the number of workers 151 should be higher than the number of workers 152 for :class:`ProcessPoolExecutor`. 153 154 .. versionadded:: 3.6 155 The *thread_name_prefix* argument was added to allow users to 156 control the :class:`threading.Thread` names for worker threads created by 157 the pool for easier debugging. 158 159 .. versionchanged:: 3.7 160 Added the *initializer* and *initargs* arguments. 161 162 .. versionchanged:: 3.8 163 Default value of *max_workers* is changed to ``min(32, os.cpu_count() + 4)``. 164 This default value preserves at least 5 workers for I/O bound tasks. 165 It utilizes at most 32 CPU cores for CPU bound tasks which release the GIL. 166 And it avoids using very large resources implicitly on many-core machines. 167 168 ThreadPoolExecutor now reuses idle worker threads before starting 169 *max_workers* worker threads too. 170 171 172.. _threadpoolexecutor-example: 173 174ThreadPoolExecutor Example 175~~~~~~~~~~~~~~~~~~~~~~~~~~ 176:: 177 178 import concurrent.futures 179 import urllib.request 180 181 URLS = ['http://www.foxnews.com/', 182 'http://www.cnn.com/', 183 'http://europe.wsj.com/', 184 'http://www.bbc.co.uk/', 185 'http://some-made-up-domain.com/'] 186 187 # Retrieve a single page and report the URL and contents 188 def load_url(url, timeout): 189 with urllib.request.urlopen(url, timeout=timeout) as conn: 190 return conn.read() 191 192 # We can use a with statement to ensure threads are cleaned up promptly 193 with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor: 194 # Start the load operations and mark each future with its URL 195 future_to_url = {executor.submit(load_url, url, 60): url for url in URLS} 196 for future in concurrent.futures.as_completed(future_to_url): 197 url = future_to_url[future] 198 try: 199 data = future.result() 200 except Exception as exc: 201 print('%r generated an exception: %s' % (url, exc)) 202 else: 203 print('%r page is %d bytes' % (url, len(data))) 204 205 206ProcessPoolExecutor 207------------------- 208 209The :class:`ProcessPoolExecutor` class is an :class:`Executor` subclass that 210uses a pool of processes to execute calls asynchronously. 211:class:`ProcessPoolExecutor` uses the :mod:`multiprocessing` module, which 212allows it to side-step the :term:`Global Interpreter Lock` but also means that 213only picklable objects can be executed and returned. 214 215The ``__main__`` module must be importable by worker subprocesses. This means 216that :class:`ProcessPoolExecutor` will not work in the interactive interpreter. 217 218Calling :class:`Executor` or :class:`Future` methods from a callable submitted 219to a :class:`ProcessPoolExecutor` will result in deadlock. 220 221.. class:: ProcessPoolExecutor(max_workers=None, mp_context=None, initializer=None, initargs=()) 222 223 An :class:`Executor` subclass that executes calls asynchronously using a pool 224 of at most *max_workers* processes. If *max_workers* is ``None`` or not 225 given, it will default to the number of processors on the machine. 226 If *max_workers* is lower or equal to ``0``, then a :exc:`ValueError` 227 will be raised. 228 On Windows, *max_workers* must be equal or lower than ``61``. If it is not 229 then :exc:`ValueError` will be raised. If *max_workers* is ``None``, then 230 the default chosen will be at most ``61``, even if more processors are 231 available. 232 *mp_context* can be a multiprocessing context or None. It will be used to 233 launch the workers. If *mp_context* is ``None`` or not given, the default 234 multiprocessing context is used. 235 236 *initializer* is an optional callable that is called at the start of 237 each worker process; *initargs* is a tuple of arguments passed to the 238 initializer. Should *initializer* raise an exception, all currently 239 pending jobs will raise a :exc:`~concurrent.futures.process.BrokenProcessPool`, 240 as well any attempt to submit more jobs to the pool. 241 242 .. versionchanged:: 3.3 243 When one of the worker processes terminates abruptly, a 244 :exc:`BrokenProcessPool` error is now raised. Previously, behaviour 245 was undefined but operations on the executor or its futures would often 246 freeze or deadlock. 247 248 .. versionchanged:: 3.7 249 The *mp_context* argument was added to allow users to control the 250 start_method for worker processes created by the pool. 251 252 Added the *initializer* and *initargs* arguments. 253 254 255.. _processpoolexecutor-example: 256 257ProcessPoolExecutor Example 258~~~~~~~~~~~~~~~~~~~~~~~~~~~ 259:: 260 261 import concurrent.futures 262 import math 263 264 PRIMES = [ 265 112272535095293, 266 112582705942171, 267 112272535095293, 268 115280095190773, 269 115797848077099, 270 1099726899285419] 271 272 def is_prime(n): 273 if n < 2: 274 return False 275 if n == 2: 276 return True 277 if n % 2 == 0: 278 return False 279 280 sqrt_n = int(math.floor(math.sqrt(n))) 281 for i in range(3, sqrt_n + 1, 2): 282 if n % i == 0: 283 return False 284 return True 285 286 def main(): 287 with concurrent.futures.ProcessPoolExecutor() as executor: 288 for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)): 289 print('%d is prime: %s' % (number, prime)) 290 291 if __name__ == '__main__': 292 main() 293 294 295Future Objects 296-------------- 297 298The :class:`Future` class encapsulates the asynchronous execution of a callable. 299:class:`Future` instances are created by :meth:`Executor.submit`. 300 301.. class:: Future 302 303 Encapsulates the asynchronous execution of a callable. :class:`Future` 304 instances are created by :meth:`Executor.submit` and should not be created 305 directly except for testing. 306 307 .. method:: cancel() 308 309 Attempt to cancel the call. If the call is currently being executed or 310 finished running and cannot be cancelled then the method will return 311 ``False``, otherwise the call will be cancelled and the method will 312 return ``True``. 313 314 .. method:: cancelled() 315 316 Return ``True`` if the call was successfully cancelled. 317 318 .. method:: running() 319 320 Return ``True`` if the call is currently being executed and cannot be 321 cancelled. 322 323 .. method:: done() 324 325 Return ``True`` if the call was successfully cancelled or finished 326 running. 327 328 .. method:: result(timeout=None) 329 330 Return the value returned by the call. If the call hasn't yet completed 331 then this method will wait up to *timeout* seconds. If the call hasn't 332 completed in *timeout* seconds, then a 333 :exc:`concurrent.futures.TimeoutError` will be raised. *timeout* can be 334 an int or float. If *timeout* is not specified or ``None``, there is no 335 limit to the wait time. 336 337 If the future is cancelled before completing then :exc:`.CancelledError` 338 will be raised. 339 340 If the call raised, this method will raise the same exception. 341 342 .. method:: exception(timeout=None) 343 344 Return the exception raised by the call. If the call hasn't yet 345 completed then this method will wait up to *timeout* seconds. If the 346 call hasn't completed in *timeout* seconds, then a 347 :exc:`concurrent.futures.TimeoutError` will be raised. *timeout* can be 348 an int or float. If *timeout* is not specified or ``None``, there is no 349 limit to the wait time. 350 351 If the future is cancelled before completing then :exc:`.CancelledError` 352 will be raised. 353 354 If the call completed without raising, ``None`` is returned. 355 356 .. method:: add_done_callback(fn) 357 358 Attaches the callable *fn* to the future. *fn* will be called, with the 359 future as its only argument, when the future is cancelled or finishes 360 running. 361 362 Added callables are called in the order that they were added and are 363 always called in a thread belonging to the process that added them. If 364 the callable raises an :exc:`Exception` subclass, it will be logged and 365 ignored. If the callable raises a :exc:`BaseException` subclass, the 366 behavior is undefined. 367 368 If the future has already completed or been cancelled, *fn* will be 369 called immediately. 370 371 The following :class:`Future` methods are meant for use in unit tests and 372 :class:`Executor` implementations. 373 374 .. method:: set_running_or_notify_cancel() 375 376 This method should only be called by :class:`Executor` implementations 377 before executing the work associated with the :class:`Future` and by unit 378 tests. 379 380 If the method returns ``False`` then the :class:`Future` was cancelled, 381 i.e. :meth:`Future.cancel` was called and returned `True`. Any threads 382 waiting on the :class:`Future` completing (i.e. through 383 :func:`as_completed` or :func:`wait`) will be woken up. 384 385 If the method returns ``True`` then the :class:`Future` was not cancelled 386 and has been put in the running state, i.e. calls to 387 :meth:`Future.running` will return `True`. 388 389 This method can only be called once and cannot be called after 390 :meth:`Future.set_result` or :meth:`Future.set_exception` have been 391 called. 392 393 .. method:: set_result(result) 394 395 Sets the result of the work associated with the :class:`Future` to 396 *result*. 397 398 This method should only be used by :class:`Executor` implementations and 399 unit tests. 400 401 .. versionchanged:: 3.8 402 This method raises 403 :exc:`concurrent.futures.InvalidStateError` if the :class:`Future` is 404 already done. 405 406 .. method:: set_exception(exception) 407 408 Sets the result of the work associated with the :class:`Future` to the 409 :class:`Exception` *exception*. 410 411 This method should only be used by :class:`Executor` implementations and 412 unit tests. 413 414 .. versionchanged:: 3.8 415 This method raises 416 :exc:`concurrent.futures.InvalidStateError` if the :class:`Future` is 417 already done. 418 419Module Functions 420---------------- 421 422.. function:: wait(fs, timeout=None, return_when=ALL_COMPLETED) 423 424 Wait for the :class:`Future` instances (possibly created by different 425 :class:`Executor` instances) given by *fs* to complete. Returns a named 426 2-tuple of sets. The first set, named ``done``, contains the futures that 427 completed (finished or cancelled futures) before the wait completed. The 428 second set, named ``not_done``, contains the futures that did not complete 429 (pending or running futures). 430 431 *timeout* can be used to control the maximum number of seconds to wait before 432 returning. *timeout* can be an int or float. If *timeout* is not specified 433 or ``None``, there is no limit to the wait time. 434 435 *return_when* indicates when this function should return. It must be one of 436 the following constants: 437 438 .. tabularcolumns:: |l|L| 439 440 +-----------------------------+----------------------------------------+ 441 | Constant | Description | 442 +=============================+========================================+ 443 | :const:`FIRST_COMPLETED` | The function will return when any | 444 | | future finishes or is cancelled. | 445 +-----------------------------+----------------------------------------+ 446 | :const:`FIRST_EXCEPTION` | The function will return when any | 447 | | future finishes by raising an | 448 | | exception. If no future raises an | 449 | | exception then it is equivalent to | 450 | | :const:`ALL_COMPLETED`. | 451 +-----------------------------+----------------------------------------+ 452 | :const:`ALL_COMPLETED` | The function will return when all | 453 | | futures finish or are cancelled. | 454 +-----------------------------+----------------------------------------+ 455 456.. function:: as_completed(fs, timeout=None) 457 458 Returns an iterator over the :class:`Future` instances (possibly created by 459 different :class:`Executor` instances) given by *fs* that yields futures as 460 they complete (finished or cancelled futures). Any futures given by *fs* that 461 are duplicated will be returned once. Any futures that completed before 462 :func:`as_completed` is called will be yielded first. The returned iterator 463 raises a :exc:`concurrent.futures.TimeoutError` if :meth:`~iterator.__next__` 464 is called and the result isn't available after *timeout* seconds from the 465 original call to :func:`as_completed`. *timeout* can be an int or float. If 466 *timeout* is not specified or ``None``, there is no limit to the wait time. 467 468 469.. seealso:: 470 471 :pep:`3148` -- futures - execute computations asynchronously 472 The proposal which described this feature for inclusion in the Python 473 standard library. 474 475 476Exception classes 477----------------- 478 479.. currentmodule:: concurrent.futures 480 481.. exception:: CancelledError 482 483 Raised when a future is cancelled. 484 485.. exception:: TimeoutError 486 487 Raised when a future operation exceeds the given timeout. 488 489.. exception:: BrokenExecutor 490 491 Derived from :exc:`RuntimeError`, this exception class is raised 492 when an executor is broken for some reason, and cannot be used 493 to submit or execute new tasks. 494 495 .. versionadded:: 3.7 496 497.. exception:: InvalidStateError 498 499 Raised when an operation is performed on a future that is not allowed 500 in the current state. 501 502 .. versionadded:: 3.8 503 504.. currentmodule:: concurrent.futures.thread 505 506.. exception:: BrokenThreadPool 507 508 Derived from :exc:`~concurrent.futures.BrokenExecutor`, this exception 509 class is raised when one of the workers of a :class:`ThreadPoolExecutor` 510 has failed initializing. 511 512 .. versionadded:: 3.7 513 514.. currentmodule:: concurrent.futures.process 515 516.. exception:: BrokenProcessPool 517 518 Derived from :exc:`~concurrent.futures.BrokenExecutor` (formerly 519 :exc:`RuntimeError`), this exception class is raised when one of the 520 workers of a :class:`ProcessPoolExecutor` has terminated in a non-clean 521 fashion (for example, if it was killed from the outside). 522 523 .. versionadded:: 3.3 524