multiprocessing — Интерфейс «потоков», основанный на процессах

1. Введение

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

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

from multiprocessing import Pool

def f(x):
    return x*x

if __name__ == '__main__':
    p = Pool(5)
    print(p.map(f, [1, 2, 3]))

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

[1, 4, 9]

1.1. Класс Process

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

from multiprocessing import Process

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

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

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

from multiprocessing import Process
import os

def info(title):
    print title
    print 'module name:', __name__
    if hasattr(os, 'getppid'):  # only available on 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__', см. Рекомендации по программированию.

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

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

Очереди

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

from multiprocessing import Process, Queue

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

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

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

Трубы

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

from multiprocessing import Process, Pipe

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

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

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

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

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

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

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

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

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

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

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

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

Например:

from multiprocessing import Pool, TimeoutError
import time
import os

def f(x):
    return x*x

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

run()

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

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

start()

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

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

join([timeout])

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

Если timeout равен None, то ожидания нет.

Процесс можно присоединить многократно.

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

name

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

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

is_alive()

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

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

daemon

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

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

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

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

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

pid

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

exitcode

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

authkey

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

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

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

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

terminate()

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

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

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

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

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

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

>>> import multiprocessing, time, signal
>>> p = multiprocessing.Process(target=time.sleep, args=(1000,))
>>> print p, p.is_alive()
<Process(Process-1, initial)> False
>>> p.start()
>>> print p, p.is_alive()
<Process(Process-1, started)> True
>>> p.terminate()
>>> time.sleep(0.1)
>>> print p, p.is_alive()
<Process(Process-1, stopped[SIGTERM])> False
>>> p.exitcode == -signal.SIGTERM
True

exception multiprocessing.BufferTooShort

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

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

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

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

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

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

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

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

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

multiprocessing.Pipe([duplex])

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

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

class multiprocessing.Queue([maxsize])

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

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

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

qsize()

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

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

empty()

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

full()

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

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

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

put_nowait(obj)

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

get([block[, timeout]])

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

get_nowait()

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

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

close()

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

join_thread()

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

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

cancel_join_thread()

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

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

Примечание

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

class multiprocessing.queues.SimpleQueue

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

empty()

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

get()

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

put(item)

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

class multiprocessing.JoinableQueue([maxsize])

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

task_done()

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

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

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

join()

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

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

2.3. Разное

multiprocessing.active_children()

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

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

multiprocessing.cpu_count()

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

multiprocessing.current_process()

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

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

multiprocessing.freeze_support()

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

Необходимо вызвать эту функцию сразу после строки if __name__ == '__main__' в основном модуле. Например:

from multiprocessing import Process, freeze_support

def f():
    print 'hello world!'

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

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

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

multiprocessing.set_executable()

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

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

(только Windows)

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

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

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

class Connection

send(obj)

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

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

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

Например:

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

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

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

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

class multiprocessing.BoundedSemaphore([value])

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

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

Примечание

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

class multiprocessing.Condition([lock])

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

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

class multiprocessing.Event

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

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

class multiprocessing.Lock

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

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

Lock поддерживает протокол контекстного менеджера и, таким образом, может быть использован в операторах with.

acquire(block=True, timeout=None)

Приобретение блокировки, блокирующим или неблокирующим образом.

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

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

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

release()

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

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

class multiprocessing.RLock

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

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

RLock поддерживает протокол контекстного менеджера и, таким образом, может быть использован в операторах with.

acquire(block=True, timeout=None)

Приобретение блокировки, блокирующим или неблокирующим образом.

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

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

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

release()

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

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

class multiprocessing.Semaphore([value])

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

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

2.6. Объекты ctypes в общей памяти

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

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

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

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

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

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

counter.value += 1

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

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

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

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

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

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

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

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

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

2.6.1. Модуль multiprocessing.sharedctypes

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

multiprocessing.sharedctypes.RawArray(typecode_or_type, size_or_initializer)

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

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

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

multiprocessing.sharedctypes.RawValue(typecode_or_type, *args)

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

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

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

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

