`multiprocessing` – процесс-ориентированный параллелизм

Доступность: не Android, не iOS, не WASI

Этот модуль не поддерживается на мобильных платформах или платформах WebAssembly.

IntroductionВведение

multiprocessing – это пакет, поддерживающий порождение процессов с использованием API, похожего на модуль threading. Пакет multiprocessing предоставляет как локальную, так и удалённую конкурентность, обходя глобальную блокировку интерпретатора с помощью подпроцессов вместо потоков. Благодаря этому модуль multiprocessing позволяет программисту полностью использовать несколько процессоров на данной машине. Он работает как на POSIX, так и на Windows.

Модуль multiprocessing также вводит объект Pool, который предоставляет удобное средство параллелизации выполнения функции над несколькими входными значениями, распределяя входные данные между процессами (распределённый параллелизм). Следующий пример демонстрирует обычную практику определения таких функций в модуле, чтобы дочерние процессы могли успешно импортировать этот модуль. Этот базовый пример распределённого параллелизма с использованием Pool,

python

from multiprocessing import Pool

def f(x):
    return x*x

if __name__ == '__main__':
    with Pool(5) as p:
        print(p.map(f, [1, 2, 3]))

выведет на стандартный вывод

python

[1, 4, 9]

Модуль multiprocessing также вводит API, не имеющие аналогов в модуле threading, например возможность terminate, interrupt или kill работающий процесс.

Смотрите также

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

Класс `Process`

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

python

from multiprocessing import Process

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

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

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

python

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 обычно должны быть распаковываемыми (unpickleable) внутри дочернего процесса. Если попытаться ввести приведённый выше пример непосредственно в REPL, это может привести к AttributeError в дочернем процессе при попытке найти функцию f в модуле __main__.

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

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

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

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

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

Доступен на системах POSIX.

Изменено в версии 3.14: Этот метод больше не является методом запуска по умолчанию ни на одной платформе. Код, требующий fork, должен явно указать это через get_context() или set_start_method().

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

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

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

Изменено в версии 3.14: Этот метод стал методом запуска по умолчанию на платформах POSIX.

Изменено в версии 3.4: spawn добавлен на всех платформах POSIX, а forkserver добавлен на некоторых платформах POSIX. Дочерние процессы больше не наследуют все наследуемые дескрипторы (handles) родителя на Windows.

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

Изменено в версии 3.14: На платформах POSIX метод запуска по умолчанию был изменён с fork на forkserver для сохранения производительности, но избежания распространённых проблем несовместимости с многопоточными процессами. См. gh-84559.

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

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

python

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

python

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

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

Методы запуска 'spawn' и 'forkserver', как правило, нельзя использовать с «замороженными» исполняемыми файлами (т.е. двоичными файлами, созданными такими пакетами, как PyInstaller и cx_Freeze) в системах POSIX. Метод запуска 'fork' может работать, если код не использует потоки.

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

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

Очереди

Класс Queue почти полностью повторяет queue.Queue. Например:
python
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()
Очереди безопасны для потоков и процессов. Любой объект, помещённый в очередь multiprocessing, будет сериализован.

Каналы

Функция Pipe() возвращает пару объектов соединения, связанных каналом, который по умолчанию является дуплексным (двусторонним). Например:
python
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() (среди прочих). Следует учитывать, что данные в канале могут быть повреждены, если два процесса (или потока) попытаются одновременно читать или писать в один и тот же конец канала. Разумеется, при использовании разными процессами разных концов канала одновременно риск повреждения отсутствует.

Метод send() сериализует объект, а recv() воссоздаёт объект.

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

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

python

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. Например, следующий код
python
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[:])
выведет
python
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, set, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value и Array. Например,
python
from multiprocessing import Process, Manager

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

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

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

        print(d)
        print(l)
        print(s)
выведет
python
{0.25: None, 1: '1', '2': 2}
[9, 8, 7, 6, 5, 4, 3, 2, 1, 0]
{'a', 'b'}
Менеджеры серверных процессов более гибкие, чем использование объектов разделяемой памяти, поскольку они могут поддерживать произвольные типы объектов. Кроме того, один менеджер может быть разделён между процессами на разных компьютерах через сеть. Однако они медленнее, чем использование разделяемой памяти.

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

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

