concurrent.futures — Запуск параллельных задач

Объекты Executor

class concurrent.futures.Executor

Абстрактный класс, предоставляющий методы для асинхронного выполнения вызовов. Его не следует использовать напрямую — только через конкретные подклассы.

submit(fn, /, *args, **kwargs)

Планирует выполнение вызываемого объекта fn как fn(*args, **kwargs) и возвращает объект Future, представляющий выполнение вызываемого объекта.

with ThreadPoolExecutor(max_workers=1) as executor:
    future = executor.submit(pow, 323, 1235)
    print(future.result())

map(fn, *iterables, timeout=None, chunksize=1)

Похоже на map(fn, *iterables), за исключением следующего:

• объекты iterables собираются немедленно, а не лениво;
• fn выполняется асинхронно, и несколько вызовов fn могут выполняться одновременно.

Возвращённый итератор вызывает исключение TimeoutError, если вызывается __next__(), а результат недоступен по истечении timeout секунд с момента первоначального вызова Executor.map(). Значение timeout может быть целым числом или числом с плавающей точкой. Если timeout не указан или None, время ожидания не ограничено.

Если вызов fn вызывает исключение, оно будет вызвано при получении соответствующего значения из итератора.

При использовании ProcessPoolExecutor этот метод разбивает объекты iterables на несколько частей и отправляет их пулу как отдельные задачи. (Приблизительный) размер этих частей можно задать, установив chunksize в положительное целое число. Для очень длинных итерируемых объектов большое значение chunksize может значительно повысить производительность по сравнению со значением по умолчанию, равным 1. В случае ThreadPoolExecutor параметр chunksize не влияет на результат.

Изменено в версии 3.5: Добавлен аргумент chunksize.

shutdown(wait=True, *, cancel_futures=False)

Сообщает исполнителю, что следует освободить все используемые ресурсы после завершения выполнения ожидающих задач. Вызовы Executor.submit() и Executor.map(), выполненные после завершения работы, вызовут исключение RuntimeError.

Если wait имеет значение True, этот метод не вернёт управление, пока не завершится выполнение всех ожидающих задач и не будут освобождены ресурсы, связанные с исполнителем. Если wait имеет значение False, этот метод вернёт управление немедленно, а ресурсы, связанные с исполнителем, будут освобождены после завершения всех ожидающих задач. Независимо от значения wait, программа Python не завершится, пока не закончат выполняться все ожидающие задачи.

Если cancel_futures имеет значение True, этот метод отменит все ожидающие задачи, выполнение которых исполнитель ещё не начал. Выполненные или выполняющиеся задачи не будут отменены независимо от значения cancel_futures.

Если значения cancel_futures и wait равны True, все задачи, выполнение которых исполнитель уже начал, завершатся до возврата этого метода. Остальные задачи будут отменены.

Можно не вызывать этот метод явно, если использовать инструкцию with: она завершит работу Executor (ожидая завершения, как если бы Executor.shutdown() был вызван со значением wait True):

import shutil
with ThreadPoolExecutor(max_workers=4) as e:
    e.submit(shutil.copy, 'src1.txt', 'dest1.txt')
    e.submit(shutil.copy, 'src2.txt', 'dest2.txt')
    e.submit(shutil.copy, 'src3.txt', 'dest3.txt')
    e.submit(shutil.copy, 'src4.txt', 'dest4.txt')

Изменено в версии 3.9: Добавлен параметр cancel_futures.

ThreadPoolExecutor

ThreadPoolExecutor — подкласс Executor, использующий пул потоков для асинхронного выполнения вызовов.

Возможны взаимоблокировки, если вызываемый объект, связанный с Future, ожидает результатов другого объекта Future. Например:

import time
def wait_on_b():
    time.sleep(5)
    print(b.result())  # b will never complete because it is waiting on a.
    return 5

def wait_on_a():
    time.sleep(5)
    print(a.result())  # a will never complete because it is waiting on b.
    return 6


executor = ThreadPoolExecutor(max_workers=2)
a = executor.submit(wait_on_b)
b = executor.submit(wait_on_a)

А также:

def wait_on_future():
    f = executor.submit(pow, 5, 2)
    # This will never complete because there is only one worker thread and
    # it is executing this function.
    print(f.result())

executor = ThreadPoolExecutor(max_workers=1)
executor.submit(wait_on_future)

class concurrent.futures.ThreadPoolExecutor(max_workers=None, thread_name_prefix='', initializer=None, initargs=())

