Назначение многопроцессорной обработки.Pool.apply и многопроцессорной обработки.Pool.apply_async

Question

См. Пример и результат выполнения ниже:

#!/usr/bin/env python3.4
from multiprocessing import Pool
import time
import os

def initializer():
    print("In initializer pid is {} ppid is {}".format(os.getpid(),os.getppid()))

def f(x):
    print("In f pid is {} ppid is {}".format(os.getpid(),os.getppid()))
    return x*x

if __name__ == '__main__':
    print("In main pid is {} ppid is {}".format(os.getpid(), os.getppid()))
    with Pool(processes=4, initializer=initializer) as pool:  # start 4 worker processes
        result = pool.apply(f, (10,)) # evaluate "f(10)" in a single process
        print(result)

        #result = pool.apply_async(f, (10,)) # evaluate "f(10)" in a single process
        #print(result.get())

Дает:

$ ./pooleg.py
In main pid is 22783 ppid is 19542
In initializer pid is 22784 ppid is 22783
In initializer pid is 22785 ppid is 22783
In initializer pid is 22787 ppid is 22783
In f pid is 22784 ppid is 22783
In initializer pid is 22786 ppid is 22783
100

Как видно из вывода: 4 процесса были созданы, но только один из них фактически выполнил работу(называется f).

Вопрос: Зачем мне создавать пул из> 1 рабочих и звонить apply(), когда работа f выполняется только одним процессом?И то же самое для apply_async(), потому что в этом случае работу также выполняет только один работник.

Я не понимаю, в каких случаях эти функции полезны.

Answer 1

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

---

Вы бы использовали Pool.apply_async вместо одной из версий Pool.map, где вам требуется более точный контроль над отдельными задачами, которые вы хотите распределить.

Версии Pool.map принимают итерацию и разделяют их на задачи, где каждая задача имеет целевую функцию с одинаковым (сопоставленным). Pool.apply_async обычно не вызывается только один раз с пулом> 1 работника. Поскольку он асинхронный, вы можете перебирать предварительно связанные задачи и отправлять их нескольким рабочие процессы, прежде чем любой из них завершен. Ваш список задач здесь может состоять из различных целевых функций, как вы можете видеть в этом ответе здесь . Он также позволяет регистрировать обратные вызовы для результатов и ошибок, как в этом примере.

Эти свойства делают Pool.apply_async довольно универсальным и лучшим инструментом для необычных сценариев проблем, которые невозможно выполнить с помощью одной из Pool.map версий.

---

Pool.apply на самом деле не очень широко используется с первого взгляда (и второго). Вы можете использовать его для синхронизации потока управления в сценарии, в котором вы сначала запускаете несколько задач с apply_async, а затем выполняете задачу, которую необходимо выполнить, прежде чем запустить еще один раунд задач с apply_async.

Использование Pool.apply может также означать, что вам придется сэкономить на создании одного дополнительного процесса для промежуточной задачи, когда у вас уже есть пул, который в настоящее время не работает.

Answer 2

Эта строка в вашем коде:

Pool(processes=4, initializer=initializer) as pool:  # start 4 worker processes

не запускает 4 рабочих процесса.Он просто создает их пул, который может поддерживать одновременную работу многих из них.Такие методы, как apply(), на самом деле запускают отдельные процессы.

Разница в том, что apply() и apply_async() состоят в том, что первые блокируют до тех пор, пока результат не будет готов, а последние возвращают объект "результат" прямодалеко.Это не имеет большого значения, если вы не хотите отправлять более чем одну задачу в Pool за один раз (что, конечно, является основным смыслом использования модуля multiprocessing).

Вот некоторыеИзменения в вашем коде, показывающие, как на самом деле выполнять некоторую параллельную обработку с Pool:

from multiprocessing import Pool
import time
import os

def initializer():
    print("In initializer pid is {} ppid is {}".format(os.getpid(),os.getppid()))

def f(x):
    print("In f pid is {} ppid is {}".format(os.getpid(),os.getppid()))
    return x*x

if __name__ == '__main__':
    print("In main pid is {} ppid is {}".format(os.getpid(), os.getppid()))
    with Pool(processes=4, initializer=initializer) as pool:  # Create 4 worker Pool.
#        result = pool.apply(f, (10,)) # evaluate "f(10)" in a single process
#        print(result)
        # Start multiple tasks.
        tasks = [pool.apply_async(f, (val,)) for val in range(10, 20)]
        pool.close()  # No more tasks.
        pool.join()  # Wait for all tasks to finish.
        results = [result.get() for result in tasks]  # Get the result of each.
        print(results)

map_sync(), лучше подходят для обработки чего-то подобного (последовательности значений), поскольку он будет обрабатывать некоторыеиз деталей, показанных в приведенном выше коде автоматически.