Например:

python

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

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

Примечание

Функциональность этого пакета требует, чтобы модуль __main__ был импортируем дочерними процессами. Об этом говорится в Рекомендациях по программированию, однако стоит указать это здесь. Это означает, что некоторые примеры, например примеры multiprocessing.pool.Pool, не будут работать в интерактивном интерпретаторе. Например:

python

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

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

ReferenceСсылка

Пакет multiprocessing в основном повторяет API модуля threading.

Global start methodГлобальный метод запуска

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

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

`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 – кортеж аргументов для вызова target. kwargs – словарь именованных аргументов для вызова target. Если указан, аргумент daemon (только ключевое слово) устанавливает флаг daemon процесса в True или False. Если None (по умолчанию), этот флаг наследуется от создающего процесса.

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

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

Примечание

В общем случае все аргументы Process должны быть сериализуемы (picklable). Это часто наблюдается при попытке создать Process или использовать concurrent.futures.ProcessPoolExecutor из REPL с локально определённой функцией target.

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

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

python

>>> 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, даёт тот же эффект.

Пример:

python

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

start()¶

Запускает действие процесса.

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

join([timeout])¶

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

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

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

name¶

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

Начальное имя устанавливается конструктором. Если конструктору не указано явное имя, создаётся имя вида ‘Process-N₁:N₂:…:N_k’, где каждый N_k – N-й дочерний элемент своего родителя.

is_alive()¶

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

Грубо говоря, объект процесса жив с момента возврата метода start() до завершения дочернего процесса.

daemon¶

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

Начальное значение наследуется от создающего процесса.

Когда процесс завершается, он пытается завершить все свои дочерние процессы-демоны.

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

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

pid¶: Возвращает идентификатор процесса. До порождения процесса это будет None.

exitcode¶

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

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

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

authkey¶

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

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

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

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

sentinel¶

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

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

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

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

interrupt()¶

Прерывает процесс. Работает в POSIX с использованием сигнала SIGINT. Поведение в Windows не определено.

По умолчанию это завершает дочерний процесс путём возбуждения KeyboardInterrupt. Это поведение можно изменить, установив соответствующий обработчик сигнала в дочернем процессе signal.signal() для SIGINT.

Примечание: если дочерний процесс перехватывает и игнорирует KeyboardInterrupt, процесс не будет завершён.

Примечание: поведение по умолчанию также устанавливает exitcode в 1, как если бы в дочернем процессе было возбуждено необработанное исключение. Чтобы получить другое exitcode, вы можете просто перехватить KeyboardInterrupt и вызвать exit(your_code).

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

terminate()¶

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

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

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

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

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

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

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

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

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

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

python

>>> 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¶: Вызывается методами с таймаутом при истечении времени ожидания.

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

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

Для передачи сообщений можно использовать 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 использует обычные исключения queue.Empty и queue.Full для сигнализации таймаута. Они недоступны в пространстве имён multiprocessing, поэтому их необходимо импортировать из queue.

Примечание

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

После помещения объекта в пустую очередь может пройти бесконечно малая задержка, прежде чем метод empty() очереди вернёт False, и get_nowait() сможет вернуться без вызова queue.Empty.
Если несколько процессов помещают объекты в очередь, возможно, что объекты будут получены на другом конце не по порядку. Однако объекты, помещённые в очередь одним и тем же процессом, всегда будут в ожидаемом порядке относительно друг друга.

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

Если процесс уничтожается с помощью Process.terminate() или os.kill() во время попытки использовать Queue, то данные в очереди, скорее всего, будут повреждены. Это может привести к тому, что любой другой процесс получит исключение при попытке использовать очередь позднее.

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

Как упоминалось выше, если дочерний процесс поместил элементы в очередь (и не использовал JoinableQueue.cancel_join_thread), то этот процесс не завершится, пока все буферизованные элементы не будут сброшены в канал.

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

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

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

multiprocessing.Pipe(duplex=True)¶

Возвращает пару (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() и shutdown().

qsize()¶

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

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

empty()¶

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

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

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

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

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

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

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

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

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

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

close()¶

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

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

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

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

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

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

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

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

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__' главного модуля. Например:

python

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

Возвращаемое значение может быть 'fork', 'spawn', 'forkserver' или None. См. Контексты и методы запуска.

Добавлено в версии 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: Теперь поддерживается на POSIX при использовании метода запуска 'spawn'.

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

multiprocessing.set_forkserver_preload(module_names, *, on_error='ignore')¶

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

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

Параметр on_error управляет тем, как обрабатываются исключения ImportError во время предварительной загрузки модулей: "ignore" (по умолчанию) молча игнорирует ошибки, "warn" заставляет подпроцесс forkserver выводить ImportWarning в stderr, а "fail" заставляет подпроцесс forkserver завершиться с трассировкой исключения в stderr, что приводит к сбою при создании последующих процессов с EOFError или ConnectionError.

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

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

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

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.

Примечание

multiprocessing не содержит аналогов threading.active_count(), threading.enumerate(), threading.settrace(), threading.setprofile(), threading.Timer или threading.local.

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(buf[, offset[, size]])¶

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

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

recv_bytes([maxlength])¶

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

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

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

recv_bytes_into(buf[, offset])¶

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

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

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

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

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

Например:

python

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

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

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

Поэтому, если объект соединения не был создан с помощью Pipe(), следует использовать методы recv() и send() только после выполнения некоторой аутентификации. См. Ключи аутентификации.

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

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

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

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

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

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

Объект-барьер: клон threading.Barrier.

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

class multiprocessing.BoundedSemaphore([value])¶

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

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

locked()¶: Возвращает логическое значение, указывающее, заблокирован ли этот объект в данный момент.

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

Примечание

На 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.

locked()¶: Возвращает логическое значение, указывающее, заблокирован ли этот объект в данный момент.

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

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

locked()¶: Возвращает логическое значение, указывающее, заблокирован ли этот объект в данный момент.

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

class multiprocessing.Semaphore([value])¶

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

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

get_value()¶

Возвращает текущее значение семафора.

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

locked()¶: Возвращает логическое значение, указывающее, заблокирован ли этот объект в данный момент.

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

Примечание

На macOS sem_timedwait не поддерживается, поэтому вызов acquire() с таймаутом будет эмулировать поведение этой функции с помощью спящего цикла.

Примечание

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

Разделяемые `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, то доступ к возвращаемому объекту не будет автоматически защищён блокировкой, поэтому он не обязательно будет «процесс-безопасным».

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

