multiprocessing — Параллелизм на основе процессов

Класс Process

В multiprocessing процессы создаются путём создания объекта Process и последующего вызова его метода start(). Класс Process соответствует API класса threading.Thread. Простейший пример многопроцессной программы:

from multiprocessing import Process

def f(name):
    print('hello', name)

if __name__ == '__main__':
    p = Process(target=f, args=('bob',))
    p.start()
    p.join()

Вот расширенный пример, показывающий идентификаторы отдельных процессов:

from multiprocessing import Process
import os

def info(title):
    print(title)
    print('module name:', __name__)
    print('parent process:', os.getppid())
    print('process id:', os.getpid())

def f(name):
    info('function f')
    print('hello', name)

if __name__ == '__main__':
    info('main line')
    p = Process(target=f, args=('bob',))
    p.start()
    p.join()

Объяснение того, почему необходима часть if __name__ == '__main__', см. в разделе Рекомендации по программированию.

Контексты и методы запуска

В зависимости от платформы multiprocessing поддерживает три способа запуска процесса. Эти методы запуска:

spawn

Родительский процесс запускает новый процесс-интерпретатор Python. Дочерний процесс наследует только те ресурсы, которые необходимы для выполнения метода run() объекта процесса. В частности, ненужные файловые дескрипторы и дескрипторы родительского процесса не наследуются. Запуск процесса этим способом довольно медленный по сравнению с использованием fork или forkserver.

Доступен в Unix и Windows. Используется по умолчанию в Windows и macOS.

fork

Родительский процесс использует os.fork() для создания копии интерпретатора Python. После запуска дочерний процесс практически идентичен родительскому. Дочерний процесс наследует все ресурсы родительского. Обратите внимание: безопасное создание копии многопоточного процесса проблематично.

Доступен только в Unix. Используется по умолчанию в Unix.

forkserver

При запуске программы с выбором метода forkserver запускается серверный процесс. Затем, когда требуется новый процесс, родительский процесс подключается к серверу и запрашивает создание копии нового процесса. Серверный процесс работает в одном потоке, поэтому может безопасно использовать os.fork(). Ненужные ресурсы не наследуются.

Доступен на платформах Unix, поддерживающих передачу файловых дескрипторов через каналы Unix.

В POSIX при использовании методов запуска spawn или forkserver также запускается процесс отслеживания ресурсов, который отслеживает именованные системные ресурсы, созданные процессами программы и удалённые из системы (например, именованные семафоры или объекты SharedMemory). После завершения всех процессов отслеживатель ресурсов удаляет все оставшиеся отслеживаемые объекты. Обычно таких объектов не должно быть, но если процесс был завершён сигналом, могут остаться некоторые «утёкшие» ресурсы. (Ни утёкшие семафоры, ни сегменты разделяемой памяти не будут удалены автоматически до следующей перезагрузки. Это проблематично для обоих типов объектов, поскольку система допускает лишь ограниченное количество именованных семафоров, а сегменты разделяемой памяти занимают место в основной памяти.)

Чтобы выбрать метод запуска, используйте set_start_method() в блоке if __name__ == '__main__' главного модуля. Например:

import multiprocessing as mp

def foo(q):
    q.put('hello')

if __name__ == '__main__':
    mp.set_start_method('spawn')
    q = mp.Queue()
    p = mp.Process(target=foo, args=(q,))
    p.start()
    print(q.get())
    p.join()

set_start_method() не следует вызывать в программе более одного раза.

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

import multiprocessing as mp

def foo(q):
    q.put('hello')

if __name__ == '__main__':
    ctx = mp.get_context('spawn')
    q = ctx.Queue()
    p = ctx.Process(target=foo, args=(q,))
    p.start()
    print(q.get())
    p.join()

Обратите внимание, что объекты, связанные с одним контекстом, могут быть несовместимы с процессами из другого контекста. В частности, блокировки, созданные с помощью контекста fork, нельзя передавать процессам, запущенным с помощью методов spawn или forkserver.

Библиотеке, которой требуется определённый метод запуска, вероятно, следует использовать get_context(), чтобы не вмешиваться в выбор пользователя библиотеки.

Обмен объектами между процессами

multiprocessing поддерживает два типа каналов связи между процессами:

Очереди

Класс Queue почти полностью повторяет класс queue.Queue. Например:

from multiprocessing import Process, Queue

def f(q):
    q.put([42, None, 'hello'])

if __name__ == '__main__':
    q = Queue()
    p = Process(target=f, args=(q,))
    p.start()
    print(q.get())    # prints "[42, None, 'hello']"
    p.join()

Очереди безопасны для потоков и процессов.

Каналы

Функция Pipe() возвращает пару объектов соединения, связанных каналом, который по умолчанию является дуплексным (двусторонним). Например:

from multiprocessing import Process, Pipe

def f(conn):
    conn.send([42, None, 'hello'])
    conn.close()

if __name__ == '__main__':
    parent_conn, child_conn = Pipe()
    p = Process(target=f, args=(child_conn,))
    p.start()
    print(parent_conn.recv())   # prints "[42, None, 'hello']"
    p.join()

Два объекта соединения, возвращаемые функцией Pipe(), представляют два конца канала. Каждый объект соединения имеет методы send() и recv() (среди прочих). Обратите внимание: данные в канале могут быть повреждены, если два процесса (или потока) одновременно попытаются читать из одного и того же конца канала или записывать в него. Разумеется, если процессы одновременно используют разные концы канала, риск повреждения данных отсутствует.

Синхронизация между процессами

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

from multiprocessing import Process, Lock

def f(l, i):
    l.acquire()
    try:
        print('hello world', i)
    finally:
        l.release()

if __name__ == '__main__':
    lock = Lock()

    for num in range(10):
        Process(target=f, args=(lock, num)).start()

Без блокировки вывод разных процессов может перемешаться.

Совместное использование состояния между процессами

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

Однако если вам действительно необходимо использовать общие данные, multiprocessing предлагает несколько способов сделать это.

Разделяемая память

Данные можно хранить в отображении разделяемой памяти с помощью Value или Array. Например, следующий код

from multiprocessing import Process, Value, Array

def f(n, a):
    n.value = 3.1415927
    for i in range(len(a)):
        a[i] = -a[i]

if __name__ == '__main__':
    num = Value('d', 0.0)
    arr = Array('i', range(10))

    p = Process(target=f, args=(num, arr))
    p.start()
    p.join()

    print(num.value)
    print(arr[:])

выведет

3.1415927
[0, -1, -2, -3, -4, -5, -6, -7, -8, -9]

Аргументы 'd' и 'i', используемые при создании num и arr, — это коды типов, применяемые в модуле array: 'd' обозначает число с плавающей точкой двойной точности, а 'i' — целое число со знаком. Эти общие объекты безопасны для процессов и потоков.

Для более гибкой работы с разделяемой памятью можно использовать модуль multiprocessing.sharedctypes, который позволяет создавать произвольные объекты ctypes, размещённые в разделяемой памяти.

Серверный процесс

Объект менеджера, возвращаемый функцией Manager(), управляет серверным процессом, в котором хранятся объекты Python и который позволяет другим процессам работать с ними через прокси-объекты.

Менеджер, возвращаемый функцией Manager(), поддерживает типы list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value и Array. Например:

from multiprocessing import Process, Manager

def f(d, l):
    d[1] = '1'
    d['2'] = 2
    d[0.25] = None
    l.reverse()

if __name__ == '__main__':
    with Manager() as manager:
        d = manager.dict()
        l = manager.list(range(10))

        p = Process(target=f, args=(d, l))
        p.start()
        p.join()

        print(d)
        print(l)

выведет

{0.25: None, 1: '1', '2': 2}
[9, 8, 7, 6, 5, 4, 3, 2, 1, 0]

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

Использование пула рабочих процессов

Класс Pool представляет пул рабочих процессов. Он имеет методы, позволяющие разными способами передавать задачи рабочим процессам.

Например:

from multiprocessing import Pool, TimeoutError
import time
import os

def f(x):
    return x*x

if __name__ == '__main__':
    # start 4 worker processes
    with Pool(processes=4) as pool:

        # print "[0, 1, 4,..., 81]"
        print(pool.map(f, range(10)))

        # print same numbers in arbitrary order
        for i in pool.imap_unordered(f, range(10)):
            print(i)

        # evaluate "f(20)" asynchronously
        res = pool.apply_async(f, (20,))      # runs in *only* one process
        print(res.get(timeout=1))             # prints "400"

        # evaluate "os.getpid()" asynchronously
        res = pool.apply_async(os.getpid, ()) # runs in *only* one process
        print(res.get(timeout=1))             # prints the PID of that process

        # launching multiple evaluations asynchronously *may* use more processes
        multiple_results = [pool.apply_async(os.getpid, ()) for i in range(4)]
        print([res.get(timeout=1) for res in multiple_results])

        # make a single worker sleep for 10 seconds
        res = pool.apply_async(time.sleep, (10,))
        try:
            print(res.get(timeout=1))
        except TimeoutError:
            print("We lacked patience and got a multiprocessing.TimeoutError")

        print("For the moment, the pool remains available for more work")

    # exiting the 'with'-block has stopped the pool
    print("Now the pool is closed and no longer available")

Обратите внимание: методы пула должен использовать только создавший его процесс.

>>> from multiprocessing import Pool
>>> p = Pool(5)
>>> def f(x):
...     return x*x
...
>>> with p:
...     p.map(f, [1,2,3])
Process PoolWorker-1:
Process PoolWorker-2:
Process PoolWorker-3:
Traceback (most recent call last):
Traceback (most recent call last):
Traceback (most recent call last):
AttributeError: Can't get attribute 'f' on <module '__main__' (built-in)>
AttributeError: Can't get attribute 'f' on <module '__main__' (built-in)>
AttributeError: Can't get attribute 'f' on <module '__main__' (built-in)>

