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

17.4.1. Объекты исполнителя¶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(func, *iterables, timeout=None)¶

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

shutdown(wait=True)¶

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

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

Можно избежать явного вызова этого метода, используя оператор with, который завершит работу исполнителя (с ожиданием, как если бы 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, 'src3.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)¶

Подкласс исполнителя, который использует пул из не более чем 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://some-made-up-domain.com/']

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

# Можно использовать инструкцию 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 – это подкласс исполнителя, который использует пул процессов для асинхронного выполнения вызовов. ProcessPoolExecutor использует модуль multiprocessing, что позволяет обойти глобальную блокировку интерпретатора, но также означает, что выполняться и возвращаться могут только сериализуемые объекты.

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

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

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

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

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

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

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

exception(timeout=None)¶

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

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

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

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

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

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

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

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

Константа	Описание
FIRST_COMPLETED	Функция вернёт результат, когда любой future завершится или будет отменён.
FIRST_EXCEPTION	Функция вернёт результат, когда любой future завершится возбуждением исключения. Если ни один future не возбудит исключение, это эквивалентно ALL_COMPLETED.
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 может быть int или float. Если timeout не указан или равен None, то время ожидания не ограничено.

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

exception concurrent.futures.BrokenProcessPool¶

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

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