python

counter.value += 1

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

python

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

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

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

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

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

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

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

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

Модуль `multiprocessing.sharedctypes`

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

Примечание

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

multiprocessing.sharedctypes.RawArray(typecode_or_type, size_or_initializer)¶

Возвращает массив ctypes, выделенный из разделяемой памяти.

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

Обратите внимание, что установка и получение элемента потенциально не атомарны – используйте Array() вместо этого, чтобы гарантировать автоматическую синхронизацию доступа с помощью блокировки.

multiprocessing.sharedctypes.RawValue(typecode_or_type, *args)¶

Возвращает объект ctypes, выделенный из разделяемой памяти.

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

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

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

То же, что и RawArray(), за исключением того, что в зависимости от значения блокировка вместо необработанного ctypes-массива может быть возвращена обёртка для синхронизации, безопасная для процессов.

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

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

Обратите внимание, что блокировка и ctx являются только ключевыми параметрами.

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

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

Обратите внимание, что блокировка и ctx являются только ключевыми параметрами.

multiprocessing.sharedctypes.copy(obj)¶: Возвращает объект ctypes, выделенный в разделяемой памяти, который является копией объекта ctypes obj.

multiprocessing.sharedctypes.synchronized(obj, lock=None, ctx=None)¶

Возвращает объект-обёртку, безопасную для процессов, для объекта ctypes, который использует блокировка для синхронизации доступа. Если блокировка равно None (по умолчанию), то объект multiprocessing.RLock создаётся автоматически.

