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, buffersize=None)

Аналогичен map(fn, *iterables), за исключением того, что:

• Элементы iterables собираются сразу, а не по мере необходимости, если только не задан параметр buffersize, ограничивающий количество отправленных задач, результаты которых ещё не были выданы. Если буфер заполнен, перебор iterables приостанавливается, пока из буфера не будет выдан результат.
• fn выполняется асинхронно, и несколько вызовов fn могут выполняться параллельно.

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

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

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

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

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

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)
future = executor.submit(wait_on_future)
# Note: calling future.result() would also cause a deadlock because
# the single worker thread is already waiting for 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)))

InterpreterPoolExecutor

Класс InterpreterPoolExecutor использует пул интерпретаторов для асинхронного выполнения вызовов. Это подкласс ThreadPoolExecutor, то есть каждый рабочий интерпретатор работает в собственном потоке. Отличие заключается в том, что у каждого рабочего потока есть собственный интерпретатор, и каждая задача выполняется с использованием этого интерпретатора.

Главное преимущество использования интерпретаторов вместо одних только потоков — настоящая параллельная работа на нескольких ядрах. У каждого интерпретатора есть собственная глобальная блокировка интерпретатора, поэтому код, выполняющийся в одном интерпретаторе, может работать на одном ядре процессора, а код в другом интерпретаторе — беспрепятственно выполняться на другом ядре.

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

Интерпретатор каждого рабочего потока изолирован от остальных интерпретаторов. «Изолирован» означает, что у каждого интерпретатора есть собственное состояние среды выполнения и он функционирует полностью независимо. Например, если перенаправить sys.stdout в одном интерпретаторе, это не приведёт к автоматическому перенаправлению в других интерпретаторах. Если импортировать модуль в одном интерпретаторе, в остальных он автоматически импортирован не будет. Модуль потребуется отдельно импортировать в том интерпретаторе, где он нужен. Более того, каждый модуль, импортированный в интерпретаторе, является совершенно отдельным объектом по сравнению с тем же модулем в другом интерпретаторе. Это относится, в том числе, к sys, builtins и даже __main__.

Изоляция означает, что изменяемый объект или другие данные не могут одновременно использоваться несколькими интерпретаторами. Фактически это означает, что интерпретаторы не могут совместно использовать такие объекты или данные. Вместо этого у каждого интерпретатора должна быть собственная копия, а синхронизировать изменения между копиями придётся вручную. На неизменяемые объекты и данные, такие как встроенные одиночные объекты, строки и кортежи из неизменяемых объектов, эти ограничения не распространяются.

Эффективнее всего обмениваться данными и синхронизировать интерпретаторы с помощью специализированных средств, например предлагаемых в PEP 734. Менее эффективный вариант — сериализовать данные с помощью pickle, а затем передать байты через общий socket или pipe.

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

Подкласс ThreadPoolExecutor, который асинхронно выполняет вызовы с помощью пула, содержащего не более max_workers потоков. Каждый поток выполняет задачи в собственном интерпретаторе. Рабочие интерпретаторы изолированы друг от друга: у каждого есть собственное состояние среды выполнения, и они не могут совместно использовать изменяемые объекты или другие данные. У каждого интерпретатора есть собственная глобальная блокировка интерпретатора, поэтому код, выполняемый с помощью этого исполнителя, действительно может работать параллельно на нескольких ядрах.

Необязательные аргументы initializer и initargs имеют тот же смысл, что и для ThreadPoolExecutor: функция-инициализатор запускается при создании каждого рабочего потока, но в данном случае она выполняется в интерпретаторе рабочего потока. При передаче функции-инициализатора и аргументов в интерпретатор рабочего потока исполнитель сериализует их с помощью pickle.

Примечание

Исполнитель может заменить неперехваченные исключения из initializer на ExecutionFailed.

Здесь действуют и другие ограничения родительского класса ThreadPoolExecutor.

submit() и map() работают обычным образом, за исключением того, что рабочий поток сериализует вызываемый объект и аргументы с помощью pickle при передаче их своему интерпретатору. Аналогичным образом рабочий поток сериализует возвращаемое значение при отправке его обратно.

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

ProcessPoolExecutor

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

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

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

Обратите внимание: при использовании submit() и map() с ProcessPoolExecutor действуют ограничения, установленные для функций и аргументов, которые должны быть сериализуемыми, как и для multiprocessing.Process. Не следует рассчитывать на работу функций, определённых в 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 по умолчанию будет использоваться метод запуска «spawn». Эта возможность несовместима с методом запуска «fork».

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

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

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

Изменено в версии 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().

Изменено в версии 3.14: Метод запуска процессов по умолчанию (см. раздел Контексты и методы запуска) изменён: теперь это не fork. Если для ProcessPoolExecutor требуется метод запуска fork, его необходимо явно передать как mp_context=multiprocessing.get_context("fork").

Изменено в версии 3.14.7: Исправлена взаимоблокировка (gh-115634), из-за которой исполнитель мог зависнуть после завершения рабочего процесса при достижении лимита max_tasks_per_child, если в очереди ещё оставались задачи.

terminate_workers()

Пытается немедленно завершить все работающие рабочие процессы, вызвав для каждого из них Process.terminate. Внутри также вызывает Executor.shutdown(), чтобы обеспечить освобождение всех остальных ресурсов, связанных с исполнителем.

После вызова этого метода вызывающей стороне больше не следует отправлять задачи исполнителю.

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

kill_workers()

Пытается немедленно принудительно завершить все работающие рабочие процессы, вызвав для каждого из них Process.kill. Внутри также вызывает Executor.shutdown(), чтобы обеспечить освобождение всех остальных ресурсов, связанных с исполнителем.

После вызова этого метода вызывающей стороне больше не следует отправлять задачи исполнителю.

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

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)

Задаёт результат работы, связанной с 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 (возможно, созданных разными экземплярами 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.interpreter.BrokenInterpreterPool

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

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

exception concurrent.futures.process.BrokenProcessPool

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

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