На этом шаге мы рассмотрим использование класса ProcessPoolExecutor из библиотеки concurrent.futures.

Задача

    У вас есть программа, которая выполняет много работы, нагружающей CPU, и вы хотите заставить ее работать быстрее путем использования нескольких CPU.

Решение

    Библиотека concurrent.futures предоставляет класс ProcessPoolExecutor, который может быть использован для выполнения тяжелых вычислительных задач в отдельно запущенных экземплярах интерпретатора Python.

    Для использования этого класса вам сначала нужно иметь, что вычислять. Давайте проиллюстрируем это с помощью простого практического примера.

    Предположим, что у вас есть каталог со сжатыми gzip логами веб-сервера Apache:

logs/
    20120701.log.gz
    20120702.log.gz
    20120703.log.gz
    20120704.log.gz
    20120705.log.gz
    20120706.log.gz
    . . .

    Предположим также, что каждый файл содержит такие строки:

124.115.6.12 - - [10/Jul/2012:00:18:50 -0500] "GET /robots.txt ..." 200 71
210.212.209.67 - [10/Jul/2012:00:18:51 -0500] "GET /ply/ ..." 200 11875
210.212.209.67 - [10/Jul/2012:00:18:51 -0500] "GET /favicon.ico ..." 404 369
61.135.216.105 - [10/Jul/2012:00:20:04 -0500] "GET /blog/atom.xml ..." 304 762
. . .

    Вот простой скрипт, который берет эти данные и находит все хосты, которые обращались к файлу robots.txt:

# findrobots.py

import gzip
import io
import glob


def find_robots(filename):
    '''
    Находит в одном лог-файле все хосты, которые обращались к robots.txt
    '''
    robots = set()
    with gzip.open(filename) as f:
        for line in io.TextIOWrapper(f, encoding='ascii'):
            fields = line.split()
            if fields[6] == '/robots.txt':
                robots.add(fields[0])
    return robots


def find_all_robots(logdir):
    '''
    Находит все хосты во всех файлах
    '''
    files = glob.glob(logdir + "/*.log.gz")
    all_robots = set()
    for robots in map(find_robots, files):
        all_robots.update(robots)
    return all_robots


if __name__ == '__main__':
    import time

    start = time.time()
    robots = find_all_robots("logs")
    end = time.time()
    for ipaddr in robots:
        print(ipaddr)
    print('Took {:f} seconds'.format(end - start))

    Приведенная выше программа написана в популярном стиле map-reduce. Функция find_robots() применяется к коллекции имен файлов, и результаты объединяются в один результат (множество all_robots в функции find_all_robots()).

    Теперь предположим, что вам нужно изменить эту программу таким образом, чтобы использовать несколько CPU. Оказывается, это довольно просто - достаточно заменить операцию map() на похожую операцию из библиотеки concurrent.futures, выполняемую пулом процессов. Вот слегка модифицированная версия:

# findrobots.py

import gzip
import io
import glob
from concurrent import futures


def find_robots(filename):
    '''
    Находит в одном лог-файле все хосты, которые обращались к robots.txt
    '''
    robots = set()
    with gzip.open(filename) as f:
        for line in io.TextIOWrapper(f, encoding='ascii'):
            fields = line.split()
            if fields[6] == '/robots.txt':
                robots.add(fields[0])
    return robots


def find_all_robots(logdir):
    '''
    Находит все хосты во всех файлах
    '''
    files = glob.glob(logdir + "/*.log.gz")
    all_robots = set()
    with futures.ProcessPoolExecutor() as pool:
        for robots in pool.map(find_robots, files):
            all_robots.update(robots)
    return all_robots


if __name__ == '__main__':
    import time

    start = time.time()
    robots = find_all_robots("logs")
    end = time.time()
    for ipaddr in robots:
        print(ipaddr)
    print('Took {:f} seconds'.format(end - start))

    Теперь скрипт производит тот же самый результат, но работает в 3,5 раза быстрее на четырехъядерном компьютере. Реальная производительность будет варьироваться в зависимости от количества CPU, доступных на вашем компьютере.

