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

Исходный код: Lib/concurrent/futures/thread.py и Lib/concurrent/futures/process.py

Модуль concurrent.futures предоставляет высокоуровневый интерфейс для асинхронного выполнения вызываемых объектов.

Асинхронное выполнение может быть реализовано с использованием потоков, с помощью ThreadPoolExecutor, или отдельных процессов, с помощью ProcessPoolExecutor. Оба реализуют один и тот же интерфейс, определяемый абстрактным классом Executor.

1. Объекты 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(func, *iterables, timeout=None, chunksize=1)

Эквивалентно map(func, *iterables), за исключением того, что func выполняется асинхронно, и несколько вызовов func могут быть выполнены одновременно. Возвращаемый итератор генерирует исключение concurrent.futures.TimeoutError, если к нему применяется метод __next__(), и результат не доступен через timeout секунд с момента первоначального вызова Executor.map(). timeout может быть целым числом или числом с плавающей точкой. Если timeout не указан или None, время ожидания не ограничено. Если вызов вызывает исключение, это исключение будет поднято при получении его значения из итератора. При использовании ProcessPoolExecutor, этот метод разбивает iterables на несколько частей, которые он отправляет в пул в виде отдельных задач. Размер (приблизительный) этих частей можно указать, установив chunksize в положительное целое число. Для очень длинных итерируемых объектов использование большого значения для chunksize может значительно улучшить производительность по сравнению с размером по умолчанию 1. С ThreadPoolExecutor, chunksize не влияет на результат.

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

shutdown(wait=True)

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

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

Вы можете избежать явного вызова этого метода, если используете оператор 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')

2. 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)

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

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

2.1. Пример ThreadPoolExecutor

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://some-made-up-domain.com/']

# 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)))

3. ProcessPoolExecutor

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

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

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

class concurrent.futures.ProcessPoolExecutor(max_workers=None)

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

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

3.1. Пример ProcessPoolExecutor

import concurrent.futures
import math

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

def is_prime(n):
    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()

4. Объекты 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 секунд, будет поднято исключение concurrent.futures.TimeoutError. timeout может быть целым или дробным числом. Если timeout не указан или None, время ожидания не ограничено.

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

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

exception(timeout=None)

Возвращает исключение, поднятое вызовом. Если вызов ещё не завершён, этот метод будет ждать до timeout секунд. Если вызов не завершится в течение timeout секунд, будет поднято исключение concurrent.futures.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 и модульными тестами.

set_exception(exception)

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

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

5. Модульные функции

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

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

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

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

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

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

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

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

exception concurrent.futures.CancelledError

Поднимается, когда задача отменяется.

exception concurrent.futures.TimeoutError

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

exception concurrent.futures.process.BrokenProcessPool

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

New in version 3.3.