Process и исключения

class multiprocessing.Process(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)

Объекты Process представляют действия, выполняемые в отдельном процессе. Класс Process имеет аналоги всех методов класса threading.Thread.

Конструктор всегда следует вызывать с аргументами-ключевыми словами. Значение group всегда должно быть None; этот параметр существует исключительно для совместимости с threading.Thread. target — вызываемый объект, который будет вызван методом run(). По умолчанию имеет значение None, то есть ничего не вызывается. name — имя процесса (подробнее см. name). args — кортеж аргументов для вызова целевого объекта. kwargs — словарь аргументов-ключевых слов для вызова целевого объекта. Если аргумент daemon, принимающий только ключевые слова, задан, он устанавливает флаг daemon процесса в True или False. Если задано значение None (по умолчанию), этот флаг наследуется от процесса, создавшего данный процесс.

По умолчанию целевому объекту target не передаются аргументы. Аргумент args, значение которого по умолчанию — (), можно использовать для указания списка или кортежа аргументов, передаваемых target.

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

Изменено в версии 3.3: Добавлен параметр daemon.

run()

Метод, представляющий действия процесса.

Этот метод можно переопределить в подклассе. Стандартный метод run() вызывает вызываемый объект, переданный при создании объекта в качестве аргумента target (если он задан), с позиционными и именованными аргументами, полученными соответственно из аргументов args и kwargs.

Тот же результат можно получить, передав список или кортеж в качестве аргумента args при вызове Process.

Пример:

>>> from multiprocessing import Process
>>> p = Process(target=print, args=[1])
>>> p.run()
1
>>> p = Process(target=print, args=(1,))
>>> p.run()
1

start()

Запускает выполнение процесса.

Для каждого объекта процесса этот метод можно вызвать не более одного раза. Он обеспечивает вызов метода объекта run() в отдельном процессе.

join([timeout])

Если необязательный аргумент timeout равен None (значение по умолчанию), метод блокируется до завершения процесса, для которого вызван метод join(). Если timeout — положительное число, метод блокируется не более чем на timeout секунд. Обратите внимание, что метод возвращает None, если процесс завершился или истекло время ожидания. Чтобы определить, завершился ли процесс, проверьте его exitcode.

К процессу можно присоединяться многократно.

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

name

Имя процесса. Имя — это строка, используемая только для идентификации. Оно не имеет семантического значения. Нескольким процессам можно присвоить одно и то же имя.

Начальное имя задается конструктором. Если в конструкторе не указано имя, формируется имя вида «Process-N₁:N₂:…:N_k», где каждый N_k — это N-й потомок своего родительского процесса.

is_alive()

Возвращает, выполняется ли процесс.

Приблизительно говоря, объект процесса считается активным с момента возврата метода start() до завершения дочернего процесса.

daemon

Флаг демонического процесса — логическое значение. Его необходимо установить до вызова start().

Начальное значение наследуется от процесса, создавшего данный процесс.

При завершении процесс пытается завершить все свои дочерние демонические процессы.

Обратите внимание, что демоническому процессу не разрешено создавать дочерние процессы. В противном случае после завершения родительского процесса его дочерние процессы останутся без родителя. Кроме того, это не демоны или службы Unix, а обычные процессы, которые будут завершены (без присоединения к ним), если завершились недемонические процессы.

Помимо API threading.Thread, объекты Process также поддерживают следующие атрибуты и методы:

pid

Возвращает идентификатор процесса. До запуска процесса возвращается None.

exitcode

Код завершения дочернего процесса. Если процесс еще не завершился, возвращается None.

Если метод run() дочернего процесса завершился обычным образом, код завершения будет равен 0. Если процесс завершился с помощью sys.exit() с целочисленным аргументом N, код завершения будет равен N.

Если дочерний процесс завершился из-за исключения, не перехваченного в run(), код завершения будет равен 1. Если процесс был завершен сигналом N, код завершения будет равен отрицательному значению -N.

authkey

Ключ аутентификации процесса (байтовая строка).

При инициализации multiprocessing основному процессу назначается случайная строка с помощью os.urandom().

При создании объекта Process он наследует ключ аутентификации родительского процесса, но его можно изменить, присвоив атрибуту authkey другую байтовую строку.

См. раздел Ключи аутентификации.

sentinel

Числовой дескриптор системного объекта, который переходит в состояние «готов» после завершения процесса.

Это значение можно использовать, если требуется одновременно ожидать несколько событий с помощью multiprocessing.connection.wait(). В противном случае проще вызвать join().

В Windows это дескриптор ОС, который можно использовать с семейством вызовов API WaitForSingleObject и WaitForMultipleObjects. В Unix это файловый дескриптор, который можно использовать с примитивами модуля select.

Добавлено в версии 3.3.

terminate()

Завершает процесс. В POSIX это выполняется с помощью сигнала SIGTERM; в Windows используется TerminateProcess(). Обратите внимание, что обработчики выхода, блоки finally и т. п. выполнены не будут.

Обратите внимание, что дочерние процессы данного процесса не будут завершены — они просто останутся без родителя.

Предупреждение

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

kill()

Аналогичен terminate(), но в Unix использует сигнал SIGKILL.

Добавлено в версии 3.7.

close()

Закрывает объект Process, освобождая все связанные с ним ресурсы. Если базовый процесс еще выполняется, вызывается исключение ValueError. После успешного возврата close() большинство остальных методов и атрибутов объекта Process будут вызывать исключение ValueError.

Добавлено в версии 3.7.

Обратите внимание, что методы start(), join(), is_alive(), terminate() и exitcode должен вызывать только процесс, создавший объект процесса.

Пример использования некоторых методов Process:

>>> import multiprocessing, time, signal
>>> p = multiprocessing.Process(target=time.sleep, args=(1000,))
>>> print(p, p.is_alive())
<Process ... initial> False
>>> p.start()
>>> print(p, p.is_alive())
<Process ... started> True
>>> p.terminate()
>>> time.sleep(0.1)
>>> print(p, p.is_alive())
<Process ... stopped exitcode=-SIGTERM> False
>>> p.exitcode == -signal.SIGTERM
True

exception multiprocessing.ProcessError

Базовый класс всех исключений multiprocessing.

exception multiprocessing.BufferTooShort

Исключение, вызываемое Connection.recv_bytes_into(), если переданный буфер слишком мал для прочитанного сообщения.

Если e является экземпляром BufferTooShort, то e.args[0] вернет сообщение в виде байтовой строки.

exception multiprocessing.AuthenticationError

Вызывается при ошибке аутентификации.

exception multiprocessing.TimeoutError

Вызывается методами с тайм-аутом по истечении времени ожидания.

Каналы и очереди

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

Для передачи сообщений можно использовать Pipe() (для соединения двух процессов) или очередь (которая допускает нескольких производителей и потребителей).

Типы Queue, SimpleQueue и JoinableQueue представляют собой очереди FIFO с несколькими производителями и потребителями, созданные по образцу класса queue.Queue из стандартной библиотеки. Они отличаются тем, что в Queue отсутствуют методы task_done() и join(), добавленные в класс queue.Queue в Python 2.5.

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

Обратите внимание, что общую очередь также можно создать с помощью объекта-менеджера — см. Менеджеры.

Пример использования очередей для межпроцессного взаимодействия см. в разделе Примеры.

multiprocessing.Pipe([duplex])

Возвращает пару (conn1, conn2) объектов Connection, представляющих концы канала.

Если duplex имеет значение True (по умолчанию), канал является двунаправленным. Если duplex имеет значение False, канал является однонаправленным: conn1 можно использовать только для получения сообщений, а conn2 — только для отправки сообщений.

class multiprocessing.Queue([maxsize])

Возвращает общую для процессов очередь, реализованную с помощью канала и нескольких блокировок/семафоров. Когда процесс впервые помещает элемент в очередь, запускается поток-поставщик, который передаёт объекты из буфера в канал.

Для обозначения истечения времени ожидания вызываются стандартные исключения queue.Empty и queue.Full из модуля queue стандартной библиотеки.

Queue реализует все методы queue.Queue, кроме task_done() и join().

qsize()

Возвращает приблизительный размер очереди. Из-за особенностей многопоточности и многопроцессности это число ненадёжно.

Обратите внимание, что на Unix-платформах, например macOS, этот метод может вызвать NotImplementedError, поскольку sem_getvalue() не реализован.

empty()

Возвращает True, если очередь пуста, и False в противном случае. Из-за особенностей многопоточности и многопроцессности этот метод ненадёжен.

full()

Возвращает True, если очередь заполнена, и False в противном случае. Из-за особенностей многопоточности и многопроцессности этот метод ненадёжен.

put(obj[, block[, timeout]])

Помещает obj в очередь. Если необязательный аргумент block имеет значение True (по умолчанию), а timeout — None (по умолчанию), выполнение приостанавливается при необходимости до появления свободного места. Если timeout — положительное число, ожидание длится не более timeout секунд, после чего вызывается исключение queue.Full, если за это время свободного места не появилось. В противном случае (если block имеет значение False) элемент помещается в очередь, если свободное место доступно немедленно, иначе вызывается исключение queue.Full (в этом случае timeout игнорируется).

Изменено в версии 3.8: Если очередь закрыта, вместо AssertionError вызывается ValueError.

put_nowait(obj)

