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

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

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

17.2.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__' необходима, см. Руководящие принципы программирования.

17.2.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(), чтобы не вмешиваться в выбор пользователя библиотеки.

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

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

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

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

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

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

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

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

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

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

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

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

17.2.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() разблокируется.

17.2.2.3. Разное

multiprocessing.active_children()

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

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

multiprocessing.cpu_count()

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

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

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

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

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

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

class multiprocessing.connection.Connection

send(obj)

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

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

recv()

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

fileno()

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

close()

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

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

poll([timeout])

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

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

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

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

Отправить данные байтов из объекта типа bytes-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, то сообщение будет записано в буфер с этой позиции. Смещение должно быть целым числом, неотрицательным, меньшим длины buffer (в байтах).

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

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

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

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

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

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

17.2.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 — отображение, используемое для указания возвращаемого типа тех методов exposed, которые должны возвращать прокси. Оно сопоставляет имена методов со строками typeid. (Если method_to_typeid None , то proxytype._method_to_typeid_ используется вместо этого, если он существует.) Если имя метода не является ключом этого отображения или если отображение None , то возвращаемый объектом метод будет скопирован по значению.

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

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

address

Используемый адрес менеджера.

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

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

class multiprocessing.managers.SyncManager

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

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

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

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

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

BoundedSemaphore([value])

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

Condition([lock])

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

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

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

Event()

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

Lock()

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

Namespace()

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

Queue([maxsize])

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

RLock()

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

Semaphore([value])

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

Array(typecode, sequence)

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

Value(typecode, value)

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

dict()

dict(mapping)

dict(sequence)

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

list()

list(sequence)

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

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

class multiprocessing.managers.Namespace

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

>>> a = manager.list()
>>> b = manager.list()
>>> a.append(b)         # referent of a now contains referent of b
>>> print(a, b)
[<ListProxy object, typeid 'list' at ...>] []
>>> b.append('hello')
>>> print(a[0], b)
['hello'] ['hello']

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

>>> l_outer = manager.list([ manager.dict() for i in range(2) ])
>>> d_first_inner = l_outer[0]
>>> d_first_inner['a'] = 1
>>> d_first_inner['b'] = 2
>>> l_outer[1]['c'] = 3
>>> l_outer[1]['z'] = 26
>>> print(l_outer[0])
{'a': 1, 'b': 2}
>>> print(l_outer[1])
{'c': 3, 'z': 26}

Если стандартные (не прокси) объекты list или dict содержатся в оригинале, изменения этих изменяемых значений не будут переданы через менеджер, поскольку у прокси нет способа узнать, когда значения внутри изменяются. Однако хранение значения в контейнерном прокси (что вызывает __setitem__ на объекте прокси) действительно передаётся через менеджер, поэтому для эффективного изменения такого элемента можно переназначить изменённое значение в контейнерный прокси:

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

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

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

class multiprocessing.managers.BaseProxy

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

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

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

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

proxy._callmethod(methodname, args, kwds)

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

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

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

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

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

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

Пример использования метода _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()

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

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

__repr__()

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

__str__()

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

17.2.2.8.1. Очистка

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

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

17.2.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). Он блокируется до тех пор, пока результат не будет готов.

Этот метод делит 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_callback]]])

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

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

close()

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

terminate()

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

join()

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

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

class multiprocessing.pool.AsyncResult

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

get([timeout])

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

wait([timeout])

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

ready()

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

successful()

Возвращает, завершился ли вызов без возбуждения исключения. Будет вызвано исключение 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

17.2.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[, authkey]])

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

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

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

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

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

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

Примечание

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

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

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

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

accept()

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

close()

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

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

address

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

last_accepted

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

Добавлена в версии 3.3: Объекты слушателей теперь поддерживают протокол управления контекстом — см. Типы менеджеров контекста. __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)

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

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

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

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

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

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

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

17.2.2.13. Модуль multiprocessing.dummy

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

17.2.3. Рекомендации по программированию

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

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

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

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

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

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

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

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

Обеспечение потоковой безопасности проксируемых объектов

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

(Разные процессы никогда не используют один и тот же объект прокси.)

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

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

Наследование предпочтительнее сериализации/десериализации

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

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

Использование метода 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

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

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

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

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

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

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

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

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

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

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

from multiprocessing import Process

def foo():
    print('hello')

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

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

from multiprocessing import Process, freeze_support, set_start_method

def foo():
    print('hello')

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

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

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

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

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