Предоставляют ли многопроцессорные пулы каждому процессу одинаковое количество задач или они назначаются как доступные?

Итак, учитывая этот непроверенный код предложения; если в пуле 4 процесса, каждому процессу выделяется 25 элементов для выполнения, или 100 элементов выбираются один за другим процессами, которые ищут элементы для выполнения, чтобы каждый процесс мог выполнять разное количество элементов, например 30 , 26, 24, 20.

Что ж, очевидный ответ - проверить это.

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

import collections
import multiprocessing
import os
import random
import time

def generate_stuff():
    for foo in range(100):
        yield foo

def process(moo):
    #print moo
    time.sleep(random.randint(0, 50) / 10.)
    return os.getpid()

pool = multiprocessing.Pool()
pids = pool.map(func=process, iterable=generate_stuff(), chunksize=1)
pool.close()
print collections.Counter(pids)

Если числа «неровные», вы знаете, что объединенные процессы должны получать новые задания как готовые. (Я явно установил chunksize в 1, чтобы убедиться, что фрагменты не настолько велики, чтобы каждый получил только один фрагмент.)

Когда я запускаю его на 8-ядерной машине:

Counter({98935: 16, 98936: 16, 98939: 13, 98937: 12, 98942: 12, 98938: 11, 98940: 11, 98941: 9})

Итак, похоже, что процессы получают новые рабочие места на лету.

Поскольку вы конкретно спросили о 4 рабочих, я изменил Pool() на Pool(4) и получил следующее:

Counter({98965: 31, 98962: 24, 98964: 23, 98963: 22})

Однако есть еще лучший способ узнать, чем тестирование: прочтите источник.

Как видите, map просто вызывает map_async, который создает группу пакетов и помещает их в объект self._taskqueue (экземпляр Queue.Queue). Если вы читаете дальше, эта очередь не используется совместно с другими процессами напрямую, но есть поток диспетчера пула, который всякий раз, когда процесс завершается и возвращает результат, выталкивает следующее задание из очереди и отправляет его обратно процессу.

Таким же образом можно узнать размер блока по умолчанию для map. Реализация 2.7, связанная выше, показывает, что она просто округлена len(iterable) / (len(self._pool) * 4) (немного более подробна, чем это, чтобы избежать дробной арифметики) - или, другими словами, достаточно большой для примерно 4 фрагментов на процесс. Но на это не стоит полагаться; документация нечетко и косвенно подразумевает, что будет использоваться какая-то эвристика, но не дает никаких гарантий относительно того, что это будет. Итак, если вам действительно нужно «около 4 частей на процесс», рассчитайте это явно. Более реалистично, если вам когда-либо понадобится что-то, кроме значения по умолчанию, вам, вероятно, понадобится значение, зависящее от предметной области, которое вы собираетесь обработать (путем расчета, предположения или профилирования).

http://docs.python.org/2/library/multiprocessing.html#multiprocessing.pool.multiprocessing.Pool.map

map(func, iterable[, chunksize])

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

Я предполагаю, что процесс забирает следующий фрагмент из очереди, когда завершает работу с предыдущим фрагментом.

Значение по умолчанию chunksize зависит от длины iterable и выбирается таким образом, чтобы количество фрагментов было примерно в четыре раза больше количества процессов. (источник)

Чтобы оценить chunksize, используемый реализацией Python, не глядя на multiprocessing исходный код модуля, запустите:

#!/usr/bin/env python
import multiprocessing as mp
from itertools import groupby

def work(index):
    mp.get_logger().info(index)
    return index, mp.current_process().name

if __name__ == "__main__":
    import logging
    import sys
    logger = mp.log_to_stderr()

    # process cmdline args
    try:
        sys.argv.remove('--verbose')
    except ValueError:
        pass  # not verbose
    else:
        logger.setLevel(logging.INFO)  # verbose
    nprocesses, nitems = int(sys.argv.pop(1)), int(sys.argv.pop(1))
    # choices: 'map', 'imap', 'imap_unordered'
    map_name = sys.argv[1] if len(sys.argv) > 1 else 'map'
    kwargs = dict(chunksize=int(sys.argv[2])) if len(sys.argv) > 2 else {}

    # estimate chunksize used
    max_chunksize = 0
    map_func = getattr(mp.Pool(nprocesses), map_name)
    for _, group in groupby(sorted(map_func(work, range(nitems), **kwargs),
                                   key=lambda x: x[0]),  # sort by index
                            key=lambda x: x[1]):  # group by process name
        max_chunksize = max(max_chunksize, len(list(group)))
    print("%s: max_chunksize %d" % (map_name, max_chunksize))

Он показывает, что imap, imap_unordered используют chunksize=1 по умолчанию, а max_chunksize для map зависит от nprocesses, nitem (количество фрагментов на процесс не фиксировано) и max_chunksize зависит от версии Python. Все *map* функции учитывают параметр chunksize, если он указан.

использование

$ ./estimate_chunksize.py nprocesses nitems [map_name [chunksize]] [--verbose]

Чтобы увидеть, как распределяются отдельные рабочие места; укажите параметр --verbose.