Синхронизированная обёртка будет иметь два метода в дополнение к методам обёртываемого объекта: get_obj() возвращает обёрнутый объект, а get_lock() возвращает объект блокировки, используемый для синхронизации.

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

Изменено в версии 3.5: Синхронизированные объекты поддерживают протокол менеджера контекста.

В таблице ниже сравнивается синтаксис создания разделяемых ctypes-объектов в разделяемой памяти с обычным синтаксисом ctypes. (В таблице MyStruct – это некоторый подкласс ctypes.Structure.)

ctypes	sharedctypes с типом	sharedctypes с кодом типа
c_double(2.4)	RawValue(c_double, 2.4)	RawValue(‘d’, 2.4)
MyStruct(4, 6)	RawValue(MyStruct, 4, 6)
(c_short * 7)()	RawArray(c_short, 7)	RawArray(‘h’, 7)
(c_int * 3)(9, 2, 8)	RawArray(c_int, (9, 2, 8))	RawArray(‘i’, (9, 2, 8))

Ниже приведён пример, в котором дочерний процесс изменяет несколько объектов ctypes:

python

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

Выводятся следующие результаты:

text

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 (используйте текущий контекст). Если None, вызов этого может установить глобальный метод запуска. Подробнее см. в разделе Глобальный метод запуска.

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

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

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

get_server()¶

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

python

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

Server дополнительно имеет атрибут address.

connect()¶

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

python

>>> 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 поддерживают протокол менеджера контекста – см. Context Manager Types. __enter__() запускает серверный процесс (если он ещё не запущен), а затем возвращает объект менеджера. __exit__() вызывает shutdown().

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

class multiprocessing.managers.SyncManager¶

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

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

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 и возвращает прокси для него.

set()¶
set(sequence)
set(mapping): Создаёт разделяемый объект set и возвращает прокси для него.

Добавлено в версии 3.14: set поддержка была добавлена.

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

class multiprocessing.managers.Namespace¶

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

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

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

python

>>> mp_context = multiprocessing.get_context('spawn')
>>> manager = mp_context.Manager()
>>> Global = manager.Namespace()
>>> Global.x = 10
>>> Global.y = 'hello'
>>> Global._z = 12.3    # это атрибут прокси
>>> print(Global)
Namespace(x=10, y='hello')

Customized managersНастраиваемые менеджеры

Чтобы создать собственный менеджер, нужно создать подкласс BaseManager и использовать метод класса register() для регистрации новых типов или вызываемых объектов в классе менеджера. Например:

python

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

Using a remote managerИспользование удалённого менеджера

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

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

python

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

Один клиент может обратиться к серверу следующим образом:

python

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

Другой клиент также может его использовать:

python

>>> 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'

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

python

>>> 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Прокси-объекты

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

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

python

>>> 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() возвращает представление самого прокси.

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

python

>>> 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']

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

python

>>> 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__ на объекте прокси) распространяется через менеджер, поэтому для эффективного изменения такого элемента можно переназначить измененное значение контейнерному прокси:

python

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

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

Примечание

Типы прокси в multiprocessing не поддерживают сравнение по значению. Так, например, имеем:

python

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

При сравнении следует просто использовать копию референта.

class multiprocessing.managers.BaseProxy¶

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

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

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

Если proxy является прокси, референтом которого является obj, то выражение

python

proxy._callmethod(methodname, args, kwds)

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

python

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

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

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

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

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

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

python

>>> 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__()¶: Возвращает представление референта.

CleanupОчистка

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

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

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

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

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

Если указан 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.

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

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

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

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

starmap_async(func, iterable[, chunksize[, callback[, error_callback]]])¶: Комбинация starmap() и map_async(), которая выполняет итерацию по итерируемому объекту из итерируемых объектов и вызывает func с распакованными итерируемыми объектами. Возвращает объект результата.

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

close()¶: Предотвращает отправку новых задач в пул. Когда все задачи будут выполнены, рабочие процессы завершатся.