Подкласс Executor, использующий пул не более чем из max_workers потоков для асинхронного выполнения вызовов.

Все потоки, поставленные в очередь для ThreadPoolExecutor, будут объединены до завершения работы интерпретатора. Обратите внимание: обработчик завершения, выполняющий это действие, запускается до всех обработчиков завершения, добавленных с помощью atexit. Это означает, что исключения в основном потоке необходимо перехватывать и обрабатывать, чтобы подать потокам сигнал о необходимости корректно завершить работу. По этой причине не рекомендуется использовать ThreadPoolExecutor для длительных задач.

initializer — необязательный вызываемый объект, который вызывается при запуске каждого рабочего потока; initargs — кортеж аргументов, передаваемых инициализатору. Если initializer вызовет исключение, все ожидающие задачи вызовут исключение BrokenThreadPool, как и любая попытка отправить в пул дополнительные задачи.

Изменено в версии 3.5: Если max_workers равен None или не задан, по умолчанию он будет равен числу процессоров на машине, умноженному на 5. Предполагается, что ThreadPoolExecutor часто используется для совмещения операций ввода-вывода, а не вычислений на процессоре, поэтому число рабочих потоков должно быть больше, чем число рабочих процессов для ProcessPoolExecutor.

Изменено в версии 3.6: Добавлен параметр thread_name_prefix, позволяющий задавать имена объектов threading.Thread для рабочих потоков, создаваемых пулом, чтобы упростить отладку.

Изменено в версии 3.7: Добавлены аргументы initializer и initargs.

Изменено в версии 3.8: Значение max_workers по умолчанию изменено на min(32, os.cpu_count() + 4). Это значение по умолчанию обеспечивает наличие как минимум 5 рабочих потоков для задач, ограниченных операциями ввода-вывода. Для задач, ограниченных вычислениями на процессоре и освобождающих GIL, используется не более 32 ядер процессора. Кроме того, оно позволяет избежать неявного использования чрезмерно большого количества ресурсов на машинах с большим числом ядер.

Теперь ThreadPoolExecutor повторно использует простаивающие рабочие потоки, прежде чем создавать потоки в количестве max_workers.

import concurrent.futures
import urllib.request

URLS = ['http://www.foxnews.com/',
        'http://www.cnn.com/',
        'http://europe.wsj.com/',
        'http://www.bbc.co.uk/',
        'http://nonexistant-subdomain.python.org/']

# Retrieve a single page and report the URL and contents
def load_url(url, timeout):
    with urllib.request.urlopen(url, timeout=timeout) as conn:
        return conn.read()

# We can use a with statement to ensure threads are cleaned up promptly
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    # Start the load operations and mark each future with its URL
    future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}
    for future in concurrent.futures.as_completed(future_to_url):
        url = future_to_url[future]
        try:
            data = future.result()
        except Exception as exc:
            print('%r generated an exception: %s' % (url, exc))
        else:
            print('%r page is %d bytes' % (url, len(data)))

ProcessPoolExecutor

Класс ProcessPoolExecutor — подкласс Executor, использующий пул процессов для асинхронного выполнения вызовов. ProcessPoolExecutor использует модуль multiprocessing, который позволяет обойти глобальную блокировку интерпретатора, но также означает, что выполняться и возвращаться могут только объекты, допускающие сериализацию с помощью pickle.

Модуль __main__ должен быть доступен для импорта рабочими подпроцессами. Это означает, что ProcessPoolExecutor не будет работать в интерактивном интерпретаторе.

Вызов методов Executor или Future из вызываемого объекта, переданного в ProcessPoolExecutor, приведёт к взаимоблокировке.

class concurrent.futures.ProcessPoolExecutor(max_workers=None, mp_context=None, initializer=None, initargs=(), max_tasks_per_child=None)

Подкласс Executor, асинхронно выполняющий вызовы с использованием пула не более чем из max_workers процессов. Если max_workers равен None или не задан, по умолчанию он будет равен числу процессоров на машине. Если max_workers меньше или равен 0, будет вызвано исключение ValueError. В Windows значение max_workers должно быть меньше или равно 61. В противном случае будет вызвано исключение ValueError. Если max_workers равен None, выбранное по умолчанию значение будет не больше 61, даже если доступно больше процессоров. Параметр mp_context может быть контекстом multiprocessing или значением None. Он будет использоваться для запуска рабочих процессов. Если mp_context равен None или не задан, используется контекст multiprocessing по умолчанию.

