16.6. multiprocessing – интерфейс «threading» на основе процессов¶multiprocessing – Process-based “threading” interface

16.6.1.1. Класс Process¶The Process class

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

from multiprocessing import Process

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

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

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

from multiprocessing import Process
import os

def info(title):
    print title
    print 'module name:', __name__
    if hasattr(os, 'getppid'):  # доступно только на Unix
        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()

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

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

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

Очереди

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

from multiprocessing import Process, Queue

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

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

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

Каналы

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

from multiprocessing import Process, Pipe

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

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

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

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

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

from multiprocessing import Process, Lock

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

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

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

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

16.6.1.4. Обмен состоянием между процессами¶Sharing state between processes

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

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

Разделяемая память

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

from multiprocessing import Process, Value, Array

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

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

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

    print num.value
    print arr[:]

выведет

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

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

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

Серверный процесс

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

Менеджер, возвращаемый Manager(), будет поддерживать типы list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, 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__':
    manager = 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]

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

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

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

Например:

from multiprocessing import Pool, TimeoutError
import time
import os

def f(x):
    return x*x

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

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

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

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

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

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

    # заставить один рабочий процесс спать 10 секунд
    res = pool.apply_async(time.sleep, (10,))
    try:
        print res.get(timeout=1)
    except TimeoutError:
        print "We lacked patience and got a multiprocessing.TimeoutError"

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

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

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

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

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

Конструктор всегда следует вызывать с именованными аргументами. group всегда должен быть None; он существует только для совместимости с threading.Thread. target – это вызываемый объект, который будет вызван методом run(). По умолчанию он равен None, то есть ничего не вызывается. name – это имя процесса. По умолчанию создаётся уникальное имя вида 'Process-N₁:N₂:…:N_k', где N₁,N₂,…,N_k – последовательность целых чисел, длина которой определяется поколением процесса. args – это кортеж аргументов для вызова цели. kwargs – это словарь именованных аргументов для вызова цели. По умолчанию никакие аргументы не передаются в target.

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

run()¶

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

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

start()¶

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

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

join([timeout])¶

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

Если timeout равен None, таймаут отсутствует.

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

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

name¶

Имя процесса.

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

is_alive()¶

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

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

daemon¶

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

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

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

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

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

pid¶

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

exitcode¶

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

authkey¶

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

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

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

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

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

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

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

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

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

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

Типы Queue, multiprocessing.queues.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).

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

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

join()¶

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

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

16.6.2.3. Разное¶Miscellaneous

multiprocessing.active_children()¶

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

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

multiprocessing.cpu_count()¶

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

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

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

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

прежде чем они смогут создавать дочерние процессы. (Только для Windows)

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

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

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

class Connection¶

send(obj)¶

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

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

recv()¶

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

fileno()¶

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

close()¶

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

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

poll([timeout])¶

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

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

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

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

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

recv_bytes([maxlength])¶

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

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

recv_bytes_into(buffer[, offset])¶

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

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

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

Например:

>>> from multiprocessing import Pipe
>>> a, b = Pipe()
>>> a.send([1, 'hello', None])
>>> b.recv()
[1, 'hello', None]
>>> b.send_bytes('thank you')
>>> a.recv_bytes()
'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])

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

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

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

class multiprocessing.BoundedSemaphore([value])¶

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

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

Примечание

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

class multiprocessing.Condition([lock])¶

Условная переменная: клон threading.Condition.

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

class multiprocessing.Event¶

Клон threading.Event. Этот метод возвращает состояние внутреннего семафора при выходе, поэтому он всегда возвращает True, за исключением случаев, когда задан таймаут и операция истекает по таймауту.

Изменено в версии 2.7: Ранее метод всегда возвращал None.

class multiprocessing.Lock¶

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

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

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

acquire(block=True, timeout=None)¶

Захватывает блокировку, блокирующую или неблокирующую.

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

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

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

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, и он поддерживает необязательный второй аргумент timeout, в соответствии с Lock.acquire().

16.6.2.6. Разделяемые ctypes объекты¶Shared ctypes Objects

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

multiprocessing.Value(typecode_or_type, *args[, lock])¶

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

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

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

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

