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

Исходный код: Lib/multiprocessing/

1. Введение

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

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

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

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

[1, 4, 9]

1.1. Класс Process

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

from multiprocessing import Process

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

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

Для демонстрации отдельных идентификаторов процессов, вот расширенный пример:

from multiprocessing import Process
import os

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

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

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

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

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

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

spawn

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

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

fork

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

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

forkserver

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

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

На Unix с методами запуска spawn или forkserver также запустится процесс отслеживания семафоров, который отслеживает созданные процессами программы не связанных именованных семафоров. Когда все процессы завершат работу, процесс-отслеживатель освобождает все оставшиеся семафоры. Обычно их не должно быть, но если процесс был убит сигналом, могут быть некоторые "утекшие" семафоры. (Освобождение именованных семафоров — серьезная проблема, так как система допускает ограниченное количество, и они не будут автоматически освобождены до следующей перезагрузки.)

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

import multiprocessing as mp

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

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

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

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

import multiprocessing as mp

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

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

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

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

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

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

Очереди

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

from multiprocessing import Process, Queue

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

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

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

Трубы

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

from multiprocessing import Process, Pipe

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

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

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

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

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

from multiprocessing import Process, Lock

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

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

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

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

1.5. Обмен состояния между процессами

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

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

Общий доступ к памяти

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

from multiprocessing import Process, Value, Array

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

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

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

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

выведет

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

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

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

Процесс сервера

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

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

from multiprocessing import Process, Manager

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

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

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

        print(d)
        print(l)

выведет

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

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

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

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

Например:

from multiprocessing import Pool, TimeoutError
import time
import os

def f(x):
    return x*x

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

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

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

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

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

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

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

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

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

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

>>> from multiprocessing import Pool
>>> p = Pool(5)
>>> def f(x):
...     return x*x
...
>>> p.map(f, [1,2,3])
Process PoolWorker-1:
Process PoolWorker-2:
Process PoolWorker-3:
Traceback (most recent call last):
AttributeError: 'module' object has no attribute 'f'
AttributeError: 'module' object has no attribute 'f'
AttributeError: 'module' object has no attribute 'f'

2. Справочник

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

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

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

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

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

По умолчанию никакие аргументы не передаются в target.

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

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

run()

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

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

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

authkey

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

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

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

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

sentinel

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

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

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

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

terminate()

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

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

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

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

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

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