initializer — необязательный вызываемый объект, который вызывается при запуске каждого рабочего процесса; initargs — кортеж аргументов, передаваемых инициализатору. Если initializer вызовет исключение, все ожидающие задачи вызовут исключение BrokenProcessPool, как и любая попытка отправить в пул дополнительные задачи.

max_tasks_per_child — необязательный аргумент, задающий максимальное количество задач, которые может выполнить один процесс, прежде чем завершить работу и быть заменённым новым рабочим процессом. По умолчанию max_tasks_per_child равен None, то есть рабочие процессы работают столько же, сколько пул. Если задано максимальное количество задач, при отсутствии параметра mp_context по умолчанию используется метод запуска multiprocessing «spawn». Эта возможность несовместима с методом запуска «fork».

Изменено в версии 3.3: Если один из рабочих процессов неожиданно завершится, теперь вызывается исключение BrokenProcessPool. Ранее поведение не было определено, однако операции с исполнителем или его задачами часто зависали или приводили к взаимоблокировке.

Изменено в версии 3.7: Добавлен аргумент mp_context, позволяющий пользователям управлять start_method для рабочих процессов, создаваемых пулом.

Добавлены аргументы initializer и initargs.

Изменено в версии 3.11: Добавлен аргумент max_tasks_per_child, позволяющий пользователям управлять временем жизни рабочих процессов в пуле.

import concurrent.futures
import math

PRIMES = [
    112272535095293,
    112582705942171,
    112272535095293,
    115280095190773,
    115797848077099,
    1099726899285419]

def is_prime(n):
    if n < 2:
        return False
    if n == 2:
        return True
    if n % 2 == 0:
        return False

    sqrt_n = int(math.floor(math.sqrt(n)))
    for i in range(3, sqrt_n + 1, 2):
        if n % i == 0:
            return False
    return True

def main():
    with concurrent.futures.ProcessPoolExecutor() as executor:
        for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)):
            print('%d is prime: %s' % (number, prime))

if __name__ == '__main__':
    main()

Объекты Future

Класс Future инкапсулирует асинхронное выполнение вызываемого объекта. Экземпляры Future создаются методом Executor.submit().

class concurrent.futures.Future

Инкапсулирует асинхронное выполнение вызываемого объекта. Экземпляры Future создаются методом Executor.submit(); создавать их напрямую не следует, за исключением тестирования.

cancel()

Пытается отменить вызов. Если вызов выполняется в данный момент или уже завершился и не может быть отменён, метод возвращает False; в противном случае вызов отменяется, а метод возвращает True.

cancelled()

Возвращает True, если вызов был успешно отменён.

running()

Возвращает True, если вызов выполняется в данный момент и не может быть отменён.

done()

Возвращает True, если вызов был успешно отменён или его выполнение завершилось.

result(timeout=None)

Возвращает значение, возвращённое вызовом. Если вызов ещё не завершился, этот метод будет ожидать не более timeout секунд. Если вызов не завершится за timeout секунд, будет вызвано исключение TimeoutError. Значение timeout может быть целым числом или числом с плавающей точкой. Если timeout не указан или None, время ожидания не ограничено.

Если задача отменена до завершения, будет вызвано исключение CancelledError.

Если вызов вызвал исключение, этот метод вызовет то же исключение.

exception(timeout=None)

Возвращает исключение, вызванное вызовом. Если вызов ещё не завершился, этот метод будет ожидать не более timeout секунд. Если вызов не завершится за timeout секунд, будет вызвано исключение TimeoutError. Значение timeout может быть целым числом или числом с плавающей точкой. Если timeout не указан или None, время ожидания не ограничено.

Если задача отменена до завершения, будет вызвано исключение CancelledError.

Если вызов завершился без исключения, возвращается None.

add_done_callback(fn)

Привязывает вызываемый объект fn к задаче. fn будет вызван с задачей в качестве единственного аргумента, когда задача будет отменена или завершит выполнение.

Добавленные вызываемые объекты вызываются в порядке добавления и всегда в потоке процесса, который их добавил. Если вызываемый объект вызовет исключение, являющееся подклассом Exception, оно будет записано в журнал и проигнорировано. Если вызываемый объект вызовет исключение, являющееся подклассом BaseException, поведение не определено.

Если задача уже завершилась или была отменена, fn будет вызван немедленно.

Следующие методы Future предназначены для использования в модульных тестах и реализациях Executor.

