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.

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

fork

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

Доступен в системах POSIX. В настоящее время используется по умолчанию в POSIX, за исключением macOS.

Примечание

В Python 3.14 метод запуска по умолчанию перестанет быть fork. Если коду требуется fork, его следует явно указать с помощью get_context() или set_start_method().

Изменено в версии 3.12: Если Python может определить, что в процессе работает несколько потоков, функция os.fork(), которую этот метод запуска вызывает внутри себя, выдаст предупреждение DeprecationWarning. Используйте другой метод запуска. Дополнительные пояснения см. в документации по os.fork().

forkserver

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

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

В 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()

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

Каналы

Функция 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() (и другие). Обратите внимание, что данные в канале могут быть повреждены, если два процесса (или потока) одновременно попытаются прочитать данные из одного и того же конца канала или записать в него. Конечно, если процессы одновременно используют разные концы канала, риск повреждения данных отсутствует.

Метод 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__' (<class '_frozen_importlib.BuiltinImporter'>)>
AttributeError: Can't get attribute 'f' on <module '__main__' (<class '_frozen_importlib.BuiltinImporter'>)>
AttributeError: Can't get attribute 'f' on <module '__main__' (<class '_frozen_importlib.BuiltinImporter'>)>

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

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

Объекты процессов представляют действия, выполняемые в отдельном процессе. Класс 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. В POSIX это файловый дескриптор, который можно использовать с примитивами модуля select.

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

terminate()

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

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

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

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

kill()

То же, что и terminate(), но в POSIX используется сигнал SIGKILL.

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

close()

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

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

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

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

>>> import multiprocessing, time, signal
>>> mp_context = multiprocessing.get_context('spawn')
>>> p = mp_context.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(). В противном случае семафор, используемый для подсчёта количества незавершённых задач, может переполниться, что приведёт к исключению.

В отличие от других реализаций очередей в Python, очереди multiprocessing сериализуют все помещаемые в них объекты с помощью pickle. Объект, возвращаемый методом get, является воссозданным объектом и не использует общую память с исходным объектом.

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

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

multiprocessing.Pipe([duplex])

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

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

Метод send() сериализует объект с помощью pickle, а recv() воссоздаёт объект.

class multiprocessing.Queue([maxsize])

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

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

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

qsize()

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

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

empty()

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

Для закрытых очередей может возникнуть исключение OSError (не гарантируется).

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 в противном случае.

Если SimpleQueue закрыта, всегда вызывает исключение OSError.

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'. Не все платформы поддерживают все методы. См. раздел Контексты и методы запуска.

Добавлено в версии 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. См. раздел Контексты и методы запуска.

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

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

multiprocessing.set_executable(executable)

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

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

до создания дочерних процессов.

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

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

multiprocessing.set_forkserver_preload(module_names)

Задаёт список имён модулей, которые основной процесс сервера ветвления должен попытаться импортировать, чтобы их уже импортированное состояние наследовалось процессами, созданными с помощью ветвления. Любые исключения ImportError, возникшие при этом, игнорируются без уведомления. Это можно использовать для повышения производительности, чтобы избежать повторения одних и тех же операций в каждом процессе.

Чтобы это работало, функцию необходимо вызвать до запуска процесса сервера ветвления (до создания Pool или запуска Process).

Имеет смысл только при использовании метода запуска 'forkserver'. См. раздел Контексты и методы запуска.

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

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.

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

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

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

class multiprocessing.connection.Connection

send(obj)

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

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

recv()

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

fileno()

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

close()

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

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

poll([timeout])

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

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

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

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

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

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

recv_bytes([maxlength])

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

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

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

recv_bytes_into(buffer[, offset])

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

buffer должен быть доступным для записи объектом, подобным bytes. Если указан 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 типа float метод будет ожидать не дольше указанного в timeout количества секунд, пока блокировку нельзя будет захватить. Вызовы с отрицательным значением 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.

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

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

>>> mp_context = multiprocessing.get_context('spawn')
>>> manager = mp_context.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()

Объекты-прокси

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

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

>>> mp_context = multiprocessing.get_context('spawn')
>>> manager = mp_context.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 не был указан в exposed.

Пример использования _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 должны быть итерируемыми объектами, которые распаковываются в аргументы.

Таким образом, 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: Объекты Pool теперь поддерживают протокол управления контекстом — см. Типы менеджеров контекста. __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, блокировка действует неограниченно долго. Отрицательное значение тайм-аута эквивалентно нулевому.

В POSIX и Windows объект может находиться в object_list, если это

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

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

POSIX: 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

Полную таблицу уровней ведения журнала см. в модуле logging.

Модуль 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.

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

Не используйте объект-прокси более чем из одного потока, если только вы не защищаете его блокировкой.

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

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

В POSIX завершившийся процесс, к которому не выполнили присоединение, становится зомби. Их не должно быть много, поскольку при каждом запуске нового процесса (или вызове 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()).

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

В POSIX при использовании метода запуска 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() этого модуля.

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