Эквивалентен put(obj, False).

get([block[, timeout]])

Извлекает и возвращает элемент из очереди. Если необязательный аргумент block имеет значение True (по умолчанию), а timeout — None (по умолчанию), выполнение приостанавливается при необходимости до появления элемента. Если timeout — положительное число, ожидание длится не более timeout секунд, после чего вызывается исключение queue.Empty, если за это время элемент не появился. В противном случае (если block имеет значение False) элемент возвращается, если он доступен немедленно, иначе вызывается исключение queue.Empty (в этом случае timeout игнорируется).

Изменено в версии 3.8: Если очередь закрыта, вместо OSError вызывается ValueError.

get_nowait()

Эквивалентен get(False).

У multiprocessing.Queue есть несколько дополнительных методов, отсутствующих в queue.Queue. Обычно большинству программ эти методы не нужны:

close()

Указывает, что текущий процесс больше не будет помещать данные в эту очередь. Фоновый поток завершится, когда передаст в канал все данные из буфера. Этот метод вызывается автоматически при сборке мусора для очереди.

join_thread()

Ожидает завершения фонового потока. Этот метод можно использовать только после вызова close(). Он блокирует выполнение до завершения фонового потока, гарантируя, что все данные из буфера переданы в канал.

По умолчанию при завершении процесс, не создавший очередь, пытается дождаться завершения её фонового потока. Процесс может вызвать cancel_join_thread(), чтобы метод join_thread() ничего не делал.

cancel_join_thread()

Предотвращает блокировку вызова join_thread(). В частности, это предотвращает автоматическое ожидание завершения фонового потока при выходе процесса — см. join_thread().

Более подходящим названием для этого метода могло бы быть allow_exit_without_flush(). При этом данные, помещённые в очередь, могут быть потеряны, и вам почти наверняка не понадобится использовать этот метод. Он нужен лишь в том случае, если текущему процессу необходимо немедленно завершиться, не дожидаясь передачи данных из очереди в нижележащий канал, и потеря данных вас не беспокоит.

Примечание

Для работы этого класса операционная система должна поддерживать работающую реализацию общего семафора. При её отсутствии функциональность этого класса будет отключена, а попытка создать экземпляр Queue приведёт к ImportError. Дополнительную информацию см. в bpo-3770. То же относится ко всем перечисленным ниже специализированным типам очередей.

class multiprocessing.SimpleQueue

Это упрощённый тип Queue, очень близкий к заблокированному Pipe.

close()

Закрывает очередь и освобождает внутренние ресурсы.

После закрытия очередь больше нельзя использовать. Например, больше нельзя вызывать методы get(), put() и empty().

Добавлено в версии 3.9.

empty()

Возвращает True, если очередь пуста, и False в противном случае.

get()

Извлекает и возвращает элемент из очереди.

put(item)

Помещает item в очередь.

class multiprocessing.JoinableQueue([maxsize])

JoinableQueue — подкласс Queue, представляющий собой очередь, в которой дополнительно есть методы task_done() и join().

task_done()

Указывает, что ранее помещённая в очередь задача выполнена. Используется потребителями очереди. Для каждого вызова get(), используемого для получения задачи, последующий вызов task_done() сообщает очереди, что обработка задачи завершена.

Если вызов join() в данный момент заблокирован, он возобновит работу после обработки всех элементов (то есть когда для каждого элемента, помещённого в очередь с помощью put(), будет получен вызов task_done()).

Вызывает ValueError, если вызван больше раз, чем число элементов, помещённых в очередь.

join()

Блокирует выполнение до получения и обработки всех элементов очереди.

Счётчик незавершённых задач увеличивается при добавлении элемента в очередь. Счётчик уменьшается, когда потребитель вызывает task_done(), указывая, что элемент получен и вся работа с ним завершена. Когда счётчик незавершённых задач достигает нуля, вызов join() разблокируется.

Разное

multiprocessing.active_children()

Возвращает список всех работающих дочерних процессов текущего процесса.

Вызов этой функции также приводит к «присоединению» всех уже завершившихся процессов.

multiprocessing.cpu_count()

Возвращает число процессоров в системе.

Это число не равно числу процессоров, которые может использовать текущий процесс. Число доступных процессоров можно получить с помощью len(os.sched_getaffinity(0))

Если определить число процессоров невозможно, вызывается NotImplementedError.

См. также

os.cpu_count()

multiprocessing.current_process()

Возвращает объект Process, соответствующий текущему процессу.

Аналог threading.current_thread().

multiprocessing.parent_process()

Возвращает объект Process, соответствующий родительскому процессу процесса, возвращаемого функцией current_process(). Для главного процесса parent_process будет равен None.

Добавлено в версии 3.8.

multiprocessing.freeze_support()

Добавляет поддержку программ, использующих multiprocessing, которые были заморожены для создания исполняемого файла Windows. (Проверено с помощью py2exe, PyInstaller и cx_Freeze.)

Эту функцию нужно вызвать сразу после строки if __name__ == '__main__' главного модуля. Например:

from multiprocessing import Process, freeze_support

def f():
    print('hello world!')

if __name__ == '__main__':
    freeze_support()
    Process(target=f).start()

Если строка freeze_support() опущена, попытка запустить замороженный исполняемый файл вызовет RuntimeError.

Вызов freeze_support() не оказывает никакого эффекта в операционных системах, отличных от Windows. Кроме того, если модуль обычно запускается интерпретатором Python в Windows (то есть программа не была заморожена), freeze_support() также не оказывает никакого эффекта.

multiprocessing.get_all_start_methods()

Возвращает список поддерживаемых методов запуска; первый метод в списке используется по умолчанию. Возможные методы запуска: 'fork', 'spawn' и 'forkserver'. В Windows доступен только 'spawn'. В Unix всегда поддерживаются 'fork' и 'spawn', при этом по умолчанию используется 'fork'.

Добавлено в версии 3.4.

multiprocessing.get_context(method=None)

Возвращает объект контекста с теми же атрибутами, что и у модуля multiprocessing.

Если method имеет значение None, возвращается контекст по умолчанию. В противном случае method должен иметь значение 'fork', 'spawn' или 'forkserver'. Если указанный метод запуска недоступен, вызывается ValueError.

Добавлено в версии 3.4.

multiprocessing.get_start_method(allow_none=False)

Возвращает название метода, используемого для запуска процессов.

Если метод запуска ещё не зафиксирован и allow_none имеет значение false, устанавливается метод запуска по умолчанию и возвращается его название. Если метод запуска ещё не зафиксирован и allow_none имеет значение true, возвращается None.

Возвращаемое значение может быть 'fork', 'spawn', 'forkserver' или None. В Unix по умолчанию используется 'fork', а в Windows и macOS — 'spawn'.

Добавлено в версии 3.4.

Изменено в версии 3.8: В macOS метод запуска spawn теперь используется по умолчанию. Метод запуска fork следует считать небезопасным, поскольку он может привести к сбоям дочернего процесса. См. bpo-33725.

multiprocessing.set_executable(executable)

Задаёт путь к интерпретатору Python, который будет использоваться при запуске дочернего процесса. (По умолчанию используется sys.executable.) Встраивающим Python-приложениям, вероятно, потребуется выполнить что-то вроде следующего:

set_executable(os.path.join(sys.exec_prefix, 'pythonw.exe'))

прежде чем создавать дочерние процессы.

Изменено в версии 3.4: Теперь поддерживается в Unix при использовании метода запуска 'spawn'.

Изменено в версии 3.11: Принимает объект, подобный пути.

multiprocessing.set_start_method(method, force=False)

Задаёт метод, который следует использовать для запуска дочерних процессов. Аргумент method может иметь значение 'fork', 'spawn' или 'forkserver'. Если метод запуска уже задан, а force не имеет значения True, вызывается RuntimeError. Если method имеет значение None, а force — True, метод запуска устанавливается в None. Если method имеет значение None, а force — False, контекст устанавливается в контекст по умолчанию.

Обратите внимание, что эту функцию следует вызывать не более одного раза и только внутри условия if __name__ == '__main__' главного модуля.

Добавлено в версии 3.4.

Объекты соединения

Объекты соединения позволяют отправлять и получать объекты, поддерживающие pickle, или строки. Их можно рассматривать как подключённые сокеты, ориентированные на сообщения.

Объекты соединения обычно создаются с помощью Pipe — см. также раздел Слушатели и клиенты.

class multiprocessing.connection.Connection

send(obj)

Отправляет объект на другой конец соединения; для его чтения следует использовать recv().

Объект должен поддерживать pickle. Очень большие объекты pickle (примерно от 32 МиБ, хотя это зависит от ОС) могут вызвать исключение ValueError.

recv()

Возвращает объект, отправленный с другого конца соединения с помощью send(). Блокирует выполнение, пока не появится объект для получения. Вызывает EOFError, если получать больше нечего и другой конец соединения закрыт.

fileno()

Возвращает файловый дескриптор или дескриптор, используемый соединением.

close()

Закрывает соединение.

Этот метод вызывается автоматически при удалении объекта соединения сборщиком мусора.

poll([timeout])

Возвращает, доступны ли данные для чтения.

Если timeout не указан, метод возвращает результат немедленно. Если timeout — число, оно задаёт максимальное время блокировки в секундах. Если timeout равен None, используется бесконечное время ожидания.

Обратите внимание: несколько объектов соединения можно опрашивать одновременно с помощью multiprocessing.connection.wait().

send_bytes(buffer[, offset[, size]])

Отправляет байтовые данные из объекта, подобного байтам, как целое сообщение.