set_running_or_notify_cancel()

Этот метод следует вызывать только реализациям Executor перед выполнением работы, связанной с Future, а также модульным тестам.

Если метод возвращает False, значит, Future была отменена, то есть был вызван Future.cancel() и вернул True. Все потоки, ожидающие завершения Future (например, через as_completed() или wait()), будут разбужены.

Если метод возвращает True, значит, Future не была отменена и переведена в состояние выполнения, то есть вызовы Future.running() будут возвращать True.

Этот метод можно вызвать только один раз; его нельзя вызывать после вызова Future.set_result() или Future.set_exception().

set_result(result)

Устанавливает значение результата работы, связанной с Future, равным result.

Этот метод следует использовать только реализациям Executor и модульным тестам.

Изменено в версии 3.8: Этот метод вызывает исключение concurrent.futures.InvalidStateError, если Future уже завершена.

set_exception(exception)

Устанавливает результат работы, связанной с Future, в виде исключения Exception exception.

Этот метод следует использовать только реализациям Executor и модульным тестам.

Изменено в версии 3.8: Этот метод вызывает исключение concurrent.futures.InvalidStateError, если Future уже завершена.

Функции модуля

concurrent.futures.wait(fs, timeout=None, return_when=ALL_COMPLETED)

Ожидает завершения экземпляров Future, заданных в fs (возможно, созданных разными экземплярами Executor). Повторяющиеся задачи из fs удаляются и возвращаются только один раз. Возвращает именованный кортеж из двух множеств. Первое множество с именем done содержит задачи, завершившиеся (выполненные или отменённые) до окончания ожидания. Второе множество с именем not_done содержит задачи, которые не завершились (ожидающие или выполняющиеся).

Параметр timeout позволяет задать максимальное количество секунд ожидания перед возвратом. Значение timeout может быть целым числом или числом с плавающей точкой. Если timeout не указан или None, время ожидания не ограничено.

Параметр return_when указывает, когда функция должна вернуть управление. Он должен принимать одно из следующих константных значений:

Константа	Описание
concurrent.futures.FIRST_COMPLETED	Функция вернёт управление, когда любая задача завершится или будет отменена.
concurrent.futures.FIRST_EXCEPTION	Функция вернёт управление, когда любая задача завершится с исключением. Если ни одна задача не вызовет исключение, это равносильно ALL_COMPLETED.
concurrent.futures.ALL_COMPLETED	Функция вернёт управление, когда все задачи завершатся или будут отменены.

concurrent.futures.as_completed(fs, timeout=None)

Возвращает итератор по экземплярам Future, заданным в fs (возможно, созданным разными экземплярами Executor), который выдаёт задачи по мере их завершения (выполненные или отменённые задачи). Повторяющиеся задачи из fs будут возвращены один раз. Задачи, завершившиеся до вызова as_completed(), будут выданы первыми. Возвращённый итератор вызывает исключение TimeoutError, если вызывается __next__(), а результат недоступен по истечении timeout секунд с момента первоначального вызова as_completed(). Значение timeout может быть целым числом или числом с плавающей точкой. Если timeout не указан или None, время ожидания не ограничено.

Классы исключений

exception concurrent.futures.CancelledError

Возникает, когда объект Future отменён.

exception concurrent.futures.TimeoutError

Устаревший псевдоним TimeoutError; возникает, когда операция с объектом Future превышает заданное время ожидания.

Изменено в версии 3.11: Этот класс стал псевдонимом TimeoutError.

exception concurrent.futures.BrokenExecutor

Этот класс исключений, производный от RuntimeError, возникает, когда исполнитель по какой-либо причине неисправен и не может использоваться для отправки или выполнения новых задач.

Добавлено в версии 3.7.

exception concurrent.futures.InvalidStateError

Возникает, когда над объектом Future выполняется операция, недопустимая в его текущем состоянии.

Добавлено в версии 3.8.

exception concurrent.futures.thread.BrokenThreadPool

Этот класс исключений, производный от BrokenExecutor, возникает, если не удалось инициализировать один из рабочих потоков ThreadPoolExecutor.

Добавлено в версии 3.7.

exception concurrent.futures.process.BrokenProcessPool

Этот класс исключений, производный от BrokenExecutor (ранее — от RuntimeError), возникает, когда один из рабочих процессов ProcessPoolExecutor завершился некорректно (например, был принудительно остановлен извне).

Добавлено в версии 3.3.