counter.value += 1

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

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

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

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

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

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

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

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

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

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

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

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

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

    n = Value('i', 7)
    x = Value(c_double, 1.0/3.0, lock=False)
    s = Array('c', '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)]

16.6.2.7. Менеджеры¶Managers

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

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='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='abc')
>>> m.connect()

shutdown()¶

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

Этот метод можно вызывать несколько раз.

register(typeid[, callable[, proxytype[, exposed[, method_to_typeid[, create_method]]]]])¶

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

typeid – это «идентификатор типа», который используется для идентификации определенного типа разделяемого объекта. Должен быть строкой.

callable – это вызываемый объект, используемый для создания объектов данного идентификатора типа. Если экземпляр менеджера будет создан с помощью метода класса from_address() или если аргумент create_method равен False, то его можно оставить как None.

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

exposed используется для указания последовательности имен методов, к которым прокси для данного typeid должны иметь доступ через BaseProxy._callmethod(). (Если exposed равно None, то вместо него используется proxytype._exposed_, если он существует.) Если список exposed не указан, то все «публичные методы» разделяемого объекта будут доступны. (Здесь «публичный метод» означает любой атрибут, у которого есть метод __call__() и имя которого не начинается с '_'.)

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

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

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

address¶

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

class multiprocessing.managers.SyncManager¶

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

Он также поддерживает создание разделяемых списков и словарей.

BoundedSemaphore([value])¶

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

Condition([lock])¶

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

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

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

Примечание

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

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

class multiprocessing.managers.Namespace¶

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

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

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

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

from multiprocessing.managers import BaseManager

class MathsClass(object):
    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__':
    manager = MyManager()
    manager.start()
    maths = manager.Maths()
    print maths.add(4, 3)         # выводит 7
    print maths.mul(7, 8)         # выводит 56

>>> from multiprocessing.managers import BaseManager
>>> import Queue
>>> queue = Queue.Queue()
>>> class QueueManager(BaseManager): pass
>>> QueueManager.register('get_queue', callable=lambda:queue)
>>> m = QueueManager(address=('', 50000), authkey='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='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='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='abracadabra')
>>> s = m.get_server()
>>> s.serve_forever()

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

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

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

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

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

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

>>> a = manager.list()
>>> b = manager.list()
>>> a.append(b)         # референт a теперь содержит референт b
>>> print a, b
[[]] []
>>> b.append('hello')
>>> print a, b
[['hello']] ['hello']

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

class multiprocessing.managers.BaseProxy¶

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

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

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

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

proxy._callmethod(methodname, args, kwds)

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

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

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

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

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

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

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

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

_getvalue()¶

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

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

__repr__()¶

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

__str__()¶

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

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

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

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

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

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

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

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

Примечание

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

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

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

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

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

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

map(func, iterable[, chunksize])¶

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

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

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

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

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

imap(func, iterable[, chunksize])¶

An equivalent of itertools.imap().

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

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

imap_unordered(func, iterable[, chunksize])¶

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

close()¶

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

terminate()¶

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

join()¶

Ожидает завершения рабочих процессов. Перед использованием join() необходимо вызвать close() или 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__':
    pool = Pool(processes=4)              # запустить 4 рабочих процесса

    result = pool.apply_async(f, (10,))   # вычислить "f(10)" асинхронно в одном процессе
    print result.get(timeout=1)           # выводит "100", если компьютер не *очень* медленный

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

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

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

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

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

Однако модуль multiprocessing.connection обеспечивает некоторую дополнительную гибкость. По сути, он предоставляет высокоуровневый API, ориентированный на сообщения, для работы с сокетами или именованными каналами Windows, а также поддерживает дайджест-аутентификацию с помощью модуля hmac.

multiprocessing.connection.deliver_challenge(connection, authkey)¶

Отправляет случайно сгенерированное сообщение на другой конец соединения и ожидает ответа.

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

multiprocessing.connection.answer_challenge(connection, authkey)¶

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

Если приветственное сообщение не получено, то AuthenticationError возбуждается.

multiprocessing.connection.Client(address[, family[, authenticate[, authkey]]])¶

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

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

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

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

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

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

Примечание

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

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

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

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

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

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

accept()¶

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

close()¶

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

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

address¶

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

last_accepted¶

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

Модуль определяет следующие исключения:

exception multiprocessing.connection.ProcessError¶

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

exception multiprocessing.connection.BufferTooShort¶

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

exception multiprocessing.connection.AuthenticationError¶

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

exception multiprocessing.connection.TimeoutError¶

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

Примеры

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

from multiprocessing.connection import Listener
from array import array

address = ('localhost', 6000)     # семейство определяется как 'AF_INET'
listener = Listener(address, authkey='secret password')

conn = listener.accept()
print 'connection accepted from', listener.last_accepted

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

conn.send_bytes('hello')

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

conn.close()
listener.close()

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

from multiprocessing.connection import Client
from array import array

address = ('localhost', 6000)
conn = Client(address, authkey='secret password')

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

conn.close()

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

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

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

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

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

16.6.2.12. Логирование¶Logging

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

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

Полную таблицу уровней журналирования см. в модуле logging.

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

>>> import multiprocessing, logging
>>> logger = multiprocessing.log_to_stderr()
>>> logger.setLevel(multiprocessing.SUBDEBUG)
>>> 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 '/.../pymp-djGBXN/listener-...'
>>> del m
[SUBDEBUG/MainProcess] finalizer calling ...
[INFO/MainProcess] sending shutdown message to manager
[DEBUG/SyncManager-...] manager received shutdown message
[SUBDEBUG/SyncManager-...] calling <Finalize object, callback=unlink, ...
[SUBDEBUG/SyncManager-...] finalizer calling <built-in function unlink> ...
[SUBDEBUG/SyncManager-...] calling <Finalize object, dead>
[SUBDEBUG/SyncManager-...] finalizer calling <function rmtree at 0x5aa730> ...
[INFO/SyncManager-...] manager exiting with exitcode 0

16.6.2.13. Модуль multiprocessing.dummy¶The multiprocessing.dummy module

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

16.6.3.1. Все платформы¶All platforms

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Пример, который приведёт к взаимоблокировке:

from multiprocessing import Process, Queue

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

if __name__ == '__main__':
    queue = Queue()
    p = Process(target=f, args=(queue,))
    p.start()
    p.join()                    # это приводит к взаимоблокировке
    obj = queue.get()

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

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

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

Помимо обеспечения (потенциальной) совместимости с 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.devnull)

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

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

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

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

16.6.3.2. Windows¶

Поскольку в Windows отсутствует os.fork(), у него есть несколько дополнительных ограничений:

Упаковываемость

Убедитесь, что все аргументы Process.__init__() сериализуемы (picklable). Это означает, в частности, что связанные или несвязанные методы не могут использоваться напрямую в качестве аргумента target в Windows – просто определите функцию и используйте её вместо этого.

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

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

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

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

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

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

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

from multiprocessing import Process

def foo():
    print 'hello'

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

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

from multiprocessing import Process, freeze_support

def foo():
    print 'hello'

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

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

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

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