terminate()¶: Немедленно останавливает рабочие процессы, не завершая выполняемую работу. Когда объект пула будет удалён сборщиком мусора, terminate() будет вызван немедленно.

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

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

Следующий пример демонстрирует использование пула:

python

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]])¶

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

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

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

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

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

address – это адрес, используемый связанным сокетом или именованным каналом объекта Listener.

Примечание

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

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, то блокировка будет неограниченной по времени. Отрицательное значение тайм-аута эквивалентно нулевому.

Как в 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' в качестве ключа аутентификации. Затем он ожидает соединения и отправляет некоторые данные клиенту:

python

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

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

python

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

python

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)

Address FormatsФорматы адресов

Адрес 'AF_INET' – это кортеж вида (hostname, port), где hostname – строка, а port – целое число.
Адрес 'AF_UNIX' – это строка, представляющая имя файла в файловой системе.
Адрес 'AF_PIPE' – это строка вида r'\\.\pipe\PipeName'. Чтобы использовать Client() для подключения к именованному каналу на удалённом компьютере с именем ServerName, следует использовать адрес вида r'\\ServerName\pipe\PipeName'.

Обратите внимание: любая строка, начинающаяся с двух обратных косых черт, по умолчанию считается адресом 'AF_PIPE', а не адресом 'AF_UNIX'.

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.

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

python

>>> 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.

Programming guidelinesРекомендации по программированию

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

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

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

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

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

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

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

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

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

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

(Никогда не возникает проблем при использовании одного и того же прокси-объекта из разных процессов.)

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

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

Лучше наследовать, чем сериализовать/десериализовать через pickle

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

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

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

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

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

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

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

Пример, который приведёт к взаимоблокировке:
python
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()).

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

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

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

Так, например,
python
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()
следует переписать как
python
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 изначально безусловно вызывал:
python
os.close(sys.stdin.fileno())
в методе multiprocessing.Process._bootstrap() – это приводило к проблемам с процессами внутри процессов. Теперь это изменено на:
python
sys.stdin.close()
sys.stdin = open(os.open(os.devnull, os.O_RDONLY), closefd=False)
Это решает основную проблему конфликта процессов, приводящего к ошибке «плохой файловый дескриптор», но вносит потенциальную опасность для приложений, которые заменяют sys.stdin() на «объект, похожий на файл», с буферизацией вывода. Опасность заключается в том, что если несколько процессов вызовут close() для этого файлоподобного объекта, это может привести к многократному сбросу одних и тех же данных в объект и, как следствие, к повреждению данных.

Если вы пишете файлоподобный объект и реализуете собственное кэширование, вы можете сделать его безопасным для fork, сохраняя pid при каждом добавлении в кэш и сбрасывая кэш при изменении pid. Например:
python
@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.

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

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

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

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

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

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

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

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

def foo():
    print('hello')

p = Process(target=foo)
p.start()
Вместо этого следует защитить «точку входа» программы с помощью if __name__ == '__main__': следующим образом:
python
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() можно опустить, если программа будет запускаться обычным образом, а не замороженной (frozen).)

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

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

ExamplesПримеры

Демонстрация того, как создавать и использовать настраиваемые менеджеры и прокси:

python

from multiprocessing import freeze_support
from multiprocessing.managers import BaseManager, BaseProxy
import operator

##

class Foo:
    def f(self):
        print('you called Foo.f()')
    def g(self):
        print('you called Foo.g()')
    def _h(self):
        print('you called Foo._h()')

# Простая функция-генератор
def baz():
    for i in range(10):
        yield i*i

# Тип прокси для объектов-генераторов
class GeneratorProxy(BaseProxy):
    _exposed_ = ['__next__']
    def __iter__(self):
        return self
    def __next__(self):
        return self._callmethod('__next__')

# Функция для возврата модуля operator
def get_operator_module():
    return operator

##

class MyManager(BaseManager):
    pass

# зарегистрировать класс Foo; сделать `f()` и `g()` доступными через прокси
MyManager.register('Foo1', Foo)

# зарегистрировать класс Foo; сделать `g()` и `_h()` доступными через прокси
MyManager.register('Foo2', Foo, exposed=('g', '_h'))

# зарегистрировать функцию-генератор baz; использовать `GeneratorProxy` для создания прокси
MyManager.register('baz', baz, proxytype=GeneratorProxy)

# зарегистрировать get_operator_module(); сделать общедоступные функции доступными через прокси
MyManager.register('operator', get_operator_module)

##

def test():
    manager = MyManager()
    manager.start()

    print('-' * 20)

    f1 = manager.Foo1()
    f1.f()
    f1.g()
    assert not hasattr(f1, '_h')
    assert sorted(f1._exposed_) == sorted(['f', 'g'])

    print('-' * 20)

    f2 = manager.Foo2()
    f2.g()
    f2._h()
    assert not hasattr(f2, 'f')
    assert sorted(f2._exposed_) == sorted(['g', '_h'])

    print('-' * 20)

    it = manager.baz()
    for i in it:
        print('<%d>' % i, end=' ')
    print()

    print('-' * 20)

    op = manager.operator()
    print('op.add(23, 45) =', op.add(23, 45))
    print('op.pow(2, 94) =', op.pow(2, 94))
    print('op._exposed_ =', op._exposed_)

##

if __name__ == '__main__':
    freeze_support()
    test()

Использование Pool:

python

import multiprocessing
import time
import random
import sys

#
# Функции, используемые тестовым кодом
#

def calculate(func, args):
    result = func(*args)
    return '%s says that %s%s = %s' % (
        multiprocessing.current_process().name,
        func.__name__, args, result
        )

def calculatestar(args):
    return calculate(*args)

def mul(a, b):
    time.sleep(0.5 * random.random())
    return a * b

def plus(a, b):
    time.sleep(0.5 * random.random())
    return a + b

def f(x):
    return 1.0 / (x - 5.0)

def pow3(x):
    return x ** 3

def noop(x):
    pass

#
# Тестовый код
#

def test():
    PROCESSES = 4
    print('Creating pool with %d processes\n' % PROCESSES)

    with multiprocessing.Pool(PROCESSES) as pool:
        #
        # Тесты
        #

        TASKS = [(mul, (i, 7)) for i in range(10)] + \
                [(plus, (i, 8)) for i in range(10)]

        results = [pool.apply_async(calculate, t) for t in TASKS]
        imap_it = pool.imap(calculatestar, TASKS)
        imap_unordered_it = pool.imap_unordered(calculatestar, TASKS)

        print('Ordered results using pool.apply_async():')
        for r in results:
            print('\t', r.get())
        print()

        print('Ordered results using pool.imap():')
        for x in imap_it:
            print('\t', x)
        print()

        print('Unordered results using pool.imap_unordered():')
        for x in imap_unordered_it:
            print('\t', x)
        print()

        print('Ordered results using pool.map() --- will block till complete:')
        for x in pool.map(calculatestar, TASKS):
            print('\t', x)
        print()

        #
        # Тестирование обработки ошибок
        #

        print('Testing error handling:')

        try:
            print(pool.apply(f, (5,)))
        except ZeroDivisionError:
            print('\tGot ZeroDivisionError as expected from pool.apply()')
        else:
            raise AssertionError('expected ZeroDivisionError')

        try:
            print(pool.map(f, list(range(10))))
        except ZeroDivisionError:
            print('\tGot ZeroDivisionError as expected from pool.map()')
        else:
            raise AssertionError('expected ZeroDivisionError')

        try:
            print(list(pool.imap(f, list(range(10)))))
        except ZeroDivisionError:
            print('\tGot ZeroDivisionError as expected from list(pool.imap())')
        else:
            raise AssertionError('expected ZeroDivisionError')

        it = pool.imap(f, list(range(10)))
        for i in range(10):
            try:
                x = next(it)
            except ZeroDivisionError:
                if i == 5:
                    pass
            except StopIteration:
                break
            else:
                if i == 5:
                    raise AssertionError('expected ZeroDivisionError')

        assert i == 9
        print('\tGot ZeroDivisionError as expected from IMapIterator.next()')
        print()

        #
        # Тестирование таймаутов
        #

        print('Testing ApplyResult.get() with timeout:', end=' ')
        res = pool.apply_async(calculate, TASKS[0])
        while 1:
            sys.stdout.flush()
            try:
                sys.stdout.write('\n\t%s' % res.get(0.02))
                break
            except multiprocessing.TimeoutError:
                sys.stdout.write('.')
        print()
        print()

        print('Testing IMapIterator.next() with timeout:', end=' ')
        it = pool.imap(calculatestar, TASKS)
        while 1:
            sys.stdout.flush()
            try:
                sys.stdout.write('\n\t%s' % it.next(0.02))
            except StopIteration:
                break
            except multiprocessing.TimeoutError:
                sys.stdout.write('.')
        print()
        print()


if __name__ == '__main__':
    multiprocessing.freeze_support()
    test()

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

python

import time
import random

from multiprocessing import Process, Queue, current_process, freeze_support

#
# Функция, выполняемая рабочими процессами
#

def worker(input, output):
    for func, args in iter(input.get, 'STOP'):
        result = calculate(func, args)
        output.put(result)

#
# Функция для вычисления результата
#

def calculate(func, args):
    result = func(*args)
    return '%s says that %s%s = %s' % \
        (current_process().name, func.__name__, args, result)

#
# Функции, на которые ссылаются задачи
#

def mul(a, b):
    time.sleep(0.5*random.random())
    return a * b

def plus(a, b):
    time.sleep(0.5*random.random())
    return a + b

#
#
#

def test():
    NUMBER_OF_PROCESSES = 4
    TASKS1 = [(mul, (i, 7)) for i in range(20)]
    TASKS2 = [(plus, (i, 8)) for i in range(10)]

    # Создание очередей
    task_queue = Queue()
    done_queue = Queue()

    # Отправка задач
    for task in TASKS1:
        task_queue.put(task)

    # Запуск рабочих процессов
    for i in range(NUMBER_OF_PROCESSES):
        Process(target=worker, args=(task_queue, done_queue)).start()

    # Получение и вывод результатов
    print('Unordered results:')
    for i in range(len(TASKS1)):
        print('\t', done_queue.get())

    # Добавление дополнительных задач с помощью `put()`
    for task in TASKS2:
        task_queue.put(task)

    # Получение и вывод ещё нескольких результатов
    for i in range(len(TASKS2)):
        print('\t', done_queue.get())

    # Указание дочерним процессам остановиться
    for i in range(NUMBER_OF_PROCESSES):
        task_queue.put('STOP')


if __name__ == '__main__':
    freeze_support()
    test()

multiprocessing – процесс-ориентированный параллелизм​

IntroductionВведение​

Класс Process​

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

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

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

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

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

ReferenceСсылка​

Global start methodГлобальный метод запуска​

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

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

MiscellaneousРазное​

Connection ObjectsОбъекты соединений​

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

Разделяемые ctypes объекты​

Модуль multiprocessing.sharedctypes​

ManagersМенеджеры​

Customized managersНастраиваемые менеджеры​

Using a remote managerИспользование удалённого менеджера​

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

CleanupОчистка​

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

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

Address FormatsФорматы адресов​

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

LoggingЛогирование​

Модуль multiprocessing.dummy​

Programming guidelinesРекомендации по программированию​

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

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

ExamplesПримеры​

`multiprocessing` – процесс-ориентированный параллелизм

IntroductionВведение

Класс `Process`

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

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

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

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

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

ReferenceСсылка

Global start methodГлобальный метод запуска

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

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

MiscellaneousРазное

Connection ObjectsОбъекты соединений

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

Разделяемые `ctypes` объекты

Модуль `multiprocessing.sharedctypes`

ManagersМенеджеры

Customized managersНастраиваемые менеджеры

Using a remote managerИспользование удалённого менеджера

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

CleanupОчистка

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

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

Address FormatsФорматы адресов

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

LoggingЛогирование

Модуль `multiprocessing.dummy`

Programming guidelinesРекомендации по программированию

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

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

ExamplesПримеры