Почему многопроцессор работает в одном и том же процессе?

Я запускаю следующее решение из Как восстановить возвращаемое значение функции, переданной для многопроцессорности .Process?:

import multiprocessing
from os import getpid

def worker(procnum):
    print('I am number %d in process %d' % (procnum, getpid()))
    return getpid()

if __name__ == '__main__':
    pool = multiprocessing.Pool(processes = 3)
    print(pool.map(worker, range(5)))

который должен выводить что-то вроде:

I am number 0 in process 19139
I am number 1 in process 19138
I am number 2 in process 19140
I am number 3 in process 19139
I am number 4 in process 19140
[19139, 19138, 19140, 19139, 19140]

но вместо этого я получаю

[4212, 4212, 4212, 4212, 4212]

Если я загружаю pool.map в диапазон 1 000 000, используя более 10 процессов, я вижу не более двух разных pids.

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

Ответы

Ответ 1

TL; DR: задачи не будут специально распределены каким-либо образом, возможно, ваши задачи настолько коротки, что все они завершены до начала других процессов.

От взгляда на источник multiprocessing, кажется, что задачи просто помещаются в Queue, из которого рабочий обрабатывает чтение (функция worker читает из Pool._inqueue). Там нет расчетного распределения, рабочие просто гонки, чтобы работать как можно труднее.

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

Отмечу, что на моей машине все задачи распределяются по процессам довольно однородно (также для #processes > #cores). Таким образом, существует некоторая системная зависимость, хотя все процессы должны иметь .start() ed до того, как работа будет поставлена ​​в очередь.

Здесь некоторый обрезанный источник из worker, который показывает, что задачи просто считываются из очереди каждым процессом, поэтому в псевдослучайном порядке:

def worker(inqueue, outqueue, ...):
    ...
    get = inqueue.get
    ...
    while maxtasks is None or (maxtasks and completed < maxtasks):
        try:
            task = get()
        ...

SimpleQueue связывается между процессами с помощью Pipe, из конструктора SimpleQueue:

self._reader, self._writer = Pipe(duplex=False)

РЕДАКТИРОВАТЬ: возможно, часть слишком медленных процессов начинается с false, поэтому я удалил ее. Все процессы .start() ed до того, как любая работа будет поставлена ​​в очередь (что может быть платформа -зависимо). Я не могу найти, готов ли процесс к моменту возврата .start().