concurrent.futures – Запуск параллельных задач¶concurrent.futures – Launching parallel tasks

Объекты исполнителя¶Executor Objects

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), но:

• Итерируемые объекты собираются немедленно, а не лениво;

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

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

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

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

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

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

Сообщает исполнителю, что он должен освободить все используемые ресурсы, когда текущие ожидающие future завершат выполнение. Вызовы Executor.submit() и Executor.map() после shutdown приведут к возбуждению 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 никогда не завершится, потому что он ожидает a.
    return 5

def wait_on_a():
    time.sleep(5)
    print(a.result())  # a никогда не завершится, потому что он ожидает 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)
    # Это никогда не завершится, потому что есть только один рабочий поток и
    # он выполняет эту функцию.
    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 рабочих для задач, связанных с вводом-выводом. Оно использует не более 32 ядер ЦП для задач, связанных с ЦП и отпускающих GIL. И оно позволяет избежать неявного использования очень больших ресурсов на многоядерных машинах.

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/']

# Получить одну страницу и вывести URL и содержимое
def load_url(url, timeout):
    with urllib.request.urlopen(url, timeout=timeout) as conn:
        return conn.read()

# Можно использовать инструкцию with, чтобы гарантировать своевременную очистку потоков
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    # Запустить операции загрузки и пометить каждый future его 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, что позволяет обойти Глобальная блокировка интерпретатора, но также означает, что только picklable-объекты могут быть выполнены и возвращены.

Модуль __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, что означает, что рабочие процессы будут жить столько же, сколько и пул. Если задан максимум, по умолчанию будет использоваться метод запуска "spawn" multiprocessing при отсутствии параметра mp_context. Эта возможность несовместима с методом запуска "fork".

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

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

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

Примечание

Метод запуска по умолчанию multiprocessing (см. Контексты и методы запуска) изменится с fork в Python 3.14. Код, которому требуется fork для его ProcessPoolExecutor, должен явно указать это, передав параметр 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 Objects

Класс 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 исключение.

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

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

Функции модуля¶Module Functions

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

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

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

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

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

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 classes

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.