multiprocessing – параллелизм на основе процессов¶multiprocessing – Process-based parallelism

Класс Process¶The Process class

В 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__', см. в руководстве по программированию.

Контексты и методы запуска¶Contexts and start methods

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

spawn

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

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

fork

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

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

forkserver

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

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

На POSIX использование методов запуска spawn или forkserver также запускает процесс resource tracker (отслеживатель ресурсов), который отслеживает удалённые именованные системные ресурсы (такие как именованные семафоры или объекты 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(), чтобы не мешать выбору пользователя библиотеки.

Обмен объектами между процессами¶Exchanging objects between processes

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())    # печатает "[42, None, 'hello']"
    p.join()

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

Каналы

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

from multiprocessing import Process, Pipe

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

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

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

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

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

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

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

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

Однако если действительно необходимо использовать некоторые общие данные, то 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]

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

Использование пула рабочих процессов¶Using a pool of workers

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

Например:

from multiprocessing import Pool, TimeoutError
import time
import os

def f(x):
    return x*x

if __name__ == '__main__':
    # запустить 4 рабочих процесса
    with Pool(processes=4) as pool:

        # печатает "[0, 1, 4,..., 81]"
        print(pool.map(f, range(10)))

        # печатает те же числа в произвольном порядке
        for i in pool.imap_unordered(f, range(10)):
            print(i)

        # вычисляет "f(20)" асинхронно
        res = pool.apply_async(f, (20,))      # выполняется *только* в одном процессе
        print(res.get(timeout=1))             # печатает "400"

        # вычисляет "os.getpid()" асинхронно
        res = pool.apply_async(os.getpid, ()) # выполняется *только* в одном процессе
        print(res.get(timeout=1))             # печатает PID этого процесса

        # запуск нескольких асинхронных вычислений *может* задействовать больше процессов
        multiple_results = [pool.apply_async(os.getpid, ()) for i in range(4)]
        print([res.get(timeout=1) for res in multiple_results])

        # усыпить один рабочий процесс на 10 секунд
        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")

    # выход из блока 'with' остановил пул
    print("Now the pool is closed and no longer available")

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

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

Process и исключения¶Process and exceptions

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

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

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

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

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

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

run()¶

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

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

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

Пример:

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

start()¶

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

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

join([timeout])¶

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

Процесс может быть присоединён много раз.

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

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 это дескриптор ОС, который можно использовать с семейством WaitForSingleObject и WaitForMultipleObjects вызовов API. В Unix это файловый дескриптор, который можно использовать с примитивами из модуля select.

Новое в версии 3.3.

terminate()¶

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

Обратите внимание, что процессы-потомки не будут завершены – они просто станут осиротевшими.

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

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

kill()¶

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

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

close()¶

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

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

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

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

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

exception multiprocessing.ProcessError¶

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

exception multiprocessing.BufferTooShort¶

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

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

exception multiprocessing.AuthenticationError¶

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

exception multiprocessing.TimeoutError¶

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

Каналы и очереди¶Pipes and Queues

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

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

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

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

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

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

multiprocessing.Pipe([duplex])¶

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

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

class multiprocessing.Queue([maxsize])¶

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

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

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

qsize()¶

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

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

empty()¶

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

full()¶

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

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

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

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

put_nowait(obj)¶

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

get([block[, timeout]])¶

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

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

get_nowait()¶

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

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

close()¶

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

join_thread()¶

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

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

cancel_join_thread()¶

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

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

Примечание

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

class multiprocessing.SimpleQueue¶

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

close()¶

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

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

Новое в версии 3.9.

empty()¶

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

get()¶

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

put(item)¶

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

class multiprocessing.JoinableQueue([maxsize])¶

JoinableQueue, подкласс Queue, – это очередь, которая дополнительно имеет методы task_done() и join().

task_done()¶

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

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

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

join()¶

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

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

Разное¶Miscellaneous

multiprocessing.active_children()¶

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

Вызов этой функции имеет побочный эффект – «присоединение» всех уже завершённых процессов.

multiprocessing.cpu_count()¶

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

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

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

См. также

os.cpu_count()

multiprocessing.current_process()¶

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

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