Если задан offset, данные считываются из этой позиции в buffer. Если задан size, из буфера считывается указанное количество байтов. Очень большие буферы (примерно от 32 МиБ, хотя это зависит от ОС) могут вызвать исключение ValueError.

recv_bytes([maxlength])

Возвращает полное сообщение с байтовыми данными, отправленное с другого конца соединения, в виде строки. Блокирует выполнение, пока не появятся данные для получения. Вызывает EOFError, если получать больше нечего и другой конец соединения закрыт.

Если задан maxlength и длина сообщения превышает maxlength, возникает исключение OSError, после чего соединение становится недоступным для чтения.

Изменено в версии 3.3: Раньше эта функция вызывала IOError, который теперь является псевдонимом OSError.

recv_bytes_into(buffer[, offset])

Считывает в buffer полное сообщение с байтовыми данными, отправленное с другого конца соединения, и возвращает количество байтов в сообщении. Блокирует выполнение, пока не появятся данные для получения. Вызывает EOFError, если получать больше нечего и другой конец соединения закрыт.

buffer должен быть доступным для записи объектом, подобным байтам. Если задан offset, сообщение записывается в буфер начиная с этой позиции. Смещение должно быть неотрицательным целым числом, меньшим длины buffer (в байтах).

Если буфер слишком короткий, возникает исключение BufferTooShort, а полное сообщение доступно как e.args[0], где e — экземпляр исключения.

Изменено в версии 3.3: Теперь объекты соединения можно передавать между процессами с помощью Connection.send() и Connection.recv().

Теперь объекты соединения также поддерживают протокол менеджера контекста — см. раздел Типы менеджеров контекста. __enter__() возвращает объект соединения, а __exit__() вызывает close().

Например:

>>> from multiprocessing import Pipe
>>> a, b = Pipe()
>>> a.send([1, 'hello', None])
>>> b.recv()
[1, 'hello', None]
>>> b.send_bytes(b'thank you')
>>> a.recv_bytes()
b'thank you'
>>> import array
>>> arr1 = array.array('i', range(5))
>>> arr2 = array.array('i', [0] * 10)
>>> a.send_bytes(arr1)
>>> count = b.recv_bytes_into(arr2)
>>> assert count == len(arr1) * arr1.itemsize
>>> arr2
array('i', [0, 1, 2, 3, 4, 0, 0, 0, 0, 0])

Примитивы синхронизации

Как правило, примитивы синхронизации не так необходимы в многопроцессной программе, как в многопоточной. См. документацию по модулю threading.

Обратите внимание: примитивы синхронизации также можно создавать с помощью объекта-менеджера — см. раздел Менеджеры.

class multiprocessing.Barrier(parties[, action[, timeout]])

Объект барьера: копия threading.Barrier.

Добавлено в версии 3.3.

class multiprocessing.BoundedSemaphore([value])

Объект ограниченного семафора: близкий аналог threading.BoundedSemaphore.

Есть лишь одно отличие от его близкого аналога: первый аргумент метода acquire называется block, что соответствует Lock.acquire().

Примечание

В macOS этот объект неотличим от Semaphore, поскольку sem_getvalue() в этой платформе не реализован.

class multiprocessing.Condition([lock])

Условная переменная: псевдоним для threading.Condition.

Если задан lock, это должен быть объект Lock или RLock из multiprocessing.

Изменено в версии 3.3: Добавлен метод wait_for().

class multiprocessing.Event

Копия threading.Event.

class multiprocessing.Lock

Нерекурсивный объект блокировки: близкий аналог threading.Lock. После захвата блокировки процессом или потоком последующие попытки захватить её из любого процесса или потока будут блокироваться до её освобождения; освободить её может любой процесс или поток. Понятия и поведение threading.Lock, применимые к потокам, воспроизведены здесь в multiprocessing.Lock для процессов и потоков, за исключением отмеченных случаев.

Обратите внимание: Lock на самом деле является фабричной функцией, возвращающей экземпляр multiprocessing.synchronize.Lock, инициализированный контекстом по умолчанию.

Lock поддерживает протокол менеджера контекста, поэтому его можно использовать в инструкциях with.

acquire(block=True, timeout=None)

Захватывает блокировку в блокирующем или неблокирующем режиме.

Если аргумент block равен True (значение по умолчанию), вызов метода блокируется до тех пор, пока блокировка не станет свободной, затем устанавливает её в состояние блокировки и возвращает True. Обратите внимание: имя этого первого аргумента отличается от имени аргумента в threading.Lock.acquire().

Если аргумент block равен False, вызов метода не блокируется. Если блокировка уже захвачена, возвращается False; в противном случае блокировка захватывается и возвращается True.

Если для timeout указано положительное число с плавающей точкой, выполнение блокируется не дольше указанного в timeout количества секунд, пока не удастся захватить блокировку. Отрицательное значение timeout эквивалентно нулевому. Значение None (по умолчанию) задаёт бесконечный период ожидания. Обратите внимание: обработка отрицательных значений или значений None для timeout отличается от реализованного поведения в threading.Lock.acquire(). Аргумент timeout не имеет практического значения, если аргумент block равен False, и поэтому игнорируется. Возвращает True, если блокировка захвачена, или False, если время ожидания истекло.

release()

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

Поведение такое же, как у threading.Lock.release(), за исключением того, что при вызове для незахваченной блокировки возникает исключение ValueError.

class multiprocessing.RLock

Рекурсивный объект блокировки: близкий аналог threading.RLock. Рекурсивную блокировку должен освобождать процесс или поток, который её захватил. После того как процесс или поток захватил рекурсивную блокировку, тот же процесс или поток может повторно захватывать её без блокировки; при этом он должен освобождать её столько раз, сколько раз захватил.

Обратите внимание: RLock на самом деле является фабричной функцией, возвращающей экземпляр multiprocessing.synchronize.RLock, инициализированный контекстом по умолчанию.

RLock поддерживает протокол менеджера контекста, поэтому его можно использовать в инструкциях with.

acquire(block=True, timeout=None)

Захватывает блокировку в блокирующем или неблокирующем режиме.

Если аргумент block равен True, выполнение блокируется до тех пор, пока блокировка не станет свободной (то есть не будет принадлежать ни одному процессу или потоку), если только она уже не принадлежит текущему процессу или потоку. Затем текущий процесс или поток становится владельцем блокировки (если ещё им не является), а уровень рекурсии блокировки увеличивается на единицу; метод возвращает True. Обратите внимание: поведение этого первого аргумента отличается от реализации threading.RLock.acquire() по нескольким причинам, начиная с самого имени аргумента.

Если аргумент block равен False, блокировка не выполняется. Если блокировка уже захвачена (и, следовательно, принадлежит) другому процессу или потоку, текущий процесс или поток не становится её владельцем, а уровень рекурсии не меняется; метод возвращает False. Если блокировка свободна, текущий процесс или поток становится её владельцем, уровень рекурсии увеличивается, а метод возвращает True.

Использование аргумента timeout и его поведение такие же, как у Lock.acquire(). Обратите внимание: некоторые особенности поведения timeout отличаются от реализованных в threading.RLock.acquire().

release()

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

Вызывайте этот метод, только если вызывающий процесс или поток владеет блокировкой. Если метод вызывает процесс или поток, не являющийся владельцем, либо блокировка свободна (не принадлежит никому), возникает исключение AssertionError. Обратите внимание: тип исключения, возникающего в этой ситуации, отличается от реализованного поведения в threading.RLock.release().

class multiprocessing.Semaphore([value])

Объект семафора: близкий аналог threading.Semaphore.

Есть лишь одно отличие от его близкого аналога: первый аргумент метода acquire называется block, что соответствует Lock.acquire().

Общие объекты ctypes

Общие объекты можно создавать с помощью разделяемой памяти, наследуемой дочерними процессами.

multiprocessing.Value(typecode_or_type, *args, lock=True)

Возвращает объект ctypes, выделенный в разделяемой памяти. По умолчанию фактически возвращаемое значение представляет собой синхронизированную обёртку объекта. Доступ к самому объекту можно получить через атрибут value объекта Value.

typecode_or_type определяет тип возвращаемого объекта: это может быть тип ctypes или однобуквенный код типа, используемый модулем array. *args передаются конструктору типа.

Если lock — True (значение по умолчанию), для синхронизации доступа к значению создаётся новая рекурсивная блокировка. Если lock — объект Lock или RLock, он будет использоваться для синхронизации доступа к значению. Если lock — False, доступ к возвращаемому объекту не будет автоматически защищён блокировкой, поэтому он не обязательно будет «безопасен для процессов».

Такие операции, как +=, включающие чтение и запись, не являются атомарными. Поэтому, если, например, вам нужно атомарно увеличить общее значение, недостаточно просто выполнить

counter.value += 1

Если связанная блокировка является рекурсивной (а по умолчанию это так), вместо этого можно выполнить

with counter.get_lock():
    counter.value += 1

Обратите внимание, что lock — это аргумент, который можно передать только по имени.

multiprocessing.Array(typecode_or_type, size_or_initializer, *, lock=True)

Возвращает массив ctypes, выделенный в разделяемой памяти. По умолчанию фактически возвращаемое значение представляет собой синхронизированную обёртку массива.

typecode_or_type определяет тип элементов возвращаемого массива: это может быть тип ctypes или однобуквенный код типа, используемый модулем array. Если size_or_initializer — целое число, оно определяет длину массива, который изначально будет заполнен нулями. В противном случае size_or_initializer — это последовательность, используемая для инициализации массива; её длина определяет длину массива.