multiprocessing.sharedctypes.Array(typecode_or_type, size_or_initializer, *args[, lock])

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

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

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

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

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

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

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

multiprocessing.sharedctypes.copy(obj)

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Выведенные результаты:

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

2.7. Менеджеры

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

multiprocessing.Manager()

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

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

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

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

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

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

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

start([initializer[, initargs]])

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

get_server()

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

>>> from multiprocessing.managers import BaseManager
>>> manager = BaseManager(address=('', 50000), authkey='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 и возвращает для него прокси.

Если передан lock, то он должен быть прокси для объекта 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 и возвращает для него прокси.

Примечание

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

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

class multiprocessing.managers.Namespace

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

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

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

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

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

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

from multiprocessing.managers import BaseManager

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

class MyManager(BaseManager):
    pass

MyManager.register('Maths', MathsClass)

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

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

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

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

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

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

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

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

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

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

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

>>> a = manager.list()
>>> b = manager.list()
>>> a.append(b)         # referent of a now contains referent of b
>>> print a, b
[[]] []
>>> 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))   # equiv to `l[2:7]`
[2, 3, 4, 5, 6]
>>> l._callmethod('__getitem__', (20,))     # equiv to `l[20]`
Traceback (most recent call last):
...
IndexError: list index out of range

_getvalue()

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

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

__repr__()

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

__str__()

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

2.8.1. Очистка

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

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

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

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

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

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

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

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

Введено в версии 2.7: maxtasksperchild — количество задач, которое может выполнить рабочий процесс, прежде чем он завершится и будет заменен новым рабочим процессом, чтобы освободить неиспользуемые ресурсы. По умолчанию maxtasksperchild равно None, что означает, что рабочие процессы будут существовать до тех пор, пока существует пул.

Примечание

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

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

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

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

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

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

map(func, iterable[, chunksize])

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

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

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

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

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

imap(func, iterable[, chunksize])

Эквивалент itertools.imap().

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

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

imap_unordered(func, iterable[, chunksize])

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

close()

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

terminate()

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

join()

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

class multiprocessing.pool.AsyncResult

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

get([timeout])

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

wait([timeout])

Дожидается готовности результата или истечения timeout секунд.

ready()

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

successful()

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

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

from multiprocessing import Pool
import time

def f(x):
    return x*x

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

    result = pool.apply_async(f, (10,))   # evaluate "f(10)" asynchronously in a single process
    print result.get(timeout=1)           # prints "100" unless your computer is *very* slow

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

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

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

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

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

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

multiprocessing.connection.deliver_challenge(connection, authkey)

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

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

multiprocessing.connection.answer_challenge(connection, authkey)

Получает сообщение, вычисляет хэш-сумму сообщения с использованием authkey в качестве ключа и отправляет хэш-сумму обратно.

Если сообщение приветствия не получено, генерируется исключение AuthenticationError.

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

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

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

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

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

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

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

Примечание

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

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

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

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

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

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

accept()

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

close()

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

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

address

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

last_accepted

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

В модуле определены следующие исключения:

exception multiprocessing.connection.ProcessError

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

exception multiprocessing.connection.BufferTooShort

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

exception multiprocessing.connection.AuthenticationError

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

exception multiprocessing.connection.TimeoutError

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

Примеры

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

from multiprocessing.connection import Listener
from array import array

address = ('localhost', 6000)     # family is deduced to be 'AF_INET'
listener = Listener(address, authkey='secret password')

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

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

conn.send_bytes('hello')

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

conn.close()
listener.close()

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

from multiprocessing.connection import Client
from array import array

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

print conn.recv()                 # => [2.25, None, 'junk', float]

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

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

conn.close()

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

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

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

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

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

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

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

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

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

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

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

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

multiprocessing.get_logger()

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

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

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

multiprocessing.log_to_stderr()

Эта функция вызывает get_logger(), но помимо возвращения созданного логгера, добавляет обработчик, который отправляет вывод в 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

2.13. Модуль multiprocessing.dummy

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

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

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

3.1. Все платформы

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

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

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

Пикловаемость

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

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

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

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

