Как справедливо назначать задачи работникам в Python? - Разделение итерируемого на куски одинакового размера

Вы должны предварительно дозировать?

Почему бы просто не создать очередь, чтобы каждый рабочий выходил из очереди, когда он заканчивает единицу работы?

После ответа Тайлера:

def doleOut(queue, workers):
    for worker,task in itertools.izip(itertools.cycle(workers),queue):
        yield worker,task

Это будет возвращать (worker, task) кортежей, пока есть очередь. Итак, если у вас есть блокировка waitForMoreWork, вы можете сделать это:

queue = []
doler = distribute_work(workers, queue)
while 1:
    queue.append(waitForMoreWork)
    currentqueuelen = len(queue)
    for i in range(0,queuelen):
        worker,item = doler.next()
        worker.passitem(item)

Таким образом, он будет блокироваться до тех пор, пока не появится больше элементов очереди, затем распределить их, а затем снова заблокировать. Вы можете настроить свое выражение waitForMoreWork так, чтобы оно выдавало столько предметов за раз, сколько кажется разумным.

Я думаю, вы хотите использовать multiprocessing.Pool.imap для обработки ваших рабочих и распределения их рабочих мест. Я считаю, что он делает все, что вы хотите.

jobs = (some generator)                   # can consume jobs from a generator
pool = multiprocessing.Pool(3)            # set number of workers here
results = pool.imap(process_job, jobs)    # returns a generator

for r in results:                         # loop will block until results arrive
    do_something(r)

Если порядок результатов не имеет значения для вашего приложения, вы также можете использовать imap_unordered.

Хорошо, после того, как сказал, что это невозможно, вот идея. Может быть, это то, что я должен перенести в codereview - мне очень интересны комментарии о том, сколько накладных расходов это влечет за собой в памяти. Другими словами, я не знаю, действительно ли это решает проблему, когда список задач очень длинный и неизвестного размера. Поскольку Blckknght упомянул, что multiprocessing может быть лучшей альтернативой.

Код:

import itertools

def distribute_work(workers, tasks):
    """Return one generator per worker with a fair share of tasks

    Task may be an arbitrary length generator.
    Workers should be an iterable.
    """
    worker_count = len(workers)
    worker_ids = range(worker_count)
    all_tasks_for_all_workers = itertools.tee(tasks, worker_count)
    assignments = [ (workers[id], itertools.islice(i, id, None, worker_count))
                    for (id,i) in enumerate(all_tasks_for_all_workers) ]    
    return(assignments)

Алгоритм заключается в том, чтобы

1. Дублируйте исходный список задач по одному разу для каждого работника. Поскольку это только дублирование объектов генератора, он не должен зависеть от размера списка задач в памяти. Даже если это относительно дорогая операция, это единовременная стоимость запуска и незначительна в памяти для очень больших списков задач.
2. Чтобы назначить задачи одному работнику, каждый работник должен взять часть списка задач. Если #W - это количество рабочих, первый рабочий берет задачи 0, #W, 2*#W, 3*#W и т. д. Вторые рабочие берут 0+1, #W+1, 2*#W+1, 3*#W+1 и т. д. Сращивание для каждого рабочего может быть выполнено с помощью itertools.islice

Для чистого разделения/назначения задач имена рабочих для этой функции на самом деле не требуются. Но количество рабочих есть. Изменение этого сделало бы функцию более универсальной и полезной, а также облегчило бы понимание возвращаемого значения. Чтобы ответить на мой собственный вопрос, я оставлю функцию как есть.

Использование и результат:

>>> for (worker,tasks) in distribute_work(['peter', 'paul', 'mary'], range(5)):
...   print(worker, list(tasks))
... 
peter [0, 3]
paul [1, 4]
mary [2]

И он также обрабатывает случаи, когда воркеры имеют одинаковые имена, но являются разными сущностями:

>>> for (worker,tasks) in distribute_work(['p', 'p', 'mary'], range(5)): 
...   print(worker, list(tasks))
... 
p [0, 3]
p [1, 4]
mary [2]

Вот подход, который мне нравится:

parallelism = os.cpu_count()
num_todos = len(todos)

# this zip fanciness makes each chunk stripe through the data sequentially overall so that the
# first items still get done first across all the workers
chunksize = math.ceil(num_todos / parallelism)
chunks = list(itertools.zip_longest(*[todos[i:i+chunksize] for i in range(0, num_todos, chunksize)]))
chunks = [[c for c in chunk if c is not None] for chunk in chunks]

with Pool(processes=parallelism) as pool:
    tasks = [pool.apply_async(my_function, args=(chunk)) for chunk in chunks]
    [task.get() for task in tasks]

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