multiprocessing.parent_process()¶

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

Новое в версии 3.8.

multiprocessing.freeze_support()¶

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

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

from multiprocessing import Process, freeze_support

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

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

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

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

multiprocessing.get_all_start_methods()¶

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

Новое в версии 3.4.

multiprocessing.get_context(method=None)¶

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

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

Новое в версии 3.4.

multiprocessing.get_start_method(allow_none=False)¶

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

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

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

Новое в версии 3.4.

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

multiprocessing.set_executable(executable)¶

Устанавливает путь к интерпретатору Python, используемому при запуске дочернего процесса. (По умолчанию используется sys.executable). Тем, кто встраивает Python, вероятно, потребуется сделать нечто вроде

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

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

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

Изменено в версии 3.11: Принимает path-like object.

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.

Объекты подключения¶Connection Objects

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

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

class multiprocessing.connection.Connection¶

send(obj)¶

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

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

recv()¶

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

fileno()¶

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

close()¶

Закрывает подключение.

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

poll([timeout])¶

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

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

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

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

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

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

recv_bytes([maxlength])¶

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

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

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

recv_bytes_into(buffer[, offset])¶

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

buffer должен быть доступным для записи байтоподобным объектом. Если задан offset, то сообщение будет записано в буфер, начиная с этой позиции. 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])

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

В целом, в многопроцессных программах примитивы синхронизации требуются не так часто, как в многопоточных. См. документацию модуля 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 или 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 объекты¶Shared ctypes Objects

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

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

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

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

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

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

counter.value += 1

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

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

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

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

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

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

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

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

Обратите внимание, что массив типа 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)]

Менеджеры¶Managers

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

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_, если он существует.) Если список exposed не указан, то все «публичные методы» разделяемого объекта будут доступны. (Здесь «публичный метод» означает любой атрибут, у которого есть метод __call__() и имя которого не начинается с '_'.)

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

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

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

address¶

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

Изменено в версии 3.3: Объекты Manager поддерживают протокол управления контекстом – см. Типы менеджеров контекста. __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 и вернуть прокси для него.

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

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

Event()¶

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

Lock()¶

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

Namespace()¶

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

Queue([maxsize])¶

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

RLock()¶

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

Semaphore([value])¶

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

Array(typecode, sequence)¶

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

Value(typecode, value)¶

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

dict()¶

dict(mapping)

dict(sequence)

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

list()¶

list(sequence)

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

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

class multiprocessing.managers.Namespace¶

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

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

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

>>> manager = multiprocessing.Manager()
>>> Global = manager.Namespace()
>>> Global.x = 10
>>> Global.y = 'hello'
>>> Global._z = 12.3    # это атрибут прокси
>>> 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))         # выводит 7
        print(maths.mul(7, 8))         # выводит 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()

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

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

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

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

Обратите внимание, что применение str() к прокси вернёт представление референта, тогда как применение repr() вернёт представление прокси.

Важная особенность объектов-прокси – они сериализуемы (picklable), поэтому их можно передавать между процессами. Таким образом, референт может содержать Объекты-прокси. Это допускает вложение управляемых списков, словарей и других Объектов-прокси:

>>> a = manager.list()
>>> b = manager.list()
>>> a.append(b)         # референт a теперь содержит референт 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__ на объекте-прокси) распространяется через менеджер, поэтому для эффективного изменения такого элемента можно присвоить изменённое значение обратно в прокси-контейнер:

# создать прокси для списка и добавить изменяемый объект (словарь)
lproxy = manager.list()
lproxy.append({})
# теперь изменить словарь
d = lproxy[0]
d['a'] = 1
d['b'] = 2
# на данный момент изменения в d еще не синхронизированы, но при
# обновлении словаря прокси уведомляется об изменении
lproxy[0] = d

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

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

class multiprocessing.managers.BaseProxy¶

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

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

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

Если proxy – это прокси, чей референт – obj, то выражение

proxy._callmethod(methodname, args, kwds)

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

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

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

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

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

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

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

>>> l = manager.list(range(10))
>>> l._callmethod('__len__')
10
>>> l._callmethod('__getitem__', (slice(2, 7),)) # эквивалентно l[2:7]
[2, 3, 4, 5, 6]
>>> l._callmethod('__getitem__', (20,))          # эквивалентно l[20]
Traceback (most recent call last):
...
IndexError: list index out of range

