Блог

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

Я понимаю, как с multiprocessing помощью вы можете запускать несколько экземпляров функции над набором данных с помощью фрагмента кода, подобного этому:

 from multiprocessing import Pool

with Pool() as pool:
    pool.map(my_func, data)
  

Теперь, что, если у меня есть две функции, первая из которых является входом для второй, чтобы они могли выполняться параллельно? Я понимаю, что идея состоит в том, чтобы использовать очередь, но все примеры, которые я нашел, не так быстро и легко понять. Я просмотрел раздел 10.4.10 «Стандартная библиотека Python 3 на примере», но не смог понять, как применить это к моему случаю.

Мой вариант использования — это программа, которая копирует файлы в целевую папку, затем вычисляет контрольную сумму MD5 копии. Я бы хотел, чтобы копия следующего файла запускалась во время вычисления контрольной суммы MD5 ранее завершенной копии. Таким образом, вывод функции (copy) представляет собой имя файла с путем (string), который является вводом второй функции (контрольная сумма MD5).

Есть ли у кого-нибудь базовая модель кода, подобная приведенной выше, чтобы я начал?

Не зная вашей реализации, но это не совсем задача Pool . Первый вопрос заключается в том, что будет делать ваша основная программа.

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

Другая возможность — выполнить обе задачи в одной и той же функции внутри пула. Если у вас есть дополнительные ресурсы, просто увеличьте количество рабочих. Обе ваши задачи требуют интенсивного ввода-вывода. Узким местом этого будет ввод-вывод, и вы ничего не выиграете от добавления большего количества процессора к задаче. Это мой любимый, основанный на том, что я мало знаю.

Если вам действительно нужны два подпроцесса для этого — один для копирования и один для подсчета контрольных сумм, вам понадобятся пул, подпроцесс и очередь. Или только два подпроцесса, если пул не нужен. Я слишком ленив, чтобы написать рабочий пример, но идея состоит в том, чтобы создать multiprocessing.queue() , а затем использовать multiprocessing.Process для запуска калькулятор контрольной суммы с очередью в качестве параметра. Наконец, вы запустите пул — или другой процесс, если одного достаточно для насыщения вашего ввода-вывода — и передадите ту же очередь рабочему.

Если вы в конечном итоге используете Pool и решите использовать map или map_async, вам также потребуется использовать partial from functools для создания копии вашего worker, у которой уже есть очередь в качестве параметра, поскольку map может отправлять только один параметр — имя файла. Если вы используете apply_async, то вы можете отправить рабочему как имя файла, так и очередь в качестве параметров.

Когда ваш копировальный аппарат скопировал файл, он put() добавит имя файла в очередь. Ваш процесс проверки контрольной суммы — это цикл while true, который находится в queue.get() и ожидает появления имен файлов. Если вам затем нужно где-то вернуть контрольную сумму, вам понадобится другая очередь для исходящего трафика, другой параметр для подпроцессов и основная программа, ожидающая чтения результатов из этой очереди.

Это работает, но я уверен, что это не улучшит производительность по сравнению с моделью, в которой ваш рабочий выполняет копирование и вычисление контрольной суммы за один раз. Как вы можете видеть, это становится намного более запутанным, и вы потенциально потеряете некоторую скорость при отправке информации в очередях туда и обратно. Очереди работают, но они не являются чудодейственными исполнителями.