Если lock — True (значение по умолчанию), для синхронизации доступа к значению создаётся новая блокировка. Если lock — объект Lock или RLock, он будет использоваться для синхронизации доступа к значению. Если lock — False, доступ к возвращаемому объекту не будет автоматически защищён блокировкой, поэтому он не обязательно будет «безопасен для процессов».

Обратите внимание, что lock — это аргумент, который можно передать только по имени.

Обратите внимание, что у массива ctypes.c_char есть атрибуты value и raw, позволяющие использовать его для хранения и получения строк.

from multiprocessing import Process, Lock
from multiprocessing.sharedctypes import Value, Array
from ctypes import Structure, c_double

class Point(Structure):
    _fields_ = [('x', c_double), ('y', c_double)]

def modify(n, x, s, A):
    n.value **= 2
    x.value **= 2
    s.value = s.value.upper()
    for a in A:
        a.x **= 2
        a.y **= 2

if __name__ == '__main__':
    lock = Lock()

    n = Value('i', 7)
    x = Value(c_double, 1.0/3.0, lock=False)
    s = Array('c', b'hello world', lock=lock)
    A = Array(Point, [(1.875,-6.25), (-5.75,2.0), (2.375,9.5)], lock=lock)

    p = Process(target=modify, args=(n, x, s, A))
    p.start()
    p.join()

    print(n.value)
    print(x.value)
    print(s.value)
    print([(a.x, a.y) for a in A])

49
0.1111111111111111
HELLO WORLD
[(3.515625, 39.0625), (33.0625, 4.0), (5.640625, 90.25)]

Менеджеры

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

multiprocessing.Manager()

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

Процессы менеджеров завершаются сразу после сборки мусора для них или выхода родительского процесса. Классы менеджеров определены в модуле multiprocessing.managers:

class multiprocessing.managers.BaseManager(address=None, authkey=None, serializer='pickle', ctx=None, *, shutdown_timeout=1.0)

Создаёт объект BaseManager.

После создания следует вызвать start() или get_server().serve_forever(), чтобы объект менеджера ссылался на запущенный процесс менеджера.

address — адрес, на котором процесс менеджера ожидает новые подключения. Если address равен None, выбирается произвольный адрес.

authkey — ключ аутентификации, используемый для проверки допустимости входящих подключений к серверному процессу. Если authkey равен None, используется current_process().authkey. В противном случае используется authkey, который должен быть байтовой строкой.

serializer должен быть равен 'pickle' (используется сериализация pickle) или 'xmlrpclib' (используется сериализация xmlrpc.client).

ctx — объект контекста или None (используется текущий контекст). См. функцию get_context().

shutdown_timeout — время ожидания в секундах завершения процесса, используемого менеджером, в методе shutdown(). Если время ожидания истекает, процесс принудительно завершается. Если и принудительное завершение занимает слишком много времени, процесс уничтожается.

Изменено в версии 3.11: Добавлен параметр shutdown_timeout.

start([initializer[, initargs]])

Запускает дочерний процесс для запуска менеджера. Если initializer не равен None, при запуске дочерний процесс вызовет initializer(*initargs).

get_server()

Возвращает объект Server, представляющий фактический сервер под управлением менеджера. Объект Server поддерживает метод serve_forever():

>>> from multiprocessing.managers import BaseManager
>>> manager = BaseManager(address=('', 50000), authkey=b'abc')
>>> server = manager.get_server()
>>> server.serve_forever()

Объект Server также имеет атрибут address.

connect()

Подключает локальный объект менеджера к удалённому процессу менеджера:

>>> from multiprocessing.managers import BaseManager
>>> m = BaseManager(address=('127.0.0.1', 50000), authkey=b'abc')
>>> m.connect()

shutdown()

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

Этот метод можно вызывать несколько раз.

register(typeid[, callable[, proxytype[, exposed[, method_to_typeid[, create_method]]]]])

Метод класса, который можно использовать для регистрации типа или вызываемого объекта в классе менеджера.

typeid — «идентификатор типа», используемый для идентификации определённого типа общего объекта. Это должна быть строка.

callable — вызываемый объект, используемый для создания объектов с этим идентификатором типа. Если экземпляр менеджера будет подключаться к серверу с помощью метода connect() или если аргумент create_method равен False, этот параметр можно оставить равным None.

proxytype — подкласс BaseProxy, используемый для создания прокси общих объектов с этим typeid. Если указано None, класс прокси создаётся автоматически.

exposed используется для указания последовательности имён методов, к которым прокси этого typeid разрешено обращаться с помощью BaseProxy._callmethod(). (Если exposed равен None, вместо него используется proxytype._exposed_, если он существует.) Если список доступных методов не указан, будут доступны все «открытые методы» общего объекта. («Открытый метод» здесь означает любой атрибут, у которого есть метод __call__() и имя которого не начинается с '_'.)

method_to_typeid — сопоставление, используемое для указания типа возвращаемого значения тех открытых методов, которые должны возвращать прокси. Оно сопоставляет имена методов строковым идентификаторам typeid. (Если method_to_typeid равен None, вместо него используется proxytype._method_to_typeid_, если он существует.) Если имя метода не является ключом этого сопоставления или если сопоставление равно None, объект, возвращённый методом, будет скопирован по значению.

create_method определяет, следует ли создать метод с именем typeid, который можно использовать, чтобы указать серверному процессу создать новый общий объект и вернуть прокси для него. По умолчанию это True.

Экземпляры BaseManager также имеют одно свойство только для чтения:

address

Адрес, используемый менеджером.

Изменено в версии 3.3: Объекты менеджеров поддерживают протокол управления контекстом — см. раздел Типы контекстных менеджеров. __enter__() запускает серверный процесс (если он ещё не запущен), а затем возвращает объект менеджера. __exit__() вызывает shutdown().

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

class multiprocessing.managers.SyncManager

Подкласс BaseManager, который можно использовать для синхронизации процессов. Объекты этого типа возвращаются функцией multiprocessing.Manager().

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

Barrier(parties[, action[, timeout]])

Создаёт общий объект threading.Barrier и возвращает прокси для него.

Добавлено в версии 3.3.

BoundedSemaphore([value])

Создаёт общий объект threading.BoundedSemaphore и возвращает прокси для него.

Condition([lock])

Создаёт общий объект threading.Condition и возвращает прокси для него.

Если указан lock, он должен быть прокси объекта threading.Lock или threading.RLock.

Изменено в версии 3.3: Добавлен метод wait_for().

Event()

Создаёт общий объект threading.Event и возвращает прокси для него.

Lock()

Создаёт общий объект threading.Lock и возвращает прокси для него.

Namespace()

Создаёт общий объект Namespace и возвращает прокси для него.

Queue([maxsize])

Создаёт общий объект queue.Queue и возвращает прокси для него.

RLock()

Создаёт общий объект threading.RLock и возвращает прокси для него.

Semaphore([value])

Создаёт общий объект threading.Semaphore и возвращает прокси для него.

Array(typecode, sequence)

Создаёт массив и возвращает прокси для него.

Value(typecode, value)

Создаёт объект с доступным для записи атрибутом value и возвращает прокси для него.

dict()

dict(mapping)

dict(sequence)

Создаёт общий объект dict и возвращает прокси для него.

list()

list(sequence)

Создаёт общий объект list и возвращает прокси для него.

Изменено в версии 3.6: Общие объекты могут быть вложенными. Например, общий контейнер, такой как общий список, может содержать другие общие объекты, которыми будет управлять и которые будет синхронизировать SyncManager.

class multiprocessing.managers.Namespace

Тип, который можно зарегистрировать в SyncManager.

Объект пространства имён не имеет открытых методов, но имеет доступные для записи атрибуты. Его представление показывает значения атрибутов.

Однако при использовании прокси объекта пространства имён атрибут, имя которого начинается с '_', будет атрибутом прокси, а не целевого объекта:

>>> manager = multiprocessing.Manager()
>>> Global = manager.Namespace()
>>> Global.x = 10
>>> Global.y = 'hello'
>>> Global._z = 12.3    # this is an attribute of the proxy
>>> print(Global)
Namespace(x=10, y='hello')

from multiprocessing.managers import BaseManager

class MathsClass:
    def add(self, x, y):
        return x + y
    def mul(self, x, y):
        return x * y

class MyManager(BaseManager):
    pass

MyManager.register('Maths', MathsClass)

if __name__ == '__main__':
    with MyManager() as manager:
        maths = manager.Maths()
        print(maths.add(4, 3))         # prints 7
        print(maths.mul(7, 8))         # prints 56

>>> from multiprocessing.managers import BaseManager
>>> from queue import Queue
>>> queue = Queue()
>>> class QueueManager(BaseManager): pass
>>> QueueManager.register('get_queue', callable=lambda:queue)
>>> m = QueueManager(address=('', 50000), authkey=b'abracadabra')
>>> s = m.get_server()
>>> s.serve_forever()

>>> from multiprocessing.managers import BaseManager
>>> class QueueManager(BaseManager): pass
>>> QueueManager.register('get_queue')
>>> m = QueueManager(address=('foo.bar.org', 50000), authkey=b'abracadabra')
>>> m.connect()
>>> queue = m.get_queue()
>>> queue.put('hello')

>>> from multiprocessing.managers import BaseManager
>>> class QueueManager(BaseManager): pass
>>> QueueManager.register('get_queue')
>>> m = QueueManager(address=('foo.bar.org', 50000), authkey=b'abracadabra')
>>> m.connect()
>>> queue = m.get_queue()
>>> queue.get()
'hello'