_getvalue()¶

Возвращает копию референта.

Если референт не может быть сериализован (unpicklable), то это вызовет исключение.

__repr__()¶

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

__str__()¶

Возвращает представление референта.

Пулы процессов¶Process Pools

Можно создать пул процессов, который будет выполнять задачи, отправленные ему с помощью класса 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.

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

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

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

map(func, iterable[, chunksize])¶

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

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

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

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

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

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

Если указан 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:Объекты пула теперь поддерживают протокол управления контекстом – см. Context Manager Types. __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:         # запустить 4 рабочих процесса
        result = pool.apply_async(f, (10,)) # вычислить "f(10)" асинхронно в одном процессе
        print(result.get(timeout=1))        # выводит "100", если компьютер не *очень* медленный

        print(pool.map(f, range(10)))       # выводит "[0, 1, 4,..., 81]"

        it = pool.imap(f, range(10))
        print(next(it))                     # выводит "0"
        print(next(it))                     # выводит "1"
        print(it.next(timeout=1))           # выводит "4", если компьютер не *очень* медленный

        result = pool.apply_async(time.sleep, (10,))
        print(result.get(timeout=1))        # вызывает multiprocessing.TimeoutError

Слушатели и клиенты¶Listeners and Clients

Обычно обмен сообщениями между процессами осуществляется с помощью очередей или объектов 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. (См. Address Formats)

Если 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().

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

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

accept()¶

Принимает соединение на привязанном сокете или именованном канале объекта listener и возвращает объект Connection. Если предпринята попытка аутентификации и она не удалась, возбуждается AuthenticationError.

close()¶

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

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

address¶

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

last_accepted¶

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

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

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

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

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

• читаемый объект Connection;

• подключенный и читаемый объект socket.socket; или

• атрибут sentinel объекта Process.

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

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

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

Новое в версии 3.3.

Примеры

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

from multiprocessing.connection import Listener
from array import array

address = ('localhost', 6000)     # семейство определяется как '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()
        # Закрываем записывающий конец канала, чтобы убедиться, что
        # p – единственный процесс, владеющий дескриптором канала. Это
        # гарантирует, что когда p закроет свой дескриптор для записывающего конца,
        # wait() немедленно сообщит, что читающий конец готов
        w.close()

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

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

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

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

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

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

Журналирование¶Logging

Доступна поддержка журналирования. Однако обратите внимание, что пакет 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¶The multiprocessing.dummy module

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.

Все методы запуска¶All start methods

Нижеследующее относится ко всем методам запуска.

Избегайте разделяемого состояния

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

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

Сериализуемость

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

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

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

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

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

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

Лучше наследовать, чем упаковывать/распаковывать

При использовании методов запуска spawn или forkserver многие типы из multiprocessing должны быть упаковываемыми (picklable), чтобы дочерние процессы могли их использовать. Однако обычно следует избегать отправки разделяемых объектов другим процессам через каналы или очереди. Вместо этого стоит организовать программу так, чтобы процесс, которому требуется доступ к разделяемому ресурсу, созданному в другом месте, мог унаследовать его от родительского процесса.

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

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

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

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

Имейте в виду, что процесс, поместивший элементы в очередь, будет ждать перед завершением, пока все буферизованные элементы не будут переданы потоком «feeder» в нижележащий канал. (Дочерний процесс может вызвать метод 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()                    # это приводит к взаимоблокировке
    obj = queue.get()

Исправление заключается в замене двух последних строк местами (или просто удалении строки с p.join()).

Явная передача ресурсов дочерним процессам

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

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

Так, например

from multiprocessing import Process, Lock

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

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

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

from multiprocessing import Process, Lock

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

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

Остерегайтесь замены sys.stdin на «файлоподобный объект»

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

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

в методе multiprocessing.Process._bootstrap() – это приводило к проблемам с процессами внутри процессов. Теперь это изменено на:

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

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

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

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

Для получения дополнительной информации см. bpo-5155, bpo-5313 и bpo-5331