Допустим, у меня есть большой список музыки различной длины, которую нужно преобразовать или изображения разных размеров, которые нужно изменить или что-то в этом роде. Заказ не имеет значения, поэтому он идеально подходит для разделения на несколько процессоров.
Если я использую функцию multiprocessing.Pool map, кажется, что вся работа делится раньше времени и не учитывает тот факт, что некоторые файлы могут занимать больше времени, чем другие.
Что происходит, если у меня есть 12 процессоров... ближе к концу обработки, у 1 или 2 процессоров будет осталось 2 или 3 файла для обработки, в то время как другие процессоры, которые могут использоваться, сидят без дела.
Есть ли какая-то реализация очереди, которая может поддерживать загрузку всех процессоров до тех пор, пока не останется больше работы?
Для этой цели существует Queue класс в модуле multiprocessing.
Редактирование. Если вы ищете полную структуру для параллельных вычислений, которая имеет функцию map() с использованием очереди задач, посмотрите на параллельные вычислительные средства IPython. В частности, вы можете использовать функцию TaskClient.map(), чтобы получить сбалансированное по нагрузке сопоставление доступных процессоров.
Это тривиально делать с jug:
def process_image(img):
....
images = glob('*.jpg')
for im in images:
Task(process_image, im)
Теперь просто запустите jug execute несколько раз, чтобы обработать рабочие процессы.
О реализации очередей. Есть некоторые.
Посмотрите на проект Сельдерей. http://celeryproject.org/
Итак, в вашем случае вы можете запускать 12 конверсий (по одному на каждый процессор) в качестве задач Celery, добавлять функцию обратного вызова (к преобразованию или задаче), а в этой функции обратного вызова добавить новую задачу преобразования, выполняемую, когда один предыдущих преобразований завершено.
Библиотека потоков Python, которая принесла мне большую радость, Параллельный Python (PP). Тривиально, чтобы PP использовал подход с пулом потоков с одной очередью для достижения того, что вам нужно.
Это не тот случай, если вы используете Pool.imap_unordered.