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 рабочих потоков.

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. Ранее поведение не было определено, однако операции с исполнителем или его задачами future часто зависали или приводили к взаимоблокировке.

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

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

Примечание

Метод запуска multiprocessing по умолчанию (см. раздел Контексты и методы запуска) в Python 3.14 изменится: вместо 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().

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 в качестве единственного аргумента, когда задача будет отменена или завершится.

Добавленные вызываемые объекты вызываются в порядке добавления и всегда в потоке, принадлежащем процессу, в котором они были добавлены. Если вызываемый объект вызывает исключение — подкласс 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 в 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.