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)

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

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

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

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

Необязательно явно вызывать этот метод, если использовать исполнитель в качестве менеджера контекста с помощью инструкции 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 рабочих потоков.

Изменено в версии 3.13: Значение max_workers по умолчанию изменено на min(32, (os.process_cpu_count() or 1) + 4).

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://nonexistent-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, приведёт к взаимной блокировке.

Обратите внимание, что ограничения на сериализуемость функций и аргументов, установленные в multiprocessing.Process, действуют при использовании submit() и map() с ProcessPoolExecutor. Не следует рассчитывать на работу функции, определённой в REPL, или лямбда-функции.

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

Подкласс Executor, который асинхронно выполняет вызовы с помощью пула, содержащего не более max_workers процессов. Если max_workers равен None или не задан, по умолчанию используется значение os.process_cpu_count(). Если 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. Ранее поведение не было определено, однако операции с исполнителем или его объектами future часто зависали или приводили к взаимной блокировке.

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

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

Примечание

В Python 3.14 метод запуска по умолчанию multiprocessing (см. раздел Контексты и методы запуска) изменится с fork на другой. В коде, которому для ProcessPoolExecutor требуется использовать fork, это следует указать явно, передав параметр mp_context=multiprocessing.get_context("fork").

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

Изменено в версии 3.12: В системах POSIX, если приложение использует несколько потоков и контекст multiprocessing использует метод запуска "fork", функция os.fork(), вызываемая внутри для создания рабочих процессов, может возбудить исключение DeprecationWarning. Передайте mp_context, настроенный на использование другого метода запуска. Дополнительные сведения см. в документации по os.fork().

Изменено в версии 3.13: По умолчанию для max_workers используется os.process_cpu_count() вместо os.cpu_count().

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, время ожидания не ограничено.

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

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

exception(timeout=None)

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

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

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

add_done_callback(fn)

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

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

Если объект future уже завершён или отменён, 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)

Задаёт значение result в качестве результата работы, связанной с Future.

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

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

set_exception(exception)

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

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

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

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

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

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

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

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

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

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

Возвращает итератор по экземплярам Future (возможно, созданным разными экземплярами Executor), переданным в fs. Итератор выдаёт объекты Future по мере их завершения (выполненные или отменённые объекты Future). Повторяющиеся объекты Future, переданные в fs, возвращаются только один раз. Объекты Future, завершившиеся до вызова 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.