17.2. multiprocessing – Процессно-ориентированный параллелизм¶multiprocessing – Process-based parallelism

17.2.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__', объясняется в разделе Рекомендации по программированию.

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

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

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

17.2.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, Блокировка, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Очередь, 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.5. Использование пула рабочих процессов¶Using a pool of workers

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

Например:

from multiprocessing import Pool

def f(x):
    return x*x

if __name__ == '__main__':
    with Pool(processes=4) as pool:        # запустить 4 рабочих процесса
        result = pool.apply_async(f, [10]) # асинхронно вычислить "f(10)"
        print(result.get(timeout=1))       # выводит "100", если компьютер не *очень* медленный
        print(pool.map(f, range(10)))      # выводит "[0, 1, 4,..., 81]"

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

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

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

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

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

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

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

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

run()¶

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

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

start()¶

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

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

join([timeout])¶

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

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

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

name¶

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

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

is_alive()¶

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

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

daemon¶

Флаг 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. Каналы и очереди¶Pipes and Queues

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

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

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

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

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

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

multiprocessing.Pipe([duplex])¶

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

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

class multiprocessing.Queue([maxsize])¶

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

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

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

qsize()¶

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

Обратите внимание, что это может вызвать NotImplementedError на платформах Unix, таких как 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]])¶

Remove and return an item from the queue. If optional args block is True (the default) and timeout is None (the default), block if necessary until an item is available. If timeout is a positive number, it blocks at most timeout seconds and raises the queue.Empty exception if no item was available within that time. Otherwise (block is False), return an item if one is immediately available, else raise the queue.Empty exception (timeout is ignored in that case).

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

class multiprocessing.SimpleQueue¶

Упрощённый тип очереди, очень близкий к блокируемому Pipe.

empty()¶

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

get()¶

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

put(item)¶

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

class multiprocessing.JoinableQueue([maxsize])¶

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

task_done()¶

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

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

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

join()¶

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

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

17.2.2.3. Разное¶Miscellaneous

multiprocessing.active_children()¶

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

Вызов этой функции приводит к «присоединению» любых процессов, которые уже завершились.

multiprocessing.cpu_count()¶

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

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

Если модуль запускается обычным образом интерпретатором Python, то freeze_support() не действует.

multiprocessing.set_executable()¶

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

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

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

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

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

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

class multiprocessing.Connection¶

send(obj)¶

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

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

recv()¶

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

fileno()¶

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

close()¶

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

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

poll([timeout])¶

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

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

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

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

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

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

recv_bytes([maxlength])¶

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

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

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

recv_bytes_into(buffer[, offset])¶

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

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

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

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

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

Например:

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

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

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

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

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

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

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

class multiprocessing.BoundedSemaphore([value])¶

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

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

class multiprocessing.Condition([lock])¶

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

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

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

class multiprocessing.Event¶

Клон threading.Event.

class multiprocessing.Lock¶

Объект нерекурсивной блокировки: копия threading.Lock.

class multiprocessing.RLock¶

Объект рекурсивной блокировки: копия threading.RLock.

class multiprocessing.Semaphore([value])¶

Объект семафора: копия threading.Semaphore.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

17.2.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, представляющий реальный сервер под управлением Manager. Объект Server поддерживает метод serve_forever():

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

Server также имеет атрибут address.

connect()¶

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

>>> from multiprocessing.managers import BaseManager
>>> m = BaseManager(address=('127.0.0.1', 5000), authkey=b'abc')
>>> m.connect()

shutdown()¶

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

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

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

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

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

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

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

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

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

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

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

address¶

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

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

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

class multiprocessing.managers.SyncManager¶

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

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

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

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

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

BoundedSemaphore([value])¶

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

Condition([lock])¶

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

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

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

Event()¶

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

Lock()¶

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

Namespace()¶

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

Queue([maxsize])¶

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

RLock()¶

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

Semaphore([value])¶

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

Array(typecode, sequence)¶

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

Value(typecode, value)¶

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

dict()¶

dict(mapping)

dict(sequence)

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

list()¶

list(sequence)

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

Примечание

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

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

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

from multiprocessing.managers import BaseManager

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

class MyManager(BaseManager):
    pass

MyManager.register('Maths', MathsClass)

if __name__ == '__main__':
    with MyManager() as manager:
        maths = manager.Maths()
        print(maths.add(4, 3))         # выводит 7
        print(maths.mul(7, 8))         # выводит 56

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

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

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

>>> from multiprocessing import Process, Queue
>>> from multiprocessing.managers import BaseManager
>>> class Worker(Process):
...     def __init__(self, q):
...         self.q = q
...         super(Worker, self).__init__()
...     def run(self):
...         self.q.put('local hello')
...
>>> queue = Queue()
>>> w = Worker(queue)
>>> w.start()
>>> class QueueManager(BaseManager): pass
...
>>> QueueManager.register('get_queue', callable=lambda: queue)
>>> m = QueueManager(address=('', 50000), authkey=b'abracadabra')
>>> s = m.get_server()
>>> s.serve_forever()

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

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

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

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

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

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

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

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

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

Примечание

Рабочие процессы внутри 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(), который возвращает объект результата.

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

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

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

map(func, iterable[, chunksize])¶

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

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

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

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

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

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

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

imap(func, iterable[, chunksize])¶

Более ленивая версия map().

The chunksize argument is the same as the one used by the map() method. For very long iterables using a large value for chunksize can make the job complete much faster than using the default value of 1.

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

imap_unordered(func, iterable[, chunksize])¶

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

starmap(func, iterable[, chunksize])¶

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

Следовательно, iterable из [(1,2), (3, 4)] даёт результат [func(1,2), func(3,4)].

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

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

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

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

close()¶

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

terminate()¶

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

join()¶

Ожидает завершения рабочих процессов. Перед вызовом 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

def f(x):
    return x*x

if __name__ == '__main__':
    with Pool(processes=4) as pool:         # запустить 4 рабочих процесса
        result = pool.apply_async(f, (10,)) # асинхронно вычислить "f(10)"
        print(result.get(timeout=1))        # выводит "100", если компьютер не *очень* медленный

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

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

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

17.2.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. См. Authentication keys.

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, то выбирается значение по умолчанию. По умолчанию выбирается семейство, которое считается самым быстрым из доступных. См. Address Formats. Обратите внимание: если family равен 'AF_UNIX', а address равен None, то сокет будет создан в частной временной директории, созданной с помощью tempfile.mkstemp().

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

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

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

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

accept()¶

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

close()¶

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

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

address¶

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

last_accepted¶

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

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

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

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

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

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

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

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

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

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

Примеры

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

from multiprocessing.connection import Listener
from array import array

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

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

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

        conn.send_bytes(b'hello')

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

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

from multiprocessing.connection import Client
from array import array

address = ('localhost', 6000)

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

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

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

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

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

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

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

    for i in range(4):
        r, w = Pipe(duplex=False)
        readers.append(r)
        p = Process(target=foo, args=(w,))
        p.start()
        # Закрываем записывающий конец канала, чтобы убедиться, что
        # p – единственный процесс, владеющий дескриптором канала. Это
        # гарантирует, что когда p закроет свой дескриптор для записывающего конца,
        # wait() немедленно сообщит, что читающий конец готов
        w.close()

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

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

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

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

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

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

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

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

multiprocessing.get_logger()¶

Возвращает регистратор (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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

17.2.3.2. Windows¶

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

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

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

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

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