Присоединение к завершённым процессам-зомби

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

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

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

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

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

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

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

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

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

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

from multiprocessing import Process, Queue

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

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

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

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

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

Помимо обеспечения (возможно) совместимости с 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() на этом объекте типа файла, это может привести к тому, что одни и те же данные будут несколько раз сброшены в объект, что приведёт к повреждению данных.

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

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

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

3.2. Windows

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

Более высокая пикловаемость

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

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

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

4. Примеры

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

#
# This module shows how to use arbitrary callables with a subclass of
# `BaseManager`.
#
# Copyright (c) 2006-2008, R Oudkerk
# All rights reserved.
#

from multiprocessing import freeze_support
from multiprocessing.managers import BaseManager, BaseProxy
import operator

##

class Foo(object):
    def f(self):
        print 'you called Foo.f()'
    def g(self):
        print 'you called Foo.g()'
    def _h(self):
        print 'you called Foo._h()'

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

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

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

##

class MyManager(BaseManager):
    pass

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

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

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

# register get_operator_module(); make public functions accessible via proxy
MyManager.register('operator', get_operator_module)

##

def test():
    manager = MyManager()
    manager.start()

    print '-' * 20

    f1 = manager.Foo1()
    f1.f()
    f1.g()
    assert not hasattr(f1, '_h')
    assert sorted(f1._exposed_) == sorted(['f', 'g'])

    print '-' * 20

    f2 = manager.Foo2()
    f2.g()
    f2._h()
    assert not hasattr(f2, 'f')
    assert sorted(f2._exposed_) == sorted(['g', '_h'])

    print '-' * 20

    it = manager.baz()
    for i in it:
        print '<%d>' % i,
    print

    print '-' * 20

    op = manager.operator()
    print 'op.add(23, 45) =', op.add(23, 45)
    print 'op.pow(2, 94) =', op.pow(2, 94)
    print 'op.getslice(range(10), 2, 6) =', op.getslice(range(10), 2, 6)
    print 'op.repeat(range(5), 3) =', op.repeat(range(5), 3)
    print 'op._exposed_ =', op._exposed_

##

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

Использование Pool:

#
# A test of `multiprocessing.Pool` class
#
# Copyright (c) 2006-2008, R Oudkerk
# All rights reserved.
#

import multiprocessing
import time
import random
import sys

#
# Functions used by test code
#

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

def calculatestar(args):
    return calculate(*args)

def mul(a, b):
    time.sleep(0.5*random.random())
    return a * b

def plus(a, b):
    time.sleep(0.5*random.random())
    return a + b

def f(x):
    return 1.0 / (x-5.0)

def pow3(x):
    return x**3

def noop(x):
    pass

#
# Test code
#

