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

Новое в версии 3.2.

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

---

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

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

17.4.1. Объекты исполнителей

classconcurrent.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 может быть int или float. Если timeout не указан или None, время ожидания не ограничено.

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

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

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

shutdown(wait=True)¶

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

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

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

Подкласс 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/']

# Получить одну страницу и вывести 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)))

17.4.3. ProcessPoolExecutor

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

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

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

class concurrent.futures.ProcessPoolExecutor(max_workers=None)¶

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

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

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

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

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

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

exception(timeout=None)¶

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

set_exception(исключение)¶

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

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

17.4.5. Функции модуля

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

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

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

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

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

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

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

Смотрите также

PEP 3148 – futures - выполнение вычислений асинхронно

Предложение, которое описывало эту возможность для включения в стандартную библиотеку Python.

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

exceptionconcurrent.futures.CancelledError¶

Возбуждается, когда future отменен.

exceptionconcurrent.futures.TimeoutError¶

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

exceptionconcurrent.futures.process.BrokenProcessPool¶

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

Новое в версии 3.3.