Обсуждение

    Обычно ProcessPoolExecutor используют так: from concurrent.futures import ProcessPoolExecutor

with ProcessPoolExecutor() as pool:
    . . .
    do work in parallel using pool
    . . .

    "Под капотом" ProcessPoolExecutor создает N независимо работающих интепретаторов Python, где N - это количество доступных обнаруженных в системе CPU. Вы можете изменить количество создаваемых процессов, предоставив необязательный аргумент (ProcessPoolExecutor(N)). Пул работает до тех пор, пока не будет выполнена последняя инструкция в блоке with, после чего пул процессов завершается. Однако программа будет ждать, пока вся отправленная работа не будет сделана.

    Работа, которая отправляется в пул, должна быть определена в форме функции. Есть два метода отправки. Если вы пытаетесь распараллелить генератор списка или операцию map(), используйте pool.map():

# Функция, которая делает много работы
def work(x):
    . . .
    return result

# Непараллельный код
results = map(work, data)

# Параллельная реализация
with ProcessPoolExecutor() as pool:
    results = pool.map(work, data)

    Или же вы можете вручную отправлять единичные задачи, используя метод pool.submit():

#  Какая-то функция 
def work(x):
    .  .  .
    return result

with ProcessPoolExecutor() as pool:
    .  .  .
    #  Пример отправки работы в пул 
    future_result = pool.submit(work, arg)
    #  Получение результата (блокирует до завершения) 
    r = future_result.result()
    .  .  .

    Если вы отправляете работу вручную, результатом будет экземпляр Future. Чтобы получить реальный результат, вы вызываете его метод result(). Это вызывает блокировку на время, пока результат не посчитается и не будет возвращен пулом.

    Вместо блокировки вы можете сделать так, чтобы по завершении работы выполнялась функция обратного вызова. Например:

def when_done(r):
    print('Got:', r.result())

with ProcessPoolExecutor() as pool:
    future_result = pool.submit(work, arg) 
    future_result.add_done_callback(when_done)

    Предоставленная пользователем функция обратного вызова получает экземпляр Future, который должен быть использован для получения реального результата (путем вызова его метода result()).

    Хотя пулы процессов могут быть простыми в использовании, есть несколько важных вопросов, которые нужно держать в голове при написании крупных программ. Приведем их без какого-либо упорядочивания:

• этот прием распараллеливания хорошо работает только для задач, которые легко раскладываются на независимые части;
  
  
• работа должна отправляться в форме простых функций. Параллельное выполнение методов экземпляров, замыканий или других конструкций не поддерживается;
  
  
• аргументы функций и возвращаемые значения должны быть совместимы с pickle. Работа выполняется в отдельном интепретаторе при использовании межпроцессной коммуникации. Так что данные, которыми обмениваются интерпретаторы, должны сериализоваться;
  
  
• функции, отправляемые на выполнение, не должны поддерживать постоянное состояние или иметь побочные эффекты. За исключением простых вещей типа логирования, у вас нет никакого контроля над поведением про- цессов-потомков после их запуска. Так что для сохранения рассудка лучше поддерживать простоту программы и выполнять работу в чистых функциях, которые не изменяют свое окружение;
  
  
• пулы процессов в Unix создаются с помощью системного вызова fork(). Он создает клон интепретатора Python, включая все состояние программы на момент копирования. В Windows запускается независимая копия интепретатора, которая не клонирует состояние. Реального разветвления процессов не произойдет до первого вызова метода pool.map() или pool.submit();
  
  
• нужно с великой осторожностью объединять пулы процессов с программами, которые используют потоки. В частности, вы, вероятно, должны создавать и запускать пул процессов до создания любых потоков (то есть создавать пул в главном потоке на старте программы).
  
  

    На следующем шаге мы поговорим о глобальной блокировке интерпретатора (GIL).

Предыдущий шаг Содержание