>>> from multiprocessing import Process, Queue
>>> from multiprocessing.managers import BaseManager
>>> class Worker(Process):
...     def __init__(self, q):
...         self.q = q
...         super().__init__()
...     def run(self):
...         self.q.put('local hello')
...
>>> queue = Queue()
>>> w = Worker(queue)
>>> w.start()
>>> class QueueManager(BaseManager): pass
...
>>> QueueManager.register('get_queue', callable=lambda: queue)
>>> m = QueueManager(address=('', 50000), authkey=b'abracadabra')
>>> s = m.get_server()
>>> s.serve_forever()

Прокси-объекты

Прокси — это объект, который ссылается на общий объект, находящийся (предположительно) в другом процессе. Общий объект называется исходным объектом прокси. Несколько прокси-объектов могут иметь один и тот же исходный объект.

Прокси-объект имеет методы, вызывающие соответствующие методы исходного объекта (хотя через прокси не обязательно доступны все методы исходного объекта). Таким образом, прокси можно использовать так же, как и исходный объект:

>>> from multiprocessing import Manager
>>> manager = Manager()
>>> l = manager.list([i*i for i in range(10)])
>>> print(l)
[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
>>> print(repr(l))
<ListProxy object, typeid 'list' at 0x...>
>>> l[4]
16
>>> l[2:5]
[4, 9, 16]

Обратите внимание: применение str() к прокси возвращает представление исходного объекта, тогда как применение repr() возвращает представление самого прокси.

Важная особенность прокси-объектов состоит в том, что их можно сериализовать с помощью pickle и передавать между процессами. Поэтому исходный объект может содержать прокси-объекты. Это позволяет вкладывать друг в друга управляемые списки, словари и другие прокси-объекты:

>>> a = manager.list()
>>> b = manager.list()
>>> a.append(b)         # referent of a now contains referent of b
>>> print(a, b)
[<ListProxy object, typeid 'list' at ...>] []
>>> b.append('hello')
>>> print(a[0], b)
['hello'] ['hello']

Аналогичным образом прокси словарей и списков можно вкладывать друг в друга:

>>> l_outer = manager.list([ manager.dict() for i in range(2) ])
>>> d_first_inner = l_outer[0]
>>> d_first_inner['a'] = 1
>>> d_first_inner['b'] = 2
>>> l_outer[1]['c'] = 3
>>> l_outer[1]['z'] = 26
>>> print(l_outer[0])
{'a': 1, 'b': 2}
>>> print(l_outer[1])
{'c': 3, 'z': 26}

Если в исходном объекте содержатся стандартные (не являющиеся прокси) объекты list или dict, изменения этих изменяемых значений не будут переданы через менеджер, поскольку прокси не может узнать, когда изменяются содержащиеся в них значения. Однако сохранение значения в прокси-контейнере (что вызывает __setitem__ для объекта-прокси) передаёт изменение через менеджер. Поэтому для эффективного изменения такого элемента можно повторно присвоить изменённое значение прокси-контейнеру:

# create a list proxy and append a mutable object (a dictionary)
lproxy = manager.list()
lproxy.append({})
# now mutate the dictionary
d = lproxy[0]
d['a'] = 1
d['b'] = 2
# at this point, the changes to d are not yet synced, but by
# updating the dictionary, the proxy is notified of the change
lproxy[0] = d

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

>>> manager.list([1,2,3]) == [1,2,3]
False

class multiprocessing.managers.BaseProxy

Прокси-объекты являются экземплярами подклассов BaseProxy.

_callmethod(methodname[, args[, kwds]])

Вызвать метод исходного объекта прокси и вернуть результат.

Если proxy — это прокси, исходным объектом которого является obj, то выражение

proxy._callmethod(methodname, args, kwds)

вычислит выражение

getattr(obj, methodname)(*args, **kwds)

в процессе менеджера.

Возвращаемое значение будет копией результата вызова или прокси для нового общего объекта — см. документацию аргумента method_to_typeid метода BaseManager.register().

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

Обратите особое внимание: исключение будет вызвано, если methodname не был открыт для доступа.

Пример использования _callmethod():

>>> l = manager.list(range(10))
>>> l._callmethod('__len__')
10
>>> l._callmethod('__getitem__', (slice(2, 7),)) # equivalent to l[2:7]
[2, 3, 4, 5, 6]
>>> l._callmethod('__getitem__', (20,))          # equivalent to l[20]
Traceback (most recent call last):
...
IndexError: list index out of range

_getvalue()

Вернуть копию исходного объекта.

Если исходный объект нельзя сериализовать с помощью pickle, будет вызвано исключение.

__repr__()

Вернуть представление объекта-прокси.

__str__()

Вернуть представление исходного объекта.

Пулы процессов

С помощью класса Pool можно создать пул процессов, которые будут выполнять отправленные им задачи.

class multiprocessing.pool.Pool([processes[, initializer[, initargs[, maxtasksperchild[, context]]]]])

Объект пула процессов управляет пулом рабочих процессов, которым можно отправлять задания. Он поддерживает асинхронное получение результатов с тайм-аутами и обратными вызовами, а также параллельную реализацию map.

processes — количество используемых рабочих процессов. Если processes равно None, используется число, возвращаемое функцией os.cpu_count().

Если initializer не равно None, каждый рабочий процесс вызовет initializer(*initargs) при запуске.

maxtasksperchild — количество задач, которое рабочий процесс может выполнить, прежде чем завершится и будет заменён новым рабочим процессом. Это позволяет освобождать неиспользуемые ресурсы. По умолчанию maxtasksperchild равно None, то есть рабочие процессы работают столько же, сколько существует пул.

Параметр context позволяет указать контекст, используемый для запуска рабочих процессов. Обычно пул создаётся с помощью функции multiprocessing.Pool() или метода Pool() объекта контекста. В обоих случаях для context устанавливается соответствующее значение.

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

Предупреждение

Объекты multiprocessing.pool используют внутренние ресурсы, которыми необходимо надлежащим образом управлять (как и любыми другими ресурсами): использовать пул как менеджер контекста либо вручную вызывать close() и terminate(). Если этого не сделать, при завершении программа может зависнуть.

Не следует полагаться на сборщик мусора при уничтожении пула: CPython не гарантирует вызов финализатора пула (подробности см. в object.__del__()).

Изменено в версии 3.2: Добавлен параметр maxtasksperchild.

Изменено в версии 3.4: Добавлен параметр context.

Примечание

Рабочие процессы в Pool обычно работают в течение всего времени, пока в очереди пула есть задания. В других системах (например, Apache, mod_wsgi и т. д.) часто используется шаблон, позволяющий освобождать ресурсы, занятые рабочими процессами: рабочему процессу пула разрешают выполнить лишь определённое количество работы, после чего его завершают, очищают и запускают новый процесс взамен старого. Аргумент maxtasksperchild класса Pool предоставляет эту возможность конечному пользователю.

apply(func[, args[, kwds]])

Вызвать func с аргументами args и именованными аргументами kwds. Метод блокирует выполнение до получения результата. Поскольку вызов блокирующий, для выполнения работы параллельно лучше использовать apply_async(). Кроме того, func выполняется только одним из рабочих процессов пула.

apply_async(func[, args[, kwds[, callback[, error_callback]]]])

Вариант метода apply(), возвращающий объект AsyncResult.

Если указан callback, он должен быть вызываемым объектом, принимающим один аргумент. Когда результат будет готов, к нему будет применён callback, если только вызов не завершился ошибкой; в этом случае вместо него будет применён error_callback.

Если указан error_callback, он должен быть вызываемым объектом, принимающим один аргумент. Если целевая функция завершится ошибкой, error_callback будет вызван с экземпляром исключения.

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

map(func, iterable[, chunksize])

Параллельный аналог встроенной функции map() (поддерживается только один аргумент iterable; для нескольких итерируемых объектов см. starmap()). Метод блокирует выполнение до получения результата.

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

Обратите внимание: для очень длинных итерируемых объектов метод может потреблять много памяти. Для повышения эффективности рассмотрите возможность использования imap() или imap_unordered() с явно заданным параметром chunksize.

map_async(func, iterable[, chunksize[, callback[, error_callback]]])

Вариант метода map(), возвращающий объект AsyncResult.

Если указан callback, он должен быть вызываемым объектом, принимающим один аргумент. Когда результат будет готов, к нему будет применён callback, если только вызов не завершился ошибкой; в этом случае вместо него будет применён error_callback.

Если указан error_callback, он должен быть вызываемым объектом, принимающим один аргумент. Если целевая функция завершится ошибкой, error_callback будет вызван с экземпляром исключения.

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

imap(func, iterable[, chunksize])

Более ленивый вариант map().

Аргумент chunksize имеет то же значение, что и в методе map(). Для очень длинных итерируемых объектов большое значение chunksize может значительно ускорить выполнение задачи по сравнению со значением по умолчанию 1.

Кроме того, если chunksize равно 1, метод next() итератора, возвращаемого методом imap(), принимает необязательный параметр timeout: next(timeout) вызовет исключение multiprocessing.TimeoutError, если результат не удастся вернуть в течение указанного количества секунд timeout.

imap_unordered(func, iterable[, chunksize])

То же, что и imap(), за исключением того, что порядок результатов, возвращаемых итератором, считается произвольным. (Порядок гарантированно будет «правильным» только при наличии одного рабочего процесса.)

starmap(func, iterable[, chunksize])

Похож на map(), но элементы iterable должны быть итерируемыми объектами, которые распаковываются в аргументы.

Таким образом, итерируемый объект из [(1,2), (3, 4)] даёт результат [func(1,2), func(3,4)].

Добавлено в версии 3.3.

starmap_async(func, iterable[, chunksize[, callback[, error_callback]]])

Сочетает starmap() и map_async(): перебирает итерируемый объект iterable, состоящий из итерируемых объектов, и вызывает func, распаковывая эти объекты в аргументы. Возвращает объект результата.

Добавлено в версии 3.3.

close()

Запрещает отправлять в пул новые задачи. После выполнения всех задач рабочие процессы завершатся.

terminate()

Немедленно останавливает рабочие процессы, не дожидаясь завершения текущей работы. При удалении объекта пула сборщиком мусора метод terminate() будет вызван немедленно.

join()

Ожидать завершения рабочих процессов. Перед вызовом join() необходимо вызвать close() или terminate().

Изменено в версии 3.3: Объекты пулов теперь поддерживают протокол менеджера контекста — см. Типы менеджеров контекста. Метод __enter__() возвращает объект пула, а __exit__() вызывает terminate().

class multiprocessing.pool.AsyncResult

Класс результата, возвращаемого методами Pool.apply_async() и Pool.map_async().

get([timeout])

Вернуть результат, когда он будет получен. Если timeout не равно None и результат не будет получен в течение указанного количества секунд timeout, вызывается исключение multiprocessing.TimeoutError. Если удалённый вызов вызвал исключение, оно будет повторно вызвано методом get().

wait([timeout])

Ожидать появления результата или истечения времени timeout в секундах.

ready()

Вернуть признак того, что вызов завершён.

successful()

Вернуть признак того, что вызов завершился без исключения. Если результат ещё не готов, будет вызвано исключение ValueError.

Изменено в версии 3.7: Если результат ещё не готов, вызывается исключение ValueError вместо AssertionError.

В следующем примере показано использование пула:

from multiprocessing import Pool
import time

def f(x):
    return x*x

if __name__ == '__main__':
    with Pool(processes=4) as pool:         # start 4 worker processes
        result = pool.apply_async(f, (10,)) # evaluate "f(10)" asynchronously in a single process
        print(result.get(timeout=1))        # prints "100" unless your computer is *very* slow

        print(pool.map(f, range(10)))       # prints "[0, 1, 4,..., 81]"

        it = pool.imap(f, range(10))
        print(next(it))                     # prints "0"
        print(next(it))                     # prints "1"
        print(it.next(timeout=1))           # prints "4" unless your computer is *very* slow

        result = pool.apply_async(time.sleep, (10,))
        print(result.get(timeout=1))        # raises multiprocessing.TimeoutError

Слушатели и клиенты

Обычно обмен сообщениями между процессами выполняется с помощью очередей или объектов Connection, возвращаемых функцией Pipe().

Однако модуль multiprocessing.connection предоставляет дополнительную гибкость. По сути, он предоставляет высокоуровневый API для обмена сообщениями через сокеты или именованные каналы Windows. Он также поддерживает аутентификацию с дайджестом с помощью модуля hmac и опрос нескольких соединений одновременно.

multiprocessing.connection.deliver_challenge(connection, authkey)

Отправляет на другой конец соединения случайно сгенерированное сообщение и ожидает ответа.

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

multiprocessing.connection.answer_challenge(connection, authkey)

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

Если приветственное сообщение не получено, возникает исключение AuthenticationError.

multiprocessing.connection.Client(address[, family[, authkey]])

Пытается установить соединение со слушателем, использующим адрес address, и возвращает объект Connection.

Тип соединения определяется аргументом family, однако обычно его можно опустить, поскольку, как правило, его можно определить по формату address. (См. Форматы адресов.)

Если задано значение authkey, отличное от None, оно должно быть байтовой строкой и будет использоваться как секретный ключ для проверки подлинности на основе HMAC. Если значение authkey равно None, проверка подлинности не выполняется. Если проверка подлинности не пройдена, возникает исключение AuthenticationError. См. Ключи аутентификации.

class multiprocessing.connection.Listener([address[, family[, backlog[, authkey]]]])

Обёртка для связанного сокета или именованного канала Windows, ожидающего («слушающего») подключений.

address — адрес связанного сокета или именованного канала объекта слушателя.

Примечание

Если используется адрес «0.0.0.0», в Windows он не будет доступной для подключения конечной точкой. Если вам нужна конечная точка, к которой можно подключиться, используйте «127.0.0.1».

family — тип используемого сокета (или именованного канала). Это может быть одна из строк 'AF_INET' (для TCP-сокета), 'AF_UNIX' (для сокета домена Unix) или 'AF_PIPE' (для именованного канала Windows). Из них гарантированно доступна только первая. Если family равен None, семейство определяется по формату address. Если address также равен None, выбирается значение по умолчанию. По умолчанию используется семейство, которое считается самым быстрым из доступных. См. Форматы адресов. Обратите внимание: если family равен 'AF_UNIX', а address равен None, сокет будет создан в закрытом временном каталоге, созданном с помощью tempfile.mkstemp().

Если объект слушателя использует сокет, значение backlog (по умолчанию 1) передаётся методу listen() сокета после его привязки.

Если задано значение authkey, отличное от None, оно должно быть байтовой строкой и будет использоваться как секретный ключ для проверки подлинности на основе HMAC. Если значение authkey равно None, проверка подлинности не выполняется. Если проверка подлинности не пройдена, возникает исключение AuthenticationError. См. Ключи аутентификации.

accept()

Принимает соединение через связанный сокет или именованный канал объекта слушателя и возвращает объект Connection. Если выполняется проверка подлинности и она не пройдена, возникает исключение AuthenticationError.

close()

Закрывает связанный сокет или именованный канал объекта слушателя. Этот метод вызывается автоматически при удалении объекта слушателя сборщиком мусора. Тем не менее рекомендуется вызывать его явно.

Объекты слушателя имеют следующие свойства только для чтения:

address

Адрес, используемый объектом Listener.

last_accepted

Адрес, с которого поступило последнее принятое соединение. Если он недоступен, значение равно None.

Изменено в версии 3.3: Объекты Listener теперь поддерживают протокол менеджера контекста — см. Типы менеджеров контекста. Метод __enter__() возвращает объект слушателя, а метод __exit__() вызывает close().

multiprocessing.connection.wait(object_list, timeout=None)

Ожидает готовности одного из объектов в object_list. Возвращает список готовых объектов из object_list. Если timeout — число с плавающей точкой, вызов блокируется не дольше указанного количества секунд. Если timeout равен None, блокировка будет продолжаться неограниченно долго. Отрицательное значение тайм-аута эквивалентно нулевому.

В Unix и Windows объект может присутствовать в object_list, если он является

• доступным для чтения объектом Connection;
• подключённым и доступным для чтения объектом socket.socket; или
• атрибутом sentinel объекта Process.

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

Unix: wait(object_list, timeout) почти эквивалентен select.select(object_list, [], [], timeout). Разница в том, что если select.select() прерывается сигналом, он может вызвать исключение OSError с номером ошибки EINTR, тогда как wait() этого не сделает.

Windows: элемент в object_list должен быть либо целочисленным дескриптором, допускающим ожидание (согласно определению в документации функции Win32 WaitForMultipleObjects()), либо объектом с методом fileno(), который возвращает дескриптор сокета или канала. (Обратите внимание, что дескрипторы каналов и сокетов не допускают ожидание.)

Добавлено в версии 3.3.

Примеры

Следующий код сервера создаёт слушателя, использующего 'secret password' в качестве ключа аутентификации. Затем он ожидает подключения и отправляет клиенту данные:

from multiprocessing.connection import Listener
from array import array

address = ('localhost', 6000)     # family is deduced to be 'AF_INET'

with Listener(address, authkey=b'secret password') as listener:
    with listener.accept() as conn:
        print('connection accepted from', listener.last_accepted)

        conn.send([2.25, None, 'junk', float])

        conn.send_bytes(b'hello')

        conn.send_bytes(array('i', [42, 1729]))

Следующий код подключается к серверу и получает от него данные:

from multiprocessing.connection import Client
from array import array

address = ('localhost', 6000)

with Client(address, authkey=b'secret password') as conn:
    print(conn.recv())                  # => [2.25, None, 'junk', float]

    print(conn.recv_bytes())            # => 'hello'

    arr = array('i', [0, 0, 0, 0, 0])
    print(conn.recv_bytes_into(arr))    # => 8
    print(arr)                          # => array('i', [42, 1729, 0, 0, 0])

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

import time, random
from multiprocessing import Process, Pipe, current_process
from multiprocessing.connection import wait

def foo(w):
    for i in range(10):
        w.send((i, current_process().name))
    w.close()

if __name__ == '__main__':
    readers = []

    for i in range(4):
        r, w = Pipe(duplex=False)
        readers.append(r)
        p = Process(target=foo, args=(w,))
        p.start()
        # We close the writable end of the pipe now to be sure that
        # p is the only process which owns a handle for it.  This
        # ensures that when p closes its handle for the writable end,
        # wait() will promptly report the readable end as being ready.
        w.close()

    while readers:
        for r in wait(readers):
            try:
                msg = r.recv()
            except EOFError:
                readers.remove(r)
            else:
                print(msg)

Ключи аутентификации

При использовании Connection.recv полученные данные автоматически десериализуются с помощью pickle. К сожалению, десериализация данных из ненадёжного источника представляет угрозу безопасности. Поэтому Listener и Client() используют модуль hmac для проверки подлинности с помощью дайджеста.

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

Если запрошена проверка подлинности, но ключ аутентификации не указан, используется возвращаемое значение current_process().authkey (см. Process). Это значение автоматически наследуется любым объектом Process, созданным текущим процессом. Это означает, что (по умолчанию) все процессы многопроцессной программы будут использовать один общий ключ аутентификации, который можно применять при установлении соединений между ними.

Подходящие ключи аутентификации также можно сгенерировать с помощью os.urandom().

Ведение журнала

Предусмотрена некоторая поддержка ведения журнала. Однако обратите внимание, что пакет logging не использует блокировки, общие для процессов, поэтому сообщения из разных процессов могут перемешиваться (в зависимости от типа обработчика).

multiprocessing.get_logger()

Возвращает средство ведения журнала, используемое multiprocessing. При необходимости создаётся новое средство.

При первом создании уровень средства ведения журнала равен logging.NOTSET, а обработчик по умолчанию отсутствует. Сообщения, отправленные этому средству, по умолчанию не передаются корневому средству ведения журнала.

Обратите внимание: в Windows дочерние процессы наследуют только уровень средства ведения журнала родительского процесса — другие настройки средства не наследуются.

multiprocessing.log_to_stderr(level=None)

Эта функция вызывает get_logger(), но, помимо возврата средства ведения журнала, созданного функцией get_logger, добавляет обработчик, который выводит данные в sys.stderr в формате '[%(levelname)s/%(processName)s] %(message)s'. Уровень levelname средства ведения журнала можно изменить, передав аргумент level.

Ниже приведён пример сеанса с включённым ведением журнала:

>>> import multiprocessing, logging
>>> logger = multiprocessing.log_to_stderr()
>>> logger.setLevel(logging.INFO)
>>> logger.warning('doomed')
[WARNING/MainProcess] doomed
>>> m = multiprocessing.Manager()
[INFO/SyncManager-...] child process calling self.run()
[INFO/SyncManager-...] created temp directory /.../pymp-...
[INFO/SyncManager-...] manager serving at '/.../listener-...'
>>> del m
[INFO/MainProcess] sending shutdown message to manager
[INFO/SyncManager-...] manager exiting with exitcode 0

Модуль multiprocessing.dummy

multiprocessing.dummy воспроизводит API модуля multiprocessing, но представляет собой лишь обёртку вокруг модуля threading.

В частности, функция Pool, предоставляемая модулем multiprocessing.dummy, возвращает экземпляр ThreadPool — подкласса Pool, поддерживающего те же вызовы методов, но использующего пул рабочих потоков вместо рабочих процессов.

class multiprocessing.pool.ThreadPool([processes[, initializer[, initargs]]])

Объект пула потоков управляет пулом рабочих потоков, которым можно передавать задания. Экземпляры ThreadPool полностью совместимы по интерфейсу с экземплярами Pool; их ресурсами также необходимо надлежащим образом управлять: использовать пул как менеджер контекста либо вручную вызывать методы close() и terminate().

processes — количество используемых рабочих потоков. Если processes равен None, используется значение, возвращаемое функцией os.cpu_count().

Если initializer не равен None, каждый рабочий процесс при запуске вызывает initializer(*initargs).

В отличие от Pool, аргументы maxtasksperchild и context задавать нельзя.

Примечание

ThreadPool использует тот же интерфейс, что и Pool, который изначально создавался для пула процессов, ещё до появления модуля concurrent.futures. Поэтому он наследует некоторые операции, не имеющие смысла для пула на основе потоков, и использует собственный тип для представления состояния асинхронных заданий — AsyncResult, который не распознаётся другими библиотеками.

В большинстве случаев пользователям следует предпочесть concurrent.futures.ThreadPoolExecutor: он имеет более простой интерфейс, изначально разработанный для потоков, и возвращает экземпляры concurrent.futures.Future, совместимые со многими другими библиотеками, включая asyncio.

Все методы запуска

Следующие рекомендации применимы ко всем методам запуска.

Избегайте общего состояния

По возможности старайтесь не передавать большие объёмы данных между процессами.

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

Возможность сериализации с помощью pickle

Убедитесь, что аргументы методов прокси-серверов можно сериализовать с помощью pickle.

Потокобезопасность прокси-серверов

Не используйте объект прокси из нескольких потоков, если не защищаете его блокировкой.

(Если разные процессы используют один и тот же прокси, проблем не возникает.)

Присоединение процессов-зомби

В Unix завершившийся процесс, к которому не был присоединён родительский процесс, становится процессом-зомби. Обычно таких процессов не должно быть много, поскольку при каждом запуске нового процесса (или вызове active_children()) выполняется присоединение всех завершённых процессов, к которым оно ещё не было выполнено. Вызов метода Process.is_alive завершившегося процесса также присоединяет его. Тем не менее рекомендуется явно присоединять все запущенные процессы.

Наследование предпочтительнее сериализации и десериализации с помощью pickle

При использовании методов запуска spawn или forkserver многие типы из multiprocessing должны поддерживать сериализацию с помощью pickle, чтобы дочерние процессы могли их использовать. Однако, как правило, не следует передавать общие объекты другим процессам через каналы или очереди. Вместо этого организуйте программу так, чтобы процесс, которому нужен доступ к общему ресурсу, созданному в другом месте, мог унаследовать его от процесса-предка.

Избегайте принудительного завершения процессов

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

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

Присоединение процессов, использующих очереди

Помните, что процесс, поместивший элементы в очередь, будет ожидать перед завершением, пока все буферизованные элементы не будут переданы «потоком-поставщиком» в базовый канал. (Чтобы избежать такого поведения, дочерний процесс может вызвать метод очереди Queue.cancel_join_thread.)

Это означает, что при использовании очереди необходимо убедиться, что все помещённые в неё элементы будут извлечены до присоединения процесса. Иначе нельзя быть уверенным, что процессы, поместившие элементы в очередь, завершатся. Помните также, что недемонические процессы будут присоединены автоматически.

Следующий пример приведёт к взаимоблокировке:

from multiprocessing import Process, Queue

def f(q):
    q.put('X' * 1000000)

if __name__ == '__main__':
    queue = Queue()
    p = Process(target=f, args=(queue,))
    p.start()
    p.join()                    # this deadlocks
    obj = queue.get()

Здесь проблему можно устранить, поменяв местами две последние строки (или просто удалив строку p.join()).

Явно передавайте ресурсы дочерним процессам

В Unix при использовании метода запуска fork дочерний процесс может обращаться к общему ресурсу, созданному в родительском процессе, через глобальный ресурс. Однако объект лучше передать аргументом конструктору дочернего процесса.

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

Например,

from multiprocessing import Process, Lock

def f():
    ... do something using "lock" ...

if __name__ == '__main__':
    lock = Lock()
    for i in range(10):
        Process(target=f).start()

следует переписать так:

from multiprocessing import Process, Lock

def f(l):
    ... do something using "l" ...

if __name__ == '__main__':
    lock = Lock()
    for i in range(10):
        Process(target=f, args=(lock,)).start()

Будьте осторожны, заменяя sys.stdin на «файлоподобный объект»

Изначально multiprocessing безусловно вызывал:

os.close(sys.stdin.fileno())

в методе multiprocessing.Process._bootstrap() — это приводило к проблемам при вложенных процессах. Поведение изменили на следующее:

sys.stdin.close()
sys.stdin = open(os.open(os.devnull, os.O_RDONLY), closefd=False)

Это устраняет основную проблему: столкновение процессов, приводившее к ошибке неверного файлового дескриптора. Однако появляется потенциальная опасность для приложений, которые заменяют sys.stdin() «файлоподобным объектом» с буферизацией вывода. Опасность заключается в том, что если несколько процессов вызывают close() для этого файлоподобного объекта, одни и те же данные могут несколько раз сбрасываться в объект, что приведёт к повреждению данных.

Если вы создаёте файлоподобный объект и реализуете собственное кэширование, его можно сделать безопасным для fork: сохраняйте PID при каждом добавлении данных в кэш и сбрасывайте кэш, если PID изменился. Например:

@property
def cache(self):
    pid = os.getpid()
    if pid != self._pid:
        self._pid = pid
        self._cache = []
    return self._cache

Дополнительные сведения см. в bpo-5155, bpo-5313 и bpo-5331.

Методы запуска spawn и forkserver

Для них действуют некоторые дополнительные ограничения, которые не применяются к методу запуска fork.

Дополнительные требования к сериализации с помощью pickle

Убедитесь, что все аргументы Process.__init__() можно сериализовать с помощью pickle. Кроме того, если вы создаёте подкласс Process, убедитесь, что его экземпляры можно сериализовать с помощью pickle при вызове метода Process.start.

Глобальные переменные

Имейте в виду: если код дочернего процесса пытается обратиться к глобальной переменной, её значение (если оно есть) может отличаться от значения в родительском процессе на момент вызова Process.start.

Однако глобальные переменные, являющиеся лишь константами на уровне модуля, проблем не вызывают.

Безопасный импорт главного модуля

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

Например, при использовании метода запуска spawn или forkserver запуск следующего модуля завершится ошибкой RuntimeError:

from multiprocessing import Process

def foo():
    print('hello')

p = Process(target=foo)
p.start()

Вместо этого следует защитить «точку входа» программы следующим образом, используя if __name__ == '__main__'::

from multiprocessing import Process, freeze_support, set_start_method

def foo():
    print('hello')

if __name__ == '__main__':
    freeze_support()
    set_start_method('spawn')
    p = Process(target=foo)
    p.start()

(Строку freeze_support() можно опустить, если программа будет запускаться обычным образом, а не в замороженном виде.)

Это позволяет только что запущенному интерпретатору Python безопасно импортировать модуль, а затем выполнить функцию foo() этого модуля.

Аналогичные ограничения действуют, если в главном модуле создаётся пул или менеджер.