16.3. multiprocessing – параллелизм на основе процессов¶multiprocessing – Process-based parallelism

16.3.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__)
    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.3.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.3.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.3.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.3.1.5. Использование пула рабочих процессов¶Using a pool of workers

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

Например:

from multiprocessing import Pool

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

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

class multiprocessing.Process([group[, target[, name[, 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¶

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

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

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

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

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

pid¶

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

exitcode¶

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

authkey¶

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

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

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

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

terminate()¶

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

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

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

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

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

Пример использования некоторых методов 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.3.2.2. Каналы и очереди¶Pipes and Queues

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

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

Типы Queue и 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(item[, block[, timeout]])¶

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

put_nowait(item)¶

Эквивалентно put(item, 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_no_wait()¶

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

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

close()¶

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

join_thread()¶

Присоединяет фоновый поток. Используется только после вызова close(). Блокирует выполнение до завершения фонового потока, гарантируя, что все данные из буфера будут сброшены в канал.

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

cancel_join_thread()¶

Предотвращает блокирование join_thread(). В частности, это не даёт фоновому потоку автоматически присоединиться при завершении процесса – см. join_thread().

class multiprocessing.JoinableQueue([maxsize])¶

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

task_done()¶

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

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

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

join()¶

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

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

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

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

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

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

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

Объекты соединения обычно создаются с помощью Pipe() – см. также Listeners and Clients.

class multiprocessing.Connection¶

send(obj)¶

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

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

recv()¶

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

fileno()¶

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

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

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

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

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

class multiprocessing.BoundedSemaphore([value])¶

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

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

class multiprocessing.Condition([lock])¶

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

If lock is specified then it should be a Lock or RLock object from multiprocessing.

class multiprocessing.Event¶

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

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

class multiprocessing.Lock¶

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

class multiprocessing.RLock¶

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

class multiprocessing.Semaphore([value])¶

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

16.3.2.6. Общие объекты ctypes¶Shared ctypes Objects

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

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

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

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

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

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

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.3.2.7. Менеджеры¶Managers

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

multiprocessing.Manager()¶

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

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

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

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

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

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

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

start([initializer[, initargs]])¶

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

serve_forever()¶

Запускает сервер в текущем процессе.

get_server()¶

Возвращает объект Server, который представляет фактический сервер под управлением Manager. Объект 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

>>> 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__':
    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.3.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.3.2.9. Пулы процессов¶Process Pools

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

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

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

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

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

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

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

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

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

map(func, iterable[, chunksize])¶

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

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

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

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

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

imap(func, iterable[, chunksize])¶

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

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

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

imap_unordered(func, iterable[, chunksize])¶

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

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

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

16.3.2.10. Прослушиватели и клиенты¶Listeners and Clients

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

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

multiprocessing.connection.deliver_challenge(connection, authkey)¶

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

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

multiprocessing.connection.answerChallenge(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. См. Ключи аутентификации.

accept()¶

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

close()¶

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

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

address¶

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

last_accepted¶

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

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

exception multiprocessing.connection.AuthenticationError¶

Исключение, возникающее при ошибке аутентификации.

Примеры

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

from multiprocessing.connection import Listener
from array import array

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

conn = listener.accept()
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]))

conn.close()
listener.close()

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

from multiprocessing.connection import Client
from array import array

address = ('localhost', 6000)
conn = Client(address, authkey=b'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.3.2.11. Ключи аутентификации¶Authentication keys

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Имейте в виду, что процесс, поместивший элементы в очередь, будет ждать перед завершением, пока все буферизованные элементы не будут переданы потоком «feeder» в нижележащий канал. (Дочерний процесс может вызвать метод очереди 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() на «объект, подобный файлу» (file-like object) с буферизацией вывода. Опасность заключается в том, что если несколько процессов вызовут 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.

16.3.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() этого модуля.

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