def test():
    print 'cpu_count() = %d\n' % multiprocessing.cpu_count()

    #
    # Create pool
    #

    PROCESSES = 4
    print 'Creating pool with %d processes\n' % PROCESSES
    pool = multiprocessing.Pool(PROCESSES)
    print 'pool = %s' % pool
    print

    #
    # Tests
    #

    TASKS = [(mul, (i, 7)) for i in range(10)] + \
            [(plus, (i, 8)) for i in range(10)]

    results = [pool.apply_async(calculate, t) for t in TASKS]
    imap_it = pool.imap(calculatestar, TASKS)
    imap_unordered_it = pool.imap_unordered(calculatestar, TASKS)

    print 'Ordered results using pool.apply_async():'
    for r in results:
        print '\t', r.get()
    print

    print 'Ordered results using pool.imap():'
    for x in imap_it:
        print '\t', x
    print

    print 'Unordered results using pool.imap_unordered():'
    for x in imap_unordered_it:
        print '\t', x
    print

    print 'Ordered results using pool.map() --- will block till complete:'
    for x in pool.map(calculatestar, TASKS):
        print '\t', x
    print

    #
    # Simple benchmarks
    #

    N = 100000
    print 'def pow3(x): return x**3'

    t = time.time()
    A = map(pow3, xrange(N))
    print '\tmap(pow3, xrange(%d)):\n\t\t%s seconds' % \
          (N, time.time() - t)

    t = time.time()
    B = pool.map(pow3, xrange(N))
    print '\tpool.map(pow3, xrange(%d)):\n\t\t%s seconds' % \
          (N, time.time() - t)

    t = time.time()
    C = list(pool.imap(pow3, xrange(N), chunksize=N//8))
    print '\tlist(pool.imap(pow3, xrange(%d), chunksize=%d)):\n\t\t%s' \
          ' seconds' % (N, N//8, time.time() - t)

    assert A == B == C, (len(A), len(B), len(C))
    print

    L = [None] * 1000000
    print 'def noop(x): pass'
    print 'L = [None] * 1000000'

    t = time.time()
    A = map(noop, L)
    print '\tmap(noop, L):\n\t\t%s seconds' % \
          (time.time() - t)

    t = time.time()
    B = pool.map(noop, L)
    print '\tpool.map(noop, L):\n\t\t%s seconds' % \
          (time.time() - t)

    t = time.time()
    C = list(pool.imap(noop, L, chunksize=len(L)//8))
    print '\tlist(pool.imap(noop, L, chunksize=%d)):\n\t\t%s seconds' % \
          (len(L)//8, time.time() - t)

    assert A == B == C, (len(A), len(B), len(C))
    print

    del A, B, C, L

    #
    # Test error handling
    #

    print 'Testing error handling:'

    try:
        print pool.apply(f, (5,))
    except ZeroDivisionError:
        print '\tGot ZeroDivisionError as expected from pool.apply()'
    else:
        raise AssertionError('expected ZeroDivisionError')

    try:
        print pool.map(f, range(10))
    except ZeroDivisionError:
        print '\tGot ZeroDivisionError as expected from pool.map()'
    else:
        raise AssertionError('expected ZeroDivisionError')

    try:
        print list(pool.imap(f, range(10)))
    except ZeroDivisionError:
        print '\tGot ZeroDivisionError as expected from list(pool.imap())'
    else:
        raise AssertionError('expected ZeroDivisionError')

    it = pool.imap(f, range(10))
    for i in range(10):
        try:
            x = it.next()
        except ZeroDivisionError:
            if i == 5:
                pass
        except StopIteration:
            break
        else:
            if i == 5:
                raise AssertionError('expected ZeroDivisionError')

    assert i == 9
    print '\tGot ZeroDivisionError as expected from IMapIterator.next()'
    print

    #
    # Testing timeouts
    #

    print 'Testing ApplyResult.get() with timeout:',
    res = pool.apply_async(calculate, TASKS[0])
    while 1:
        sys.stdout.flush()
        try:
            sys.stdout.write('\n\t%s' % res.get(0.02))
            break
        except multiprocessing.TimeoutError:
            sys.stdout.write('.')
    print
    print

    print 'Testing IMapIterator.next() with timeout:',
    it = pool.imap(calculatestar, TASKS)
    while 1:
        sys.stdout.flush()
        try:
            sys.stdout.write('\n\t%s' % it.next(0.02))
        except StopIteration:
            break
        except multiprocessing.TimeoutError:
            sys.stdout.write('.')
    print
    print

    #
    # Testing callback
    #

    print 'Testing callback:'

    A = []
    B = [56, 0, 1, 8, 27, 64, 125, 216, 343, 512, 729]

    r = pool.apply_async(mul, (7, 8), callback=A.append)
    r.wait()

    r = pool.map_async(pow3, range(10), callback=A.extend)
    r.wait()

    if A == B:
        print '\tcallbacks succeeded\n'
    else:
        print '\t*** callbacks failed\n\t\t%s != %s\n' % (A, B)

    #
    # Check there are no outstanding tasks
    #

    assert not pool._cache, 'cache = %r' % pool._cache

    #
    # Check close() methods
    #

    print 'Testing close():'

    for worker in pool._pool:
        assert worker.is_alive()

    result = pool.apply_async(time.sleep, [0.5])
    pool.close()
    pool.join()

    assert result.get() is None

    for worker in pool._pool:
        assert not worker.is_alive()

    print '\tclose() succeeded\n'

    #
    # Check terminate() method
    #

    print 'Testing terminate():'

    pool = multiprocessing.Pool(2)
    DELTA = 0.1
    ignore = pool.apply(pow3, [2])
    results = [pool.apply_async(time.sleep, [DELTA]) for i in range(100)]
    pool.terminate()
    pool.join()

    for worker in pool._pool:
        assert not worker.is_alive()

    print '\tterminate() succeeded\n'

    #
    # Check garbage collection
    #

    print 'Testing garbage collection:'

    pool = multiprocessing.Pool(2)
    DELTA = 0.1
    processes = pool._pool
    ignore = pool.apply(pow3, [2])
    results = [pool.apply_async(time.sleep, [DELTA]) for i in range(100)]

    results = pool = None

    time.sleep(DELTA * 2)

    for worker in processes:
        assert not worker.is_alive()

    print '\tgarbage collection succeeded\n'


if __name__ == '__main__':
    multiprocessing.freeze_support()

    assert len(sys.argv) in (1, 2)

    if len(sys.argv) == 1 or sys.argv[1] == 'processes':
        print ' Using processes '.center(79, '-')
    elif sys.argv[1] == 'threads':
        print ' Using threads '.center(79, '-')
        import multiprocessing.dummy as multiprocessing
    else:
        print 'Usage:\n\t%s [processes | threads]' % sys.argv[0]
        raise SystemExit(2)

    test()

Типы синхронизации, такие как блокировки, условия и очереди:

#
# A test file for the `multiprocessing` package
#
# Copyright (c) 2006-2008, R Oudkerk
# All rights reserved.
#

import time, sys, random
from Queue import Empty

import multiprocessing               # may get overwritten


#### TEST_VALUE

def value_func(running, mutex):
    random.seed()
    time.sleep(random.random()*4)

    mutex.acquire()
    print '\n\t\t\t' + str(multiprocessing.current_process()) + ' has finished'
    running.value -= 1
    mutex.release()

def test_value():
    TASKS = 10
    running = multiprocessing.Value('i', TASKS)
    mutex = multiprocessing.Lock()

    for i in range(TASKS):
        p = multiprocessing.Process(target=value_func, args=(running, mutex))
        p.start()

    while running.value > 0:
        time.sleep(0.08)
        mutex.acquire()
        print running.value,
        sys.stdout.flush()
        mutex.release()

    print
    print 'No more running processes'


#### TEST_QUEUE

def queue_func(queue):
    for i in range(30):
        time.sleep(0.5 * random.random())
        queue.put(i*i)
    queue.put('STOP')

def test_queue():
    q = multiprocessing.Queue()

    p = multiprocessing.Process(target=queue_func, args=(q,))
    p.start()

    o = None
    while o != 'STOP':
        try:
            o = q.get(timeout=0.3)
            print o,
            sys.stdout.flush()
        except Empty:
            print 'TIMEOUT'

    print


#### TEST_CONDITION

def condition_func(cond):
    cond.acquire()
    print '\t' + str(cond)
    time.sleep(2)
    print '\tchild is notifying'
    print '\t' + str(cond)
    cond.notify()
    cond.release()

def test_condition():
    cond = multiprocessing.Condition()

    p = multiprocessing.Process(target=condition_func, args=(cond,))
    print cond

    cond.acquire()
    print cond
    cond.acquire()
    print cond

    p.start()

    print 'main is waiting'
    cond.wait()
    print 'main has woken up'

    print cond
    cond.release()
    print cond
    cond.release()

    p.join()
    print cond


#### TEST_SEMAPHORE

def semaphore_func(sema, mutex, running):
    sema.acquire()

    mutex.acquire()
    running.value += 1
    print running.value, 'tasks are running'
    mutex.release()

    random.seed()
    time.sleep(random.random()*2)

    mutex.acquire()
    running.value -= 1
    print '%s has finished' % multiprocessing.current_process()
    mutex.release()

    sema.release()

def test_semaphore():
    sema = multiprocessing.Semaphore(3)
    mutex = multiprocessing.RLock()
    running = multiprocessing.Value('i', 0)

    processes = [
        multiprocessing.Process(target=semaphore_func,
                                args=(sema, mutex, running))
        for i in range(10)
        ]

    for p in processes:
        p.start()

    for p in processes:
        p.join()


#### TEST_JOIN_TIMEOUT

def join_timeout_func():
    print '\tchild sleeping'
    time.sleep(5.5)
    print '\n\tchild terminating'

def test_join_timeout():
    p = multiprocessing.Process(target=join_timeout_func)
    p.start()

    print 'waiting for process to finish'

    while 1:
        p.join(timeout=1)
        if not p.is_alive():
            break
        print '.',
        sys.stdout.flush()


#### TEST_EVENT

def event_func(event):
    print '\t%r is waiting' % multiprocessing.current_process()
    event.wait()
    print '\t%r has woken up' % multiprocessing.current_process()

def test_event():
    event = multiprocessing.Event()

    processes = [multiprocessing.Process(target=event_func, args=(event,))
                 for i in range(5)]

    for p in processes:
        p.start()

    print 'main is sleeping'
    time.sleep(2)

    print 'main is setting event'
    event.set()

    for p in processes:
        p.join()


#### TEST_SHAREDVALUES

def sharedvalues_func(values, arrays, shared_values, shared_arrays):
    for i in range(len(values)):
        v = values[i][1]
        sv = shared_values[i].value
        assert v == sv

    for i in range(len(values)):
        a = arrays[i][1]
        sa = list(shared_arrays[i][:])
        assert a == sa

    print 'Tests passed'

def test_sharedvalues():
    values = [
        ('i', 10),
        ('h', -2),
        ('d', 1.25)
        ]
    arrays = [
        ('i', range(100)),
        ('d', [0.25 * i for i in range(100)]),
        ('H', range(1000))
        ]

    shared_values = [multiprocessing.Value(id, v) for id, v in values]
    shared_arrays = [multiprocessing.Array(id, a) for id, a in arrays]

    p = multiprocessing.Process(
        target=sharedvalues_func,
        args=(values, arrays, shared_values, shared_arrays)
        )
    p.start()
    p.join()

    assert p.exitcode == 0


####

def test(namespace=multiprocessing):
    global multiprocessing

    multiprocessing = namespace

    for func in [ test_value, test_queue, test_condition,
                  test_semaphore, test_join_timeout, test_event,
                  test_sharedvalues ]:

        print '\n\t######## %s\n' % func.__name__
        func()

    ignore = multiprocessing.active_children()      # cleanup any old processes
    if hasattr(multiprocessing, '_debug_info'):
        info = multiprocessing._debug_info()
        if info:
            print info
            raise ValueError('there should be no positive refcounts left')


if __name__ == '__main__':
    multiprocessing.freeze_support()

    assert len(sys.argv) in (1, 2)

    if len(sys.argv) == 1 or sys.argv[1] == 'processes':
        print ' Using processes '.center(79, '-')
        namespace = multiprocessing
    elif sys.argv[1] == 'manager':
        print ' Using processes and a manager '.center(79, '-')
        namespace = multiprocessing.Manager()
        namespace.Process = multiprocessing.Process
        namespace.current_process = multiprocessing.current_process
        namespace.active_children = multiprocessing.active_children
    elif sys.argv[1] == 'threads':
        print ' Using threads '.center(79, '-')
        import multiprocessing.dummy as namespace
    else:
        print 'Usage:\n\t%s [processes | manager | threads]' % sys.argv[0]
        raise SystemExit(2)

    test(namespace)

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

#
# Simple example which uses a pool of workers to carry out some tasks.
#
# Notice that the results will probably not come out of the output
# queue in the same in the same order as the corresponding tasks were
# put on the input queue.  If it is important to get the results back
# in the original order then consider using `Pool.map()` or
# `Pool.imap()` (which will save on the amount of code needed anyway).
#
# Copyright (c) 2006-2008, R Oudkerk
# All rights reserved.
#

import time
import random

from multiprocessing import Process, Queue, current_process, freeze_support

#
# Function run by worker processes
#

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

#
# Function used to calculate result
#

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

#
# Functions referenced by tasks
#

def mul(a, b):
    time.sleep(0.5*random.random())
    return a * b

def plus(a, b):
    time.sleep(0.5*random.random())
    return a + b

#
#
#

def test():
    NUMBER_OF_PROCESSES = 4
    TASKS1 = [(mul, (i, 7)) for i in range(20)]
    TASKS2 = [(plus, (i, 8)) for i in range(10)]

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

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

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

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

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

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

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


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

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

#
# Example where a pool of http servers share a single listening socket
#
# On Windows this module depends on the ability to pickle a socket
# object so that the worker processes can inherit a copy of the server
# object.  (We import `multiprocessing.reduction` to enable this pickling.)
#
# Not sure if we should synchronize access to `socket.accept()` method by
# using a process-shared lock -- does not seem to be necessary.
#
# Copyright (c) 2006-2008, R Oudkerk
# All rights reserved.
#

import os
import sys

from multiprocessing import Process, current_process, freeze_support
from BaseHTTPServer import HTTPServer
from SimpleHTTPServer import SimpleHTTPRequestHandler

if sys.platform == 'win32':
    import multiprocessing.reduction    # make sockets pickable/inheritable


def note(format, *args):
    sys.stderr.write('[%s]\t%s\n' % (current_process().name, format%args))


class RequestHandler(SimpleHTTPRequestHandler):
    # we override log_message() to show which process is handling the request
    def log_message(self, format, *args):
        note(format, *args)

def serve_forever(server):
    note('starting server')
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass


def runpool(address, number_of_processes):
    # create a single server object -- children will each inherit a copy
    server = HTTPServer(address, RequestHandler)

    # create child processes to act as workers
    for i in range(number_of_processes-1):
        Process(target=serve_forever, args=(server,)).start()

    # main process also acts as a worker
    serve_forever(server)


def test():
    DIR = os.path.join(os.path.dirname(__file__), '..')
    ADDRESS = ('localhost', 8000)
    NUMBER_OF_PROCESSES = 4

    print 'Serving at http://%s:%d using %d worker processes' % \
          (ADDRESS[0], ADDRESS[1], NUMBER_OF_PROCESSES)
    print 'To exit press Ctrl-' + ['C', 'Break'][sys.platform=='win32']

    os.chdir(DIR)
    runpool(ADDRESS, NUMBER_OF_PROCESSES)


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

Некоторые простые бенчмарки, сравнивающие multiprocessing с threading:

#
# Simple benchmarks for the multiprocessing package
#
# Copyright (c) 2006-2008, R Oudkerk
# All rights reserved.
#

import time, sys, multiprocessing, threading, Queue, gc

if sys.platform == 'win32':
    _timer = time.clock
else:
    _timer = time.time

delta = 1


#### TEST_QUEUESPEED

def queuespeed_func(q, c, iterations):
    a = '0' * 256
    c.acquire()
    c.notify()
    c.release()

    for i in xrange(iterations):
        q.put(a)

    q.put('STOP')

def test_queuespeed(Process, q, c):
    elapsed = 0
    iterations = 1

    while elapsed < delta:
        iterations *= 2

        p = Process(target=queuespeed_func, args=(q, c, iterations))
        c.acquire()
        p.start()
        c.wait()
        c.release()

        result = None
        t = _timer()

        while result != 'STOP':
            result = q.get()

        elapsed = _timer() - t

        p.join()

    print iterations, 'objects passed through the queue in', elapsed, 'seconds'
    print 'average number/sec:', iterations/elapsed


#### TEST_PIPESPEED

def pipe_func(c, cond, iterations):
    a = '0' * 256
    cond.acquire()
    cond.notify()
    cond.release()

    for i in xrange(iterations):
        c.send(a)

    c.send('STOP')

def test_pipespeed():
    c, d = multiprocessing.Pipe()
    cond = multiprocessing.Condition()
    elapsed = 0
    iterations = 1

    while elapsed < delta:
        iterations *= 2

        p = multiprocessing.Process(target=pipe_func,
                                    args=(d, cond, iterations))
        cond.acquire()
        p.start()
        cond.wait()
        cond.release()

        result = None
        t = _timer()

        while result != 'STOP':
            result = c.recv()

        elapsed = _timer() - t
        p.join()

    print iterations, 'objects passed through connection in',elapsed,'seconds'
    print 'average number/sec:', iterations/elapsed


#### TEST_SEQSPEED

def test_seqspeed(seq):
    elapsed = 0
    iterations = 1

    while elapsed < delta:
        iterations *= 2

        t = _timer()

        for i in xrange(iterations):
            a = seq[5]

        elapsed = _timer()-t

    print iterations, 'iterations in', elapsed, 'seconds'
    print 'average number/sec:', iterations/elapsed


#### TEST_LOCK

def test_lockspeed(l):
    elapsed = 0
    iterations = 1

    while elapsed < delta:
        iterations *= 2

        t = _timer()

        for i in xrange(iterations):
            l.acquire()
            l.release()

        elapsed = _timer()-t

    print iterations, 'iterations in', elapsed, 'seconds'
    print 'average number/sec:', iterations/elapsed


#### TEST_CONDITION

def conditionspeed_func(c, N):
    c.acquire()
    c.notify()

    for i in xrange(N):
        c.wait()
        c.notify()

    c.release()

def test_conditionspeed(Process, c):
    elapsed = 0
    iterations = 1

    while elapsed < delta:
        iterations *= 2

        c.acquire()
        p = Process(target=conditionspeed_func, args=(c, iterations))
        p.start()

        c.wait()

        t = _timer()

        for i in xrange(iterations):
            c.notify()
            c.wait()

        elapsed = _timer()-t

        c.release()
        p.join()

    print iterations * 2, 'waits in', elapsed, 'seconds'
    print 'average number/sec:', iterations * 2 / elapsed

####

def test():
    manager = multiprocessing.Manager()

    gc.disable()

    print '\n\t######## testing Queue.Queue\n'
    test_queuespeed(threading.Thread, Queue.Queue(),
                    threading.Condition())
    print '\n\t######## testing multiprocessing.Queue\n'
    test_queuespeed(multiprocessing.Process, multiprocessing.Queue(),
                    multiprocessing.Condition())
    print '\n\t######## testing Queue managed by server process\n'
    test_queuespeed(multiprocessing.Process, manager.Queue(),
                    manager.Condition())
    print '\n\t######## testing multiprocessing.Pipe\n'
    test_pipespeed()

    print

    print '\n\t######## testing list\n'
    test_seqspeed(range(10))
    print '\n\t######## testing list managed by server process\n'
    test_seqspeed(manager.list(range(10)))
    print '\n\t######## testing Array("i", ..., lock=False)\n'
    test_seqspeed(multiprocessing.Array('i', range(10), lock=False))
    print '\n\t######## testing Array("i", ..., lock=True)\n'
    test_seqspeed(multiprocessing.Array('i', range(10), lock=True))

    print

    print '\n\t######## testing threading.Lock\n'
    test_lockspeed(threading.Lock())
    print '\n\t######## testing threading.RLock\n'
    test_lockspeed(threading.RLock())
    print '\n\t######## testing multiprocessing.Lock\n'
    test_lockspeed(multiprocessing.Lock())
    print '\n\t######## testing multiprocessing.RLock\n'
    test_lockspeed(multiprocessing.RLock())
    print '\n\t######## testing lock managed by server process\n'
    test_lockspeed(manager.Lock())
    print '\n\t######## testing rlock managed by server process\n'
    test_lockspeed(manager.RLock())

    print

    print '\n\t######## testing threading.Condition\n'
    test_conditionspeed(threading.Thread, threading.Condition())
    print '\n\t######## testing multiprocessing.Condition\n'
    test_conditionspeed(multiprocessing.Process, multiprocessing.Condition())
    print '\n\t######## testing condition managed by a server process\n'
    test_conditionspeed(multiprocessing.Process, manager.Condition())

    gc.enable()

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