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

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

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

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

17.4.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), но с отличиями:

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

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

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

При использовании 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')

17.4.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, thread_name_prefix='')

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

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

Новое в версии 3.6: Добавлен аргумент thread_name_prefix, позволяющий пользователям управлять именами потоков threading.Thread, созданных пулом, для более удобной отладки.

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

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

17.4.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()

17.4.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 и модульными тестами.

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

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

exception concurrent.futures.CancelledError

Выбрасывается, когда задача отменена.

exception concurrent.futures.TimeoutError

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

exception concurrent.futures.process.BrokenProcessPool

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

Введено в версии 3.3.