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

Класс Process

В multiprocessing процессы создаются путём создания объекта Process и последующего вызова его метода start(). API класса 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__' см. в разделе Рекомендации по программированию.

Аргументы для Process обычно должны сериализоваться с помощью pickle в дочернем процессе. Если ввести приведённый выше пример непосредственно в REPL, это может привести к возникновению исключения AttributeError в дочернем процессе, который попытается найти функцию f в модуле __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 создаётся серверный процесс. После этого, когда требуется новый процесс, родительский процесс подключается к серверу и запрашивает создание нового процесса с помощью fork. Серверный процесс fork работает в одном потоке, если только системные библиотеки или предварительно загруженные импорты не создают потоки как побочный эффект, поэтому, как правило, использовать в нём 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.

Библиотеки, использующие multiprocessing или ProcessPoolExecutor, следует проектировать так, чтобы их пользователи могли передавать собственный контекст multiprocessing. Использование библиотекой собственного конкретного контекста может привести к несовместимости с остальной частью приложения пользователя библиотеки. Всегда документируйте, если для вашей библиотеки требуется определённый метод запуска.

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

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 представляют деятельность, выполняемую в отдельном процессе. Класс Process имеет аналоги всех методов класса threading.Thread.

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

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

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

Примечание

Как правило, все аргументы Process должны быть сериализуемыми с помощью pickle. Эта проблема часто возникает при попытке создать Process или использовать concurrent.futures.ProcessPoolExecutor из REPL, если целевая функция target определена локально.

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

Пример этой неперехватываемой ошибки в дочернем процессе:

>>> import multiprocessing as mp
>>> def knigit():
...     print("Ni!")
...
>>> process = mp.Process(target=knigit)
>>> process.start()
>>> Traceback (most recent call last):
  File ".../multiprocessing/spawn.py", line ..., in spawn_main
  File ".../multiprocessing/spawn.py", line ..., in _main
AttributeError: module '__main__' has no attribute 'knigit'
>>> process
<SpawnProcess name='SpawnProcess-1' pid=379473 parent=378707 stopped exitcode=1>

См. Методы запуска spawn и forkserver. Это ограничение не действует при использовании метода запуска "fork", однако начиная с Python 3.14 этот метод больше не используется по умолчанию ни на одной платформе. См. Контексты и методы запуска. См. также gh-132898.

Изменено в версии 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 — положительное число, метод блокируется не дольше указанного количества секунд. Обратите внимание: метод возвращает 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, этот метод может вызвать NotImplementedError, поскольку sem_getvalue() не реализован.

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

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

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

Фоновый поток завершит работу после сброса всех буферизованных данных в канал. Этот метод вызывается автоматически при сборке мусора для очереди.

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

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

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

Если количество процессоров не удаётся определить, возбуждается исключение NotImplementedError.

См. также

os.cpu_count() os.process_cpu_count()

Изменено в версии 3.13: Возвращаемое значение также можно переопределить с помощью флага -X cpu_count или переменной PYTHON_CPU_COUNT, поскольку эта функция является лишь обёрткой над API модуля os для получения количества процессоров.

multiprocessing.current_process()

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

Аналог функции threading.current_thread().

multiprocessing.parent_process()

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

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

multiprocessing.freeze_support()

Добавляет поддержку для программ, использующих multiprocessing, которые были заморожены для создания исполняемого файла. (Работа проверена с 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() не оказывает никакого эффекта, если метод запуска не равен spawn. Кроме того, если модуль выполняется обычным образом интерпретатором Python (программа не была заморожена), 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)

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

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

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

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

multiprocessing.set_start_method(method, force=False)

Задаёт метод, который следует использовать для запуска дочерних процессов. Аргумент method может быть равен 'fork', 'spawn' или 'forkserver'. Возбуждает исключение RuntimeError, если метод запуска уже установлен и force не равен True. Если 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]])

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

Изменено в версии 3.13: По умолчанию для processes используется os.process_cpu_count() вместо os.cpu_count().

Примечание

Рабочие процессы в 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(), чтобы одновременно ожидать сообщения от нескольких процессов:

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.process_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 для завершившегося процесса также выполняет присоединение. Тем не менее рекомендуется явно выполнять присоединение всех запущенных процессов.

Наследуйте, а не сериализуйте и десериализуйте

При использовании методов запуска 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 можно сериализовать с помощью pickle. Кроме того, если вы создаёте подкласс Process.__init__, необходимо убедиться, что его экземпляры можно сериализовать с помощью 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() этого модуля.

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