>>> import multiprocessing, time, signal
>>> p = multiprocessing.Process(target=time.sleep, args=(1000,))
>>> print(p, p.is_alive())
<Process(Process-1, initial)> False
>>> p.start()
>>> print(p, p.is_alive())
<Process(Process-1, started)> True
>>> p.terminate()
>>> time.sleep(0.1)
>>> print(p, p.is_alive())
<Process(Process-1, stopped[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

Выбрасывается методами с таймаутом, когда таймаут истекает.

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

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

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

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

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

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

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

multiprocessing.Pipe([duplex])

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

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

class multiprocessing.Queue([maxsize])

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

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

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

qsize()

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

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

empty()

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

full()

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

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

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

put_nowait(obj)

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

get([block[, timeout]])

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

get_nowait()

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

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

close()

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

join_thread()

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

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

cancel_join_thread()

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

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

Примечание

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

class multiprocessing.SimpleQueue

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

empty()

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

get()

Удалить и вернуть элемент из очереди.

put(item)

Поместить item в очередь.

class multiprocessing.JoinableQueue([maxsize])

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

task_done()

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

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

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

join()

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

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

2.3. Разное

multiprocessing.active_children()

Возвращает список всех живых потомков текущего процесса.

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

multiprocessing.cpu_count()

Возвращает количество процессоров в системе. Может вызвать NotImplementedError.

См. также

os.cpu_count()

multiprocessing.current_process()

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

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

multiprocessing.freeze_support()

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

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

from multiprocessing import Process, freeze_support

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

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

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

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

multiprocessing.get_all_start_methods()

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

Новая функция в версии 3.4.

multiprocessing.get_context(method=None)

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

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

Новая функция в версии 3.4.

multiprocessing.get_start_method(allow_none=False)

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

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

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

Новая функция в версии 3.4.

multiprocessing.set_executable()

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

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

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

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

multiprocessing.set_start_method(method)

Устанавливает метод, который следует использовать для запуска дочерних процессов. method может быть 'fork', 'spawn' или 'forkserver'.

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

Новая функция в версии 3.4.

2.4. Объекты соединений

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

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

class multiprocessing.Connection

send(obj)

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

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

recv()

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

fileno()

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

close()

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

Это происходит автоматически при сборке мусора соединения.

poll([timeout])

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

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

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

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

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

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

recv_bytes([maxlength])

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

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

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

recv_bytes_into(buffer[, offset])

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

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

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

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

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

Например:

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

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

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

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

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

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

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

class multiprocessing.BoundedSemaphore([value])

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

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

Примечание

На Mac OS X это неотличимо от Semaphore, потому что sem_getvalue() не реализован на этой платформе.

class multiprocessing.Condition([lock])

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

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

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

class multiprocessing.Event

Копия threading.Event.

class multiprocessing.Lock

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

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

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

acquire(block=True, timeout=None)

Получение блокировки, блокирующая или неблокирующая.

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

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

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

release()

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

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

class multiprocessing.RLock

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

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

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

acquire(block=True, timeout=None)

Получение блокировки, блокирующая или неблокирующая.

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

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

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

release()

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

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

class multiprocessing.Semaphore([value])

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

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

2.6. Объекты ctypes общего доступа

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

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

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

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

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

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

counter.value += 1

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

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

Обратите внимание, что lock — ключевое аргумент.

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

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

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

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

Обратите внимание, что lock — ключевой аргумент.

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

2.6.1. Модуль 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, выделенный из общей памяти.

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

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

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

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

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

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

Обратите внимание, что lock — ключевой аргумент.

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

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

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

Обратите внимание, что lock — ключевой аргумент.

multiprocessing.sharedctypes.copy(obj)

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

multiprocessing.sharedctypes.synchronized(obj[, lock])

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

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

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

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

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

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

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

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

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

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

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

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

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

Печатаемые результаты

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

2.7. Менеджеры

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

multiprocessing.Manager()

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

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

class multiprocessing.managers.BaseManager([address[, authkey]])

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

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

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

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

start([initializer[, initargs]])

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

get_server()

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

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

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

connect()

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

>>> from multiprocessing.managers import BaseManager
>>> m = BaseManager(address=('127.0.0.1', 5000), 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: Объекты менеджера поддерживают протокол управления контекстом — см. Типы менеджеров контекста. __enter__() запускает процесс сервера (если он ещё не запущен) и затем возвращает объект менеджера. __exit__() вызывает shutdown().

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

class multiprocessing.managers.SyncManager

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

Также поддерживает создание общих списков и словарей.

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

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

Введено в версии 3.3.

BoundedSemaphore([value])

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

Condition([lock])

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

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

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

Event()

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

Lock()

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

Namespace()

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

Queue([maxsize])

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

RLock()

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

Semaphore([value])

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

Array(typecode, sequence)

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

Value(typecode, value)

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

dict()

dict(mapping)

dict(sequence)

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

list()

list(sequence)

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

Примечание

Изменения в изменяемых значениях или элементах прокси-объектов dict и list не будут переданы через менеджер, так как у прокси нет способа узнать, когда значения или элементы изменены. Чтобы изменить элемент, можно повторно присвоить изменённый объект прокси-контейнеру:

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

class multiprocessing.managers.Namespace

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

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

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

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

2.7.1. Настраиваемые менеджеры

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

from multiprocessing.managers import BaseManager

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

class MyManager(BaseManager):
    pass

MyManager.register('Maths', MathsClass)

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

2.7.2. Использование удалённого менеджера

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

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

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

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

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

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

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

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

>>> from multiprocessing import Process, Queue
>>> from multiprocessing.managers import BaseManager
>>> class Worker(Process):
...     def __init__(self, q):
...         self.q = q
...         super(Worker, self).__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()

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

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

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

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

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

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

>>> a = manager.list()
>>> b = manager.list()
>>> a.append(b)         # referent of a now contains referent of b
>>> print(a, b)
[[]] []
>>> b.append('hello')
>>> print(a, b)
[['hello']] ['hello']

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

class multiprocessing.managers.BaseProxy

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

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

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

Если proxy — это прокси, объект-ссылочный которого — obj, то выражение

proxy._callmethod(methodname, args, kwds)

выполнит выражение

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

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

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

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

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

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

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

_getvalue()

Возвращает копию объекта-ссылочного.

Если объект-ссылочный не сериализуем (unpicklable), будет сгенерировано исключение.

__repr__()

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

__str__()

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

2.8.1. Очистка

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

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

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

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

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

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

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

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

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

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

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

Добавлен в версии 3.2: maxtasksperchild

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

Примечание

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

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

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

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

Вариант метода apply(), который возвращает объект результата.

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

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

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

map(func, iterable[, chunksize])

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

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

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

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

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

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

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

imap(func, iterable[, chunksize])

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

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

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

imap_unordered(func, iterable[, chunksize])

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

starmap(func, iterable[, chunksize])

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

Таким образом, iterable из [(1,2), (3, 4)] приводит к [func(1,2), func(3,4)].

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

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

Комбинация starmap() и map_async(), которая перебирает iterable итерируемых объектов и вызывает func с распакованными итерируемыми объектами. Возвращает объект результата.

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

close()

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

terminate()

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

join()

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

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

class multiprocessing.pool.AsyncResult

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

get([timeout])

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

wait([timeout])

Ожидает, пока результат не станет доступным или не пройдёт timeout секунд.

ready()

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

successful()

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

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

from multiprocessing import Pool
import time

def f(x):
    return x*x

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

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

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

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

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

Обычно передача сообщений между процессами выполняется с помощью очередей или с использованием объектов 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[, authenticate[, authkey]]])

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

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

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

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

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

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

Примечание

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

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

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

Если authenticate равно True (по умолчанию False) или authkey не равно None, то используется аутентификация по дайджестам.

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

Если authkey равно None и authenticate равно True, то используется current_process().authkey в качестве ключа аутентификации. Если authkey равно None и authenticate равно False, то аутентификация не выполняется. Если аутентификация завершится неудачей, то будет поднято исключение AuthenticationError. См. Ключи аутентификации.

accept()

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

close()

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

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

address

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

last_accepted

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

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

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

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

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

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

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

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

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

Введено в версии 3.3.

Примеры

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

from multiprocessing.connection import Listener
from array import array

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

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

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

        conn.send_bytes(b'hello')

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

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

from multiprocessing.connection import Client
from array import array

address = ('localhost', 6000)

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

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

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

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

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

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

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

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

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

2.10.1. Форматы адресов

• Адрес 'AF_INET' — это кортеж вида (hostname, port), где hostname — строка, а port — целое число.
• Адрес 'AF_UNIX' — это строка, представляющая имя файла в файловой системе.

• An 'AF_PIPE' address is a string of the form

  r'\\.\pipe{PipeName}'. Чтобы использовать Client() для подключения к именованному каналу на удалённом компьютере под названием ServerName, следует использовать адрес вида r'\ServerName\pipe{PipeName}'.

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

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

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

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

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

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

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

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

multiprocessing.get_logger()

Возвращает используемый в multiprocessing регистратор. При необходимости будет создан новый.

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

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

multiprocessing.log_to_stderr()

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

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

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

2.13. Модуль multiprocessing.dummy

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

3. Руководство по программированию

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

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

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

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

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

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

Возможность сериализации

Безопасность потоков прокси

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

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

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

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

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

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

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

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

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

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

Вот пример, который приведёт к тупику:

from multiprocessing import Process, Queue

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

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

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

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

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

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

Например

from multiprocessing import Process, Lock

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

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

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

from multiprocessing import Process, Lock

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

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

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

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

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

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

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

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

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

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

Дополнительную информацию см. в bpo-5155, bpo-5313 и bpo-5331

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

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

Большая пиклируемость

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

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

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

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

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

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

from multiprocessing import Process

def foo():
    print('hello')

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

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

from multiprocessing import Process, freeze_support, set_start_method

def foo():
    print('hello')

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

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

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

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

4. Примеры

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

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

# A simple generator function
def baz():
    for i in range(10):
        yield i*i

# Proxy type for generator objects
class GeneratorProxy(BaseProxy):
    _exposed_ = ['__next__']
    def __iter__(self):
        return self
    def __next__(self):
        return self._callmethod('__next__')

# Function to return the operator module
def get_operator_module():
    return operator

##

class MyManager(BaseManager):
    pass

# register the Foo class; make `f()` and `g()` accessible via proxy
MyManager.register('Foo1', Foo)

# register the Foo class; make `g()` and `_h()` accessible via proxy
MyManager.register('Foo2', Foo, exposed=('g', '_h'))

# register the generator function baz; use `GeneratorProxy` to make proxies
MyManager.register('baz', baz, proxytype=GeneratorProxy)

# register get_operator_module(); make public functions accessible via proxy
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:

import multiprocessing
import time
import random
import sys

#
# Functions used by test code
#

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

#
# Test code
#

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

    with multiprocessing.Pool(PROCESSES) as pool:
        #
        # Tests
        #

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

        #
        # Test error handling
        #

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

        #
        # Testing timeouts
        #

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

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

import time
import random

from multiprocessing import Process, Queue, current_process, freeze_support

#
# Function run by worker processes
#

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

#
# Function used to calculate result
#

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

#
# Functions referenced by tasks
#

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

    # Create queues
    task_queue = Queue()
    done_queue = Queue()

    # Submit tasks
    for task in TASKS1:
        task_queue.put(task)

    # Start worker processes
    for i in range(NUMBER_OF_PROCESSES):
        Process(target=worker, args=(task_queue, done_queue)).start()

    # Get and print results
    print('Unordered results:')
    for i in range(len(TASKS1)):
        print('\t', done_queue.get())

    # Add more tasks using `put()`
    for task in TASKS2:
        task_queue.put(task)

    # Get and print some more results
    for i in range(len(TASKS2)):
        print('\t', done_queue.get())

    # Tell child processes to stop
    for i in range(NUMBER_OF_PROCESSES):
        task_queue.put('STOP')


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