multiprocessing.md

1> **Источник:** https://python-all.ru/3.0/library/multiprocessing.html2>3> «Документация Python на русском» – неофициальный перевод официальной документации Python: версии от 2.6 до 3.16, полнотекстовый поиск, английский оригинал рядом с переводом. Эта Markdown-версия страницы предназначена для работы с LLM: вставьте её в ChatGPT, Claude или Cursor.45---67# `multiprocessing` – интерфейс «потоков» на основе процессов89## Введение1011`multiprocessing` – это пакет, поддерживающий порождение процессов с помощью API, похожего на модуль [`threading`](https://python-all.ru/3.0/library/threading.html#module-threading). Пакет `multiprocessing` предоставляет как локальный, так и удалённый параллелизм, обходя [*глобальную блокировку интерпретатора*](https://python-all.ru/3.0/glossary.html#term-global-interpreter-lock) за счёт использования подпроцессов вместо потоков. Благодаря этому модуль `multiprocessing` позволяет программисту полностью задействовать несколько процессоров на данной машине. Он работает как под Unix, так и под Windows.1213> **Предупреждение**14>15> Часть функционала этого пакета требует работающей реализации разделяемого семафора в операционной системе. При её отсутствии модуль `multiprocessing.synchronize` будет отключён, а попытки его импорта приведут к [`ImportError`](https://python-all.ru/3.0/library/exceptions.html#exceptions.ImportError). Дополнительную информацию см. в [issue 3770](https://python-all.ru/3.0/library/multiprocessing.html).1617> **Примечание**18>19> Функциональность этого пакета требует, чтобы метод `__main__` был импортируем дочерними процессами. Это описано в [*Рекомендациях по программированию*](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing-programming), однако здесь стоит об этом упомянуть. Это означает, что некоторые примеры, такие как `multiprocessing.Pool`, не будут работать в интерактивном интерпретаторе. Например:20>21> ```python22> >>> from multiprocessing import Pool23> >>> p = Pool(5)24> >>> def f(x):25> ...     return x*x26> ...27> >>> p.map(f, [1,2,3])28> Process PoolWorker-1:29> Process PoolWorker-2:30> Traceback (most recent call last):31> AttributeError: 'module' object has no attribute 'f'32> AttributeError: 'module' object has no attribute 'f'33> AttributeError: 'module' object has no attribute 'f'34> ```3536### Класс [`Process`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Process)3738В `multiprocessing` процессы порождаются созданием объекта [`Process`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Process) и последующим вызовом его метода [`start()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Process.start). [`Process`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Process) следует API [`threading.Thread`](https://python-all.ru/3.0/library/threading.html#threading.Thread). Простейший пример многопроцессной программы:3940```python41 from multiprocessing import Process4243def f(name):44    print('hello', name)4546 if __name__ == '__main__':47     p = Process(target=f, args=('bob',))48     p.start()49     p.join()50```5152Чтобы показать идентификаторы задействованных процессов, вот расширенный пример:5354```python55from multiprocessing import Process56import os5758def info(title):59    print title60    print 'module name:', __name__61    print 'parent process:', os.getppid()62    print 'process id:', os.getpid()6364def f(name):65    info('function f')66    print 'hello', name6768if __name__ == '__main__':69    info('main line')70    p = Process(target=f, args=('bob',))71    p.start()72    p.join()73```7475Зачем (в Windows) нужна конструкция `if __name__ == '__main__'`, объясняется в разделе [*Рекомендации по программированию*](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing-programming).7677### Обмен объектами между процессами7879`multiprocessing` поддерживает два типа коммуникационных каналов между процессами:8081**Очереди**8283> Класс [`Queue`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Queue) является почти копией [`queue.Queue`](https://python-all.ru/3.0/library/queue.html#queue.Queue). Например:84>85> ```python86> from multiprocessing import Process, Queue87>88> def f(q):89>     q.put([42, None, 'hello'])90>91> if __name__ == '__main__':92>     q = Queue()93>     p = Process(target=f, args=(q,))94>     p.start()95>     print(q.get())    # prints "[42, None, 'hello']"96>     p.join()97> ```98>99> Очереди потокобезопасны и безопасны для процессов.100101**Каналы**102103> Функция [`Pipe()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Pipe) возвращает пару объектов соединения, связанных каналом, который по умолчанию является двунаправленным. Например:104>105> ```python106> from multiprocessing import Process, Pipe107>108> def f(conn):109>     conn.send([42, None, 'hello'])110>     conn.close()111>112> if __name__ == '__main__':113>     parent_conn, child_conn = Pipe()114>     p = Process(target=f, args=(child_conn,))115>     p.start()116>     print(parent_conn.recv())   # prints "[42, None, 'hello']"117>     p.join()118> ```119>120> Два объекта соединения, возвращаемые [`Pipe()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Pipe), представляют два конца канала. Каждый объект соединения имеет методы [`send()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Connection.send) и [`recv()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Connection.recv) (среди прочих). Следует учитывать, что данные в канале могут быть повреждены, если два процесса (или потока) попытаются одновременно читать или писать в *один и тот же* конец канала. Разумеется, нет риска повреждения, если процессы одновременно используют разные концы канала.121122### Синхронизация между процессами123124`multiprocessing` содержит аналоги всех примитивов синхронизации из [`threading`](https://python-all.ru/3.0/library/threading.html#module-threading). Например, можно использовать блокировку, чтобы гарантировать, что только один процесс выводит на стандартный вывод за раз:125126```python127from multiprocessing import Process, Lock128129def f(l, i):130    l.acquire()131    print('hello world', i)132    l.release()133134if __name__ == '__main__':135    lock = Lock()136137    for num in range(10):138        Process(target=f, args=(lock, num)).start()139```140141Без использования блокировки вывод от разных процессов может полностью перепутаться.142143### Совместное использование состояния между процессами144145Как уже упоминалось, при параллельном программировании обычно лучше избегать использования общего состояния, насколько это возможно. Это особенно верно при использовании нескольких процессов.146147Однако если всё же требуется использовать общие данные, то `multiprocessing` предоставляет несколько способов сделать это.148149**Разделяемая память**150151> Данные могут храниться в карте разделяемой памяти с помощью [`Value`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Value) или [`Array`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Array). Например, следующий код152>153> ```python154> from multiprocessing import Process, Value, Array155>156> def f(n, a):157>     n.value = 3.1415927158>     for i in range(len(a)):159>         a[i] = -a[i]160>161> if __name__ == '__main__':162>     num = Value('d', 0.0)163>     arr = Array('i', range(10))164>165>     p = Process(target=f, args=(num, arr))166>     p.start()167>     p.join()168>169>     print(num.value)170>     print(arr[:])171> ```172>173> выведет174>175> ```python176> 3.1415927177> [0, -1, -2, -3, -4, -5, -6, -7, -8, -9]178> ```179>180> Аргументы `'d'` и `'i'`, используемые при создании `num` и `arr`, являются кодами типов, используемыми модулем [`array`](https://python-all.ru/3.0/library/array.html#module-array): `'d'` обозначает число двойной точности с плавающей запятой, а `'i'` – целое число со знаком. Эти разделяемые объекты будут потокобезопасны и безопасны для процессов.181>182> Для большей гибкости при использовании общей памяти можно применить модуль `multiprocessing.sharedctypes`, который поддерживает создание произвольных объектов ctypes, размещённых в общей памяти.183184**Серверный процесс**185186> Объект менеджера, возвращаемый `Manager()`, управляет серверным процессом, который содержит объекты Python и позволяет другим процессам манипулировать ими через прокси.187>188> Менеджер, возвращаемый `Manager()`, будет поддерживать типы [`list`](https://python-all.ru/3.0/library/functions.html#list), [`dict`](https://python-all.ru/3.0/library/stdtypes.html#dict), `Namespace`, [`Lock`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Lock), [`RLock`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.RLock), [`Semaphore`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Semaphore), [`BoundedSemaphore`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.BoundedSemaphore), [`Condition`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Condition), [`Event`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Event), [`Queue`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Queue), [`Value`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Value) и [`Array`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Array). Например,189>190> ```python191> from multiprocessing import Process, Manager192>193> def f(d, l):194>     d[1] = '1'195>     d['2'] = 2196>     d[0.25] = None197>     l.reverse()198>199> if __name__ == '__main__':200>     manager = Manager()201>202>     d = manager.dict()203>     l = manager.list(range(10))204>205>     p = Process(target=f, args=(d, l))206>     p.start()207>     p.join()208>209>     print(d)210>     print(l)211> ```212>213> выведет214>215> ```python216> {0.25: None, 1: '1', '2': 2}217> [9, 8, 7, 6, 5, 4, 3, 2, 1, 0]218> ```219>220> Серверные менеджеры процессов более гибкие, чем использование объектов разделяемой памяти, потому что их можно настроить на поддержку произвольных типов объектов. Кроме того, один менеджер может быть разделён между процессами на разных компьютерах через сеть. Однако они медленнее, чем использование разделяемой памяти.221222### Использование пула рабочих процессов223224Класс `Pool` представляет пул рабочих процессов. Он имеет методы, которые позволяют делегировать задачи рабочим процессам несколькими различными способами.225226Например:227228```python229from multiprocessing import Pool230231def f(x):232    return x*x233234if __name__ == '__main__':235    pool = Pool(processes=4)              # start 4 worker processes236    result = pool.apply_async(f, [10])     # evaluate "f(10)" asynchronously237    print result.get(timeout=1)           # prints "100" unless your computer is *very* slow238    print pool.map(f, range(10))          # prints "[0, 1, 4,..., 81]"239```240241## Ссылка242243Пакет `multiprocessing` в основном повторяет API модуля [`threading`](https://python-all.ru/3.0/library/threading.html#module-threading).244245### [`Process`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Process) и исключения246247#### `[multiprocessing.Process]class multiprocessing.Process([group[, target[, name[, args[, kwargs]]]]])`248249Объекты Process представляют активность, которая выполняется в отдельном процессе. Класс [`Process`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Process) содержит эквиваленты всех методов [`threading.Thread`](https://python-all.ru/3.0/library/threading.html#threading.Thread).250251Конструктор всегда следует вызывать с именованными аргументами. *group* всегда должно быть `None`; он существует только для совместимости с [`threading.Thread`](https://python-all.ru/3.0/library/threading.html#threading.Thread). *target* – это вызываемый объект, который будет вызван методом [`run()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Process.run). По умолчанию он равен `None`, то есть ничего не вызывается. *name* – это имя процесса. По умолчанию генерируется уникальное имя вида 'Process-N1:N2:...:Nk', где N1,N2,...,Nk – последовательность целых чисел, длина которой определяется *поколением* процесса. *args* – кортеж аргументов для вызова цели. *kwargs* – словарь именованных аргументов для вызова цели. По умолчанию в *target* не передается никаких аргументов.252253Если подкласс переопределяет конструктор, он должен убедиться, что вызывает конструктор базового класса (`Process.__init__()`) перед любыми другими действиями над процессом.254255#### `[multiprocessing.Process.run]run()`256257Метод, представляющий активность процесса.258259Вы можете переопределить этот метод в подклассе. Стандартный метод [`run()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Process.run) вызывает вызываемый объект, переданный конструктору объекта в качестве аргумента target, если он есть, с позиционными и именованными аргументами, взятыми из аргументов *args* и *kwargs* соответственно.260261#### `[multiprocessing.Process.start]start()`262263Запускает активность процесса.264265Этот метод должен вызываться не более одного раза для каждого объекта процесса. Он обеспечивает вызов метода [`run()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Process.run) объекта в отдельном процессе.266267#### `[multiprocessing.Process.join]join([timeout])`268269Блокирует вызывающий поток до тех пор, пока процесс, для которого вызван метод [`join()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Process.join), не завершится или не истечет указанное время ожидания.270271Если *timeout* равен `None`, то время ожидания не ограничено.272273Процесс может быть присоединён много раз.274275Процесс не может присоединить сам себя, так как это приведёт к взаимоблокировке. Попытка вызвать join для процесса до его запуска является ошибкой.276277#### `[multiprocessing.Process.name]name`278279Имя процесса.280281Имя – это строка, используемая только для идентификации. Она не несет семантической нагрузки. Несколько процессов могут иметь одинаковое имя. Начальное имя задается конструктором.282283#### `[multiprocessing.Process.is_alive]is_alive()`284285Возвращает, жив ли процесс.286287Грубо говоря, объект процесса считается живым с момента возврата метода [`start()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Process.start) до завершения дочернего процесса.288289#### `[multiprocessing.Process.daemon]daemon`290291Флаг daemon процесса, логическое значение. Должен быть установлен до вызова [`start()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Process.start).292293Начальное значение наследуется от создающего процесса.294295Когда процесс завершается, он пытается завершить все свои дочерние процессы-демоны.296297Обратите внимание, что демоническому процессу не разрешается создавать дочерние процессы. В противном случае, если демонический процесс завершится при выходе из родительского процесса, его дочерние процессы останутся бесхозными.298299В дополнение к API `Threading.Thread`, объекты [`Process`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Process) также поддерживают следующие атрибуты и методы:300301#### `[multiprocessing.Process.pid]pid`302303Возвращает идентификатор процесса. До запуска процесса будет304305`None`306307.308309#### `[multiprocessing.Process.exitcode]exitcode`310311Код завершения дочернего процесса. Будет312313`None`314315, если процесс ещё не завершился. Отрицательное значение316317*-N*318319указывает, что дочерний процесс был завершён сигналом320321*N*322323.324325#### `[multiprocessing.Process.authkey]authkey`326327Ключ аутентификации процесса (байтовая строка).328329При инициализации `multiprocessing` главному процессу назначается случайная строка с помощью `os.random()`.330331При создании объекта [`Process`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Process) он наследует ключ аутентификации своего родительского процесса, хотя это можно изменить, установив [`authkey`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Process.authkey) в другую строку байтов.332333См. [*Ключи аутентификации*](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing-auth-keys).334335#### `[multiprocessing.Process.terminate]terminate()`336337Завершает процесс. В Unix это делается с помощью сигнала `SIGTERM`; в Windows используется `TerminateProcess`. Обратите внимание, что обработчики выхода и блоки finally и т.д. выполнены не будут.338339Обратите внимание, что процессы-потомки *не* будут завершены – они просто станут осиротевшими.340341> **Предупреждение**342>343> Если этот метод используется, когда связанный процесс использует канал или очередь, то канал или очередь могут быть повреждены и стать непригодными для использования другими процессами. Аналогично, если процесс захватил блокировку или семафор и т.п., то его завершение может привести к взаимной блокировке других процессов.344345Обратите внимание, что методы [`start()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Process.start), [`join()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Process.join), [`is_alive()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Process.is_alive) и `exit_code` должны вызываться только тем процессом, который создал объект процесса.346347Пример использования некоторых методов [`Process`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Process):348349```python350>>> import multiprocessing, time, signal351>>> p = multiprocessing.Process(target=time.sleep, args=(1000,))352>>> print(p, p.is_alive())353<Process(Process-1, initial)> False354>>> p.start()355>>> print(p, p.is_alive())356<Process(Process-1, started)> True357>>> p.terminate()358>>> print(p, p.is_alive())359<Process(Process-1, stopped[SIGTERM])> False360>>> p.exitcode == -signal.SIGTERM361True362```363364#### `[multiprocessing.BufferTooShort]exception multiprocessing.BufferTooShort`365366Исключение, вызываемое [`Connection.recv_bytes_into()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Connection.recv_bytes_into), когда переданный объект буфера слишком мал для читаемого сообщения.367368Если `e` является экземпляром [`BufferTooShort`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.BufferTooShort), то `e.args[0]` вернёт сообщение в виде строки байтов.369370### Каналы и очереди371372При использовании нескольких процессов обычно применяется передача сообщений для взаимодействия между процессами, что позволяет избежать использования примитивов синхронизации, таких как блокировки.373374Для передачи сообщений можно использовать [`Pipe()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Pipe) (для соединения между двумя процессами) или очередь (которая допускает множество производителей и потребителей).375376Типы [`Queue`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Queue) и [`JoinableQueue`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.JoinableQueue) – это очереди FIFO с множеством производителей и множеством потребителей, построенные по образцу класса [`queue.Queue`](https://python-all.ru/3.0/library/queue.html#queue.Queue) из стандартной библиотеки. Они различаются тем, что [`Queue`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Queue) не содержит методов [`task_done()`](https://python-all.ru/3.0/library/queue.html#queue.Queue.task_done) и [`join()`](https://python-all.ru/3.0/library/queue.html#queue.Queue.join), добавленных в класс [`queue.Queue`](https://python-all.ru/3.0/library/queue.html#queue.Queue) в Python 2.5.377378Если используется [`JoinableQueue`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.JoinableQueue), то **необходимо** вызывать [`JoinableQueue.task_done()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.JoinableQueue.task_done) для каждой задачи, извлечённой из очереди; иначе семафор, используемый для подсчёта незавершённых задач, может в конечном итоге переполниться, вызвав исключение.379380Обратите внимание, что общую очередь также можно создать с помощью объекта-менеджера – см. [*Менеджеры*](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing-managers).381382> **Примечание**383>384> `multiprocessing` использует обычные исключения [`queue.Empty`](https://python-all.ru/3.0/library/queue.html#queue.Empty) и [`queue.Full`](https://python-all.ru/3.0/library/queue.html#queue.Full) для сигнализации о тайм-ауте. Они недоступны в пространстве имён `multiprocessing`, поэтому их необходимо импортировать из [`queue`](https://python-all.ru/3.0/library/queue.html#module-queue).385386> **Предупреждение**387>388> Если процесс завершается с помощью [`Process.terminate()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Process.terminate) или [`os.kill()`](https://python-all.ru/3.0/library/os.html#os.kill) в то время, как он пытается использовать [`Queue`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Queue), то данные в очереди, скорее всего, окажутся повреждены. Это может привести к тому, что любой другой процесс получит исключение при последующей попытке использовать эту очередь.389390> **Предупреждение**391>392> Как упоминалось выше, если дочерний процесс поместил элементы в очередь (и не вызывал `JoinableQueue.cancel_join_thread()`), то этот процесс не завершится, пока все буферизованные элементы не будут сброшены в канал.393>394> Это означает, что при попытке присоединиться к этому процессу может возникнуть взаимоблокировка, если не быть уверенным, что все помещённые в очередь элементы были потреблены. Аналогично, если дочерний процесс не является демоническим, родительский процесс может зависнуть при завершении, когда попытается присоединиться ко всем своим недемоническим потомкам.395>396> Обратите внимание, что очередь, созданная с помощью менеджера, не имеет этой проблемы. См. [*Рекомендации по программированию*](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing-programming).397398Пример использования очередей для межпроцессного взаимодействия см. в разделе [*Примеры*](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing-examples).399400#### `[multiprocessing.Pipe]multiprocessing.Pipe([duplex])`401402Возвращает пару `(conn1, conn2)` объектов [`Connection`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Connection), представляющих концы канала.403404Если *duplex* равен `True` (по умолчанию), то канал двунаправленный. Если *duplex* равен `False`, то канал однонаправленный: `conn1` может использоваться только для получения сообщений, а `conn2` – только для отправки сообщений.405406#### `[multiprocessing.Queue]class multiprocessing.Queue([maxsize])`407408Возвращает общедоступную очередь для процессов, реализованную с помощью канала и нескольких блокировок/семафоров. Когда процесс впервые помещает элемент в очередь, запускается поток-загрузчик, который передаёт объекты из буфера в канал.409410Обычные исключения [`queue.Empty`](https://python-all.ru/3.0/library/queue.html#queue.Empty) и [`queue.Full`](https://python-all.ru/3.0/library/queue.html#queue.Full) из модуля `Queue` стандартной библиотеки возбуждаются для указания тайм-аутов.411412[`Queue`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Queue) реализует все методы [`queue.Queue`](https://python-all.ru/3.0/library/queue.html#queue.Queue), за исключением [`task_done()`](https://python-all.ru/3.0/library/queue.html#queue.Queue.task_done) и [`join()`](https://python-all.ru/3.0/library/queue.html#queue.Queue.join).413414#### `[multiprocessing.Queue.qsize]qsize()`415416Возвращает приблизительный размер очереди. Из-за семантики многопоточности/многопроцессности это число не является надёжным.417418Обратите внимание, что это может вызвать [`NotImplementedError`](https://python-all.ru/3.0/library/exceptions.html#exceptions.NotImplementedError) на платформах Unix, таких как Mac OS X, где `sem_getvalue()` не реализован.419420#### `[multiprocessing.Queue.empty]empty()`421422Возвращает423424`True`425426, если очередь пуста, и427428`False`429430в противном случае. Из-за семантики многопоточности/многопроцессорности это ненадёжно.431432#### `[multiprocessing.Queue.full]full()`433434Возвращает435436`True`437438, если очередь полна, и439440`False`441442в противном случае. Из-за семантики многопоточности/многопроцессорности это ненадёжно.443444#### `[multiprocessing.Queue.put]put(item[, block[, timeout]])`445446Помещает элемент в очередь. Если необязательный аргумент447448*block*449450равен451452`True`453454(по умолчанию), а455456*timeout*457458равен459460`None`461462(по умолчанию), то блокируется при необходимости до появления свободного места. Если463464*timeout*465466– положительное число, то блокируется не более чем на467468*timeout*469470секунд и возбуждает исключение471472[`queue.Full`](https://python-all.ru/3.0/library/queue.html#queue.Full)473474, если свободное место не появилось за это время. В противном случае (475476*block*477478равен479480`False`481482), помещает элемент в очередь, если свободное место доступно немедленно, иначе возбуждает исключение483484[`queue.Full`](https://python-all.ru/3.0/library/queue.html#queue.Full)485486(в этом случае487488*timeout*489490игнорируется).491492#### `[multiprocessing.Queue.put_nowait]put_nowait(item)`493494Эквивалентно495496`put(item, False)`497498.499500#### `[multiprocessing.Queue.get]get([block[, timeout]])`501502Remove and return an item from the queue. If optional args503504*block*505506is507508`True`509510(the default) and511512*timeout*513514is515516`None`517518(the default), block if necessary until an item is available. If519520*timeout*521522is a positive number, it blocks at most523524*timeout*525526seconds and raises the527528[`queue.Empty`](https://python-all.ru/3.0/library/queue.html#queue.Empty)529530exception if no item was available within that time. Otherwise (block is531532`False`533534), return an item if one is immediately available, else raise the535536[`queue.Empty`](https://python-all.ru/3.0/library/queue.html#queue.Empty)537538exception (539540*timeout*541542is ignored in that case).543544#### `[multiprocessing.Queue.get_nowait]get_nowait()`545546#### `get_no_wait()`547548Эквивалентно549550`get(False)`551552.553554У [`multiprocessing.Queue`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Queue) есть несколько дополнительных методов, отсутствующих в [`queue.Queue`](https://python-all.ru/3.0/library/queue.html#queue.Queue). В большинстве кода эти методы обычно не нужны:555556#### `[multiprocessing.Queue.close]close()`557558Указывает, что текущий процесс больше не будет помещать данные в эту очередь. Фоновый поток завершится после того, как сбросит все буферизованные данные в канал. Этот метод вызывается автоматически при сборке мусора очереди.559560#### `[multiprocessing.Queue.join_thread]join_thread()`561562Присоединяет фоновый поток. Используется только после вызова [`close()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Queue.close). Блокирует выполнение до завершения фонового потока, гарантируя, что все данные из буфера будут сброшены в канал.563564По умолчанию, если процесс не является создателем очереди, то при завершении он попытается присоединить фоновый поток очереди. Процесс может вызвать [`cancel_join_thread()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Queue.cancel_join_thread), чтобы [`join_thread()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Queue.join_thread) не делала ничего.565566#### `[multiprocessing.Queue.cancel_join_thread]cancel_join_thread()`567568Предотвращает блокирование569570[`join_thread()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Queue.join_thread)571572. В частности, это не даёт фоновому потоку автоматически присоединиться при завершении процесса – см.573574[`join_thread()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Queue.join_thread)575576.577578#### `[multiprocessing.JoinableQueue]class multiprocessing.JoinableQueue([maxsize])`579580[`JoinableQueue`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.JoinableQueue), подкласс [`очереди`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Queue), – это очередь, которая дополнительно содержит методы [`task_done()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.JoinableQueue.task_done) и [`join()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.JoinableQueue.join).581582#### `[multiprocessing.JoinableQueue.task_done]task_done()`583584Указывает, что ранее поставленная в очередь задача выполнена. Используется потоками-потребителями очереди. Для каждого вызова [`get()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Queue.get), используемого для получения задачи, последующий вызов [`task_done()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.JoinableQueue.task_done) сообщает очереди, что обработка задачи завершена.585586Если `join()` в данный момент блокируется, он возобновит работу, когда все элементы будут обработаны (т.е. когда для каждого элемента, помещенного в очередь с помощью [`put()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Queue.put), будет получен вызов [`task_done()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.JoinableQueue.task_done)).587588Вызывает [`ValueError`](https://python-all.ru/3.0/library/exceptions.html#exceptions.ValueError), если вызывается больше раз, чем было помещено элементов в очередь.589590#### `[multiprocessing.JoinableQueue.join]join()`591592Блокируется до тех пор, пока все элементы в очереди не будут извлечены и обработаны.593594Счетчик незавершенных задач увеличивается при каждом добавлении элемента в очередь. Он уменьшается, когда поток-потребитель вызывает [`task_done()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.JoinableQueue.task_done), чтобы указать, что элемент получен и вся работа с ним завершена. Когда счетчик незавершенных задач падает до нуля, `join()` разблокируется.595596### Разное597598#### `[multiprocessing.active_children]multiprocessing.active_children()`599600Возвращает список всех активных дочерних процессов текущего процесса.601602Вызов этой функции приводит к «присоединению» любых процессов, которые уже завершились.603604#### `[multiprocessing.cpu_count]multiprocessing.cpu_count()`605606Возвращает количество процессоров в системе. Может вызвать607608[`NotImplementedError`](https://python-all.ru/3.0/library/exceptions.html#exceptions.NotImplementedError)609610.611612#### `[multiprocessing.current_process]multiprocessing.current_process()`613614Возвращает объект [`процесса`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Process), соответствующий текущему процессу.615616Аналог [`threading.current_thread()`](https://python-all.ru/3.0/library/threading.html#threading.current_thread).617618#### `[multiprocessing.freeze_support]multiprocessing.freeze_support()`619620Добавляет поддержку для случая, когда программа, использующая `multiprocessing`, была заморожена для создания исполняемого файла Windows. (Протестировано с **py2exe**, **PyInstaller** и **cx\_Freeze**.)621622Эту функцию нужно вызывать сразу после строки `if __name__ == '__main__'` главного модуля. Например:623624```python625from multiprocessing import Process, freeze_support626627def f():628    print('hello world!')629630if __name__ == '__main__':631    freeze_support()632    Process(target=f).start()633```634635Если строка `freeze_support()` пропущена, то попытка запустить замороженный исполняемый файл вызовет [`RuntimeError`](https://python-all.ru/3.0/library/exceptions.html#exceptions.RuntimeError).636637Если модуль запускается обычным образом интерпретатором Python, то [`freeze_support()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.freeze_support) не действует.638639#### `[multiprocessing.set_executable]multiprocessing.set_executable()`640641Устанавливает путь к интерпретатору Python, который будет использоваться при запуске дочернего процесса. (По умолчанию используется [`sys.executable`](https://python-all.ru/3.0/library/sys.html#sys.executable)). Встраивающим, вероятно, потребуется сделать что-то вроде642643```python644  setExecutable(os.path.join(sys.exec_prefix, 'pythonw.exe'))645646before they can create child processes.  (Windows only)647```648649> **Примечание**650>651> `multiprocessing` не содержит аналогов [`threading.active_count()`](https://python-all.ru/3.0/library/threading.html#threading.active_count), [`threading.enumerate()`](https://python-all.ru/3.0/library/threading.html#threading.enumerate), [`threading.settrace()`](https://python-all.ru/3.0/library/threading.html#threading.settrace), [`threading.setprofile()`](https://python-all.ru/3.0/library/threading.html#threading.setprofile), [`threading.Timer`](https://python-all.ru/3.0/library/threading.html#threading.Timer) или [`threading.local`](https://python-all.ru/3.0/library/threading.html#threading.local).652653### Объекты подключения654655Объекты подключения позволяют отправлять и получать сериализуемые (picklable) объекты или строки. Их можно рассматривать как ориентированные на сообщения соединённые сокеты.656657Объекты соединения обычно создаются с помощью [`Pipe()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Pipe) – см. также [*Listeners and Clients*](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing-listeners-clients).658659#### `[multiprocessing.Connection]class multiprocessing.Connection`660661#### `[multiprocessing.Connection.send]send(obj)`662663Отправляет объект на другой конец соединения; этот объект должен быть прочитан с помощью [`recv()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Connection.recv).664665Объект должен быть сериализуемым (picklable).666667#### `[multiprocessing.Connection.recv]recv()`668669Возвращает объект, отправленный с другого конца соединения с помощью670671[`send()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Connection.send)672673. Возбуждает674675[`EOFError`](https://python-all.ru/3.0/library/exceptions.html#exceptions.EOFError)676677, если больше нечего получать и другой конец был закрыт.678679#### `[multiprocessing.Connection.fileno]fileno()`680681Возвращает файловый дескриптор или дескриптор, используемый соединением.682683#### `[multiprocessing.Connection.close]close()`684685Закрывает подключение.686687Вызывается автоматически при сборке мусора для подключения.688689#### `[multiprocessing.Connection.poll]poll([timeout])`690691Возвращает, есть ли данные для чтения.692693Если *timeout* не указан, то возврат происходит немедленно. Если *timeout* – число, оно задаёт максимальное время блокировки в секундах. Если *timeout* равно `None`, используется бесконечное ожидание.694695#### `[multiprocessing.Connection.send_bytes]send_bytes(buffer[, offset[, size]])`696697Отправляет байтовые данные из объекта, поддерживающего интерфейс буфера, в виде целого сообщения.698699Если задан *offset*, то данные читаются с этой позиции в *buffer*. Если задан *size*, то из буфера будет прочитано соответствующее количество байт.700701#### `[multiprocessing.Connection.recv_bytes]recv_bytes([maxlength])`702703Возвращает полное сообщение с байтовыми данными, отправленное с другого конца соединения, в виде строки. Возбуждает [`EOFError`](https://python-all.ru/3.0/library/exceptions.html#exceptions.EOFError), если больше нечего получать и другой конец был закрыт.704705Если задан *maxlength* и сообщение длиннее *maxlength*, то генерируется [`IOError`](https://python-all.ru/3.0/library/exceptions.html#exceptions.IOError), и соединение становится непригодным для чтения.706707#### `[multiprocessing.Connection.recv_bytes_into]recv_bytes_into(buffer[, offset])`708709Считывает в *buffer* полное сообщение с байтовыми данными, отправленное с другого конца соединения, и возвращает количество байт в сообщении. Возбуждает [`EOFError`](https://python-all.ru/3.0/library/exceptions.html#exceptions.EOFError), если больше нечего получать и другой конец был закрыт.710711*buffer* must be an object satisfying the writable buffer interface. If *offset* is given then the message will be written into the buffer from *that position. Offset must be a non-negative integer less than the \*length of \*buffer* (in bytes).712713Если буфер слишком мал, то вызывается исключение [`BufferTooShort`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.BufferTooShort), и полное сообщение доступно как `e.args[0]`, где `e` – это экземпляр исключения.714715Например:716717```python718>>> from multiprocessing import Pipe719>>> a, b = Pipe()720>>> a.send([1, 'hello', None])721>>> b.recv()722[1, 'hello', None]723>>> b.send_bytes('thank you')724>>> a.recv_bytes()725'thank you'726>>> import array727>>> arr1 = array.array('i', range(5))728>>> arr2 = array.array('i', [0] * 10)729>>> a.send_bytes(arr1)730>>> count = b.recv_bytes_into(arr2)731>>> assert count == len(arr1) * arr1.itemsize732>>> arr2733array('i', [0, 1, 2, 3, 4, 0, 0, 0, 0, 0])734```735736> **Предупреждение**737>738> Метод [`Connection.recv()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Connection.recv) автоматически десериализует полученные данные, что может представлять угрозу безопасности, если нет доверия к процессу, отправившему сообщение.739>740> Поэтому, если объект соединения не был создан с помощью [`Pipe()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Pipe), методы [`recv()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Connection.recv) и [`send()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Connection.send) следует использовать только после выполнения некоторой аутентификации. См. [*Authentication keys*](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing-auth-keys).741742> **Предупреждение**743>744> Если процесс завершается принудительно во время чтения или записи в канал, данные в канале, скорее всего, повредятся, так как может стать невозможно определить границы сообщений.745746### Примитивы синхронизации747748Обычно примитивы синхронизации не так необходимы в многопроцессной программе, как в многопоточной. См. документацию модуля [`threading`](https://python-all.ru/3.0/library/threading.html#module-threading).749750Обратите внимание, что примитивы синхронизации можно также создавать с помощью объекта менеджера – см. [*Менеджеры*](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing-managers).751752#### `[multiprocessing.BoundedSemaphore]class multiprocessing.BoundedSemaphore([value])`753754Объект ограниченного семафора: копия [`threading.BoundedSemaphore`](https://python-all.ru/3.0/library/threading.html#threading.BoundedSemaphore).755756(На Mac OS X это неотличимо от [`Semaphore`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Semaphore), поскольку `sem_getvalue()` не реализована на этой платформе).757758#### `[multiprocessing.Condition]class multiprocessing.Condition([lock])`759760Условная переменная: клон [`threading.Condition`](https://python-all.ru/3.0/library/threading.html#threading.Condition).761762If *lock* is specified then it should be a [`Lock`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Lock) or [`RLock`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.RLock) object from `multiprocessing`.763764#### `[multiprocessing.Event]class multiprocessing.Event`765766Клон767768[`threading.Event`](https://python-all.ru/3.0/library/threading.html#threading.Event)769770.771772#### `[multiprocessing.Lock]class multiprocessing.Lock`773774Объект нерекурсивной блокировки: копия775776[`threading.Lock`](https://python-all.ru/3.0/library/threading.html#threading.Lock)777778.779780#### `[multiprocessing.RLock]class multiprocessing.RLock`781782Объект рекурсивной блокировки: копия783784[`threading.RLock`](https://python-all.ru/3.0/library/threading.html#threading.RLock)785786.787788#### `[multiprocessing.Semaphore]class multiprocessing.Semaphore([value])`789790Объект ограниченного семафора: клон791792[`threading.Semaphore`](https://python-all.ru/3.0/library/threading.html#threading.Semaphore)793794.795796> **Примечание**797>798> Метод `acquire()` объектов [`BoundedSemaphore`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.BoundedSemaphore), [`Lock`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Lock), [`RLock`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.RLock) и [`Semaphore`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Semaphore) имеет параметр timeout, который не поддерживается аналогичными объектами в модуле [`threading`](https://python-all.ru/3.0/library/threading.html#module-threading). Сигнатура: `acquire(block=True, timeout=None)`, при этом допускается передача именованных аргументов. Если *block* равен `True`, а *timeout* не равен `None`, то параметр задает тайм-аут в секундах. Если *block* равен `False`, то *timeout* игнорируется.799>800> Обратите внимание, что на OS/X `sem_timedwait` не поддерживается, поэтому аргументы тайм-аута для них будут проигнорированы.801802> **Примечание**803>804> Если сигнал SIGINT, сгенерированный Ctrl-C, поступает в то время, как главный поток заблокирован вызовом `BoundedSemaphore.acquire()`, `Lock.acquire()`, `RLock.acquire()`, `Semaphore.acquire()`, `Condition.acquire()` или `Condition.wait()`, то вызов будет немедленно прерван и будет возбуждено [`KeyboardInterrupt`](https://python-all.ru/3.0/library/exceptions.html#exceptions.KeyboardInterrupt).805>806> Это отличается от поведения [`threading`](https://python-all.ru/3.0/library/threading.html#module-threading), где SIGINT будет игнорироваться, пока выполняются эквивалентные блокирующие вызовы.807808### Общие [`ctypes`](https://python-all.ru/3.0/library/ctypes.html#module-ctypes) объекты809810Можно создавать общие объекты с помощью разделяемой памяти, которые могут наследоваться дочерними процессами.811812#### `[multiprocessing.Value]multiprocessing.Value(typecode_or_type, *args[, lock])`813814Возвращает объект [`ctypes`](https://python-all.ru/3.0/library/ctypes.html#module-ctypes), выделенный из разделяемой памяти. По умолчанию возвращаемое значение на самом деле является синхронизированной обёрткой для этого объекта.815816*typecode\_or\_type* определяет тип возвращаемого объекта: это либо тип ctypes, либо однобуквенный код типа, как в модуле [`array`](https://python-all.ru/3.0/library/array.html#module-array). *\*args* передаётся конструктору типа.817818Если *блокировка* равно `True` (по умолчанию), создаётся новый объект блокировки для синхронизации доступа к значению. Если *блокировка* является объектом [`Lock`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Lock) или [`RLock`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.RLock), то он будет использоваться для синхронизации доступа к значению. Если *блокировка* равно `False`, доступ к возвращаемому объекту не будет автоматически защищён блокировкой, поэтому он не обязательно будет «процесс-безопасным».819820Обратите внимание, что *блокировка* – это аргумент, передаваемый только по ключевому слову.821822#### `[multiprocessing.Array]multiprocessing.Array(typecode_or_type, size_or_initializer, *, lock=True)`823824Возвращает массив ctypes, выделенный из разделяемой памяти. По умолчанию возвращаемое значение на самом деле является синхронизированной обёрткой для массива.825826*typecode\_or\_type* определяет тип элементов возвращаемого массива: это либо тип ctypes, либо однобуквенный код типа, используемый модулем [`array`](https://python-all.ru/3.0/library/array.html#module-array). Если *size\_or\_initializer* является целым числом, то оно определяет длину массива, и массив будет изначально заполнен нулями. В противном случае *size\_or\_initializer* является последовательностью, которая используется для инициализации массива, и её длина определяет длину массива.827828Если *блокировка* равно `True` (по умолчанию), создаётся новый объект блокировки для синхронизации доступа к значению. Если *блокировка* является объектом [`Lock`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Lock) или [`RLock`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.RLock), то он будет использоваться для синхронизации доступа к значению. Если *блокировка* равно `False`, доступ к возвращаемому объекту не будет автоматически защищён блокировкой, поэтому он не обязательно будет «процесс-безопасным».829830Обратите внимание, что *блокировка* – это аргумент, передаваемый только по ключевому слову.831832Обратите внимание, что массив [`ctypes.c_char`](https://python-all.ru/3.0/library/ctypes.html#ctypes.c_char) имеет атрибуты *value* и *raw*, которые позволяют использовать его для хранения и извлечения строк.833834#### `multiprocessing.sharedctypes` модуль835836Модуль `multiprocessing.sharedctypes` предоставляет функции для выделения объектов [`ctypes`](https://python-all.ru/3.0/library/ctypes.html#module-ctypes) из разделяемой памяти, которые могут быть унаследованы дочерними процессами.837838> **Примечание**839>840> Хотя можно сохранить указатель в разделяемой памяти, следует помнить, что он будет ссылаться на местоположение в адресном пространстве конкретного процесса. Однако указатель, скорее всего, будет недействительным в контексте второго процесса, и попытка разыменования указателя из второго процесса может привести к аварийному завершению.841842#### `[multiprocessing.sharedctypes.RawArray]multiprocessing.sharedctypes.RawArray(typecode_or_type, size_or_initializer)`843844Возвращает массив ctypes, выделенный из разделяемой памяти.845846*typecode\_or\_type* определяет тип элементов возвращаемого массива: это либо тип ctypes, либо однобуквенный код типа, как в модуле [`array`](https://python-all.ru/3.0/library/array.html#module-array). Если *size\_or\_initializer* является целым числом, то оно задаёт длину массива, и массив будет заполнен нулями. В противном случае *size\_or\_initializer* – это последовательность, которая используется для инициализации массива, и её длина определяет длину массива.847848Обратите внимание, что запись и чтение элемента потенциально неатомарны – используйте вместо этого [`Array()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.sharedctypes.Array), чтобы гарантировать автоматическую синхронизацию доступа с помощью блокировки.849850#### `[multiprocessing.sharedctypes.RawValue]multiprocessing.sharedctypes.RawValue(typecode_or_type, *args)`851852Возвращает объект ctypes, выделенный из разделяемой памяти.853854*typecode\_or\_type* определяет тип возвращаемого объекта: это либо тип ctypes, либо однобуквенный код типа, как в модуле [`array`](https://python-all.ru/3.0/library/array.html#module-array). *\*args* передаётся конструктору типа.855856Обратите внимание, что запись и чтение значения потенциально неатомарны – используйте вместо этого [`Value()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.sharedctypes.Value), чтобы гарантировать автоматическую синхронизацию доступа с помощью блокировки.857858Обратите внимание, что массив [`ctypes.c_char`](https://python-all.ru/3.0/library/ctypes.html#ctypes.c_char) имеет атрибуты `value` и `raw`, которые позволяют хранить и извлекать строки – см. документацию по [`ctypes`](https://python-all.ru/3.0/library/ctypes.html#module-ctypes).859860#### `[multiprocessing.sharedctypes.Array]multiprocessing.sharedctypes.Array(typecode_or_type, size_or_initializer, *args[, lock])`861862То же, что и [`RawArray()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.sharedctypes.RawArray), но в зависимости от значения *блокировка* может быть возвращена процесс-безопасная синхронизирующая обёртка вместо необработанного массива ctypes.863864Если *блокировка* равна `True` (по умолчанию), то создаётся новый объект блокировки для синхронизации доступа к значению. Если *блокировка* является объектом `Lock` или `RLock`, то он будет использоваться для синхронизации доступа к значению. Если *блокировка* равна `False`, то доступ к возвращаемому объекту не будет автоматически защищён блокировкой, поэтому он не обязательно будет «процесс-безопасным».865866Обратите внимание, что *блокировка* – это аргумент, передаваемый только по ключевому слову.867868#### `[multiprocessing.sharedctypes.Value]multiprocessing.sharedctypes.Value(typecode_or_type, *args[, lock])`869870То же, что и [`RawValue()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.sharedctypes.RawValue), но в зависимости от значения *блокировка* может быть возвращена процесс-безопасная синхронизирующая обёртка вместо необработанного объекта ctypes.871872Если *блокировка* равна `True` (по умолчанию), то создаётся новый объект блокировки для синхронизации доступа к значению. Если *блокировка* является объектом `Lock` или `RLock`, то он будет использоваться для синхронизации доступа к значению. Если *блокировка* равна `False`, то доступ к возвращаемому объекту не будет автоматически защищён блокировкой, поэтому он не обязательно будет «процесс-безопасным».873874Обратите внимание, что *блокировка* – это аргумент, передаваемый только по ключевому слову.875876#### `[multiprocessing.sharedctypes.copy]multiprocessing.sharedctypes.copy(obj)`877878Возвращает объект ctypes, выделенный в общей памяти и являющийся копией объекта ctypes879880*obj*881882.883884#### `[multiprocessing.sharedctypes.synchronized]multiprocessing.sharedctypes.synchronized(obj[, lock])`885886Возвращает процесс-безопасный объект-обёртку для объекта ctypes, который использует *блокировка* для синхронизации доступа. Если *блокировка* равно `None` (по умолчанию), то автоматически создаётся объект [`multiprocessing.RLock`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.RLock).887888Синхронизированная обёртка будет иметь два дополнительных метода по сравнению с объектом, который она оборачивает: `get_obj()` возвращает обёрнутый объект, а `get_lock()` возвращает объект блокировки, используемый для синхронизации.889890Обратите внимание, что доступ к объекту ctypes через обёртку может быть значительно медленнее, чем доступ к исходному объекту ctypes.891892В таблице ниже сравнивается синтаксис создания разделяемых объектов ctypes из общей памяти с обычным синтаксисом ctypes. (В таблице `MyStruct` – это некий подкласс [`ctypes.Structure`](https://python-all.ru/3.0/library/ctypes.html#ctypes.Structure).)893894| ctypes | sharedctypes с типом | sharedctypes с typecode |895| --- | --- | --- |896| c\_double(2.4) | RawValue(c\_double, 2.4) | RawValue(‘d’, 2.4) |897| MyStruct(4, 6) | RawValue(MyStruct, 4, 6) |  |898| (c\_short \* 7)() | RawArray(c\_short, 7) | RawArray(‘h’, 7) |899| (c\_int \* 3)(9, 2, 8) | RawArray(c\_int, (9, 2, 8)) | RawArray(‘i’, (9, 2, 8)) |900901Ниже приведён пример, в котором дочерний процесс изменяет несколько объектов ctypes:902903```python904from multiprocessing import Process, Lock905from multiprocessing.sharedctypes import Value, Array906from ctypes import Structure, c_double907908class Point(Structure):909    _fields_ = [('x', c_double), ('y', c_double)]910911def modify(n, x, s, A):912    n.value **= 2913    x.value **= 2914    s.value = s.value.upper()915    for a in A:916        a.x **= 2917        a.y **= 2918919if __name__ == '__main__':920    lock = Lock()921922    n = Value('i', 7)923    x = Value(ctypes.c_double, 1.0/3.0, lock=False)924    s = Array('c', 'hello world', lock=lock)925    A = Array(Point, [(1.875,-6.25), (-5.75,2.0), (2.375,9.5)], lock=lock)926927    p = Process(target=modify, args=(n, x, s, A))928    p.start()929    p.join()930931    print(n.value)932    print(x.value)933    print(s.value)934    print([(a.x, a.y) for a in A])935```936937Выводятся следующие результаты:938939```text940499410.1111111111111111942HELLO WORLD943[(3.515625, 39.0625), (33.0625, 4.0), (5.640625, 90.25)]944```945946### Менеджеры947948Менеджеры позволяют создавать данные, которыми могут обмениваться разные процессы. Объект менеджера управляет серверным процессом, который ведает *разделяемыми объектами*. Другие процессы могут получать доступ к разделяемым объектам через прокси.949950#### `[multiprocessing.sharedctypes.multiprocessing.Manager]multiprocessing.Manager()`951952Возвращает запущенный объект953954[`SyncManager`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.managers.SyncManager)955956, который можно использовать для обмена объектами между процессами. Возвращаемый объект менеджера соответствует порождённому дочернему процессу и имеет методы, которые создают разделяемые объекты и возвращают соответствующие прокси.957958Процессы менеджера завершаются сразу после сборки мусора или завершения родительского процесса. Классы менеджеров определены в модуле `multiprocessing.managers`:959960#### `[multiprocessing.managers.BaseManager]class multiprocessing.managers.BaseManager([address[, authkey]])`961962Создает объект BaseManager.963964После создания следует вызвать [`start()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.managers.BaseManager.start) или [`serve_forever()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.managers.BaseManager.serve_forever), чтобы убедиться, что объект менеджера ссылается на запущенный процесс менеджера.965966*address* – это адрес, по которому процесс менеджера ожидает новые подключения. Если *address* равен `None`, то выбирается произвольный.967968*authkey* – это ключ аутентификации, который используется для проверки подлинности входящих подключений к серверному процессу. Если *authkey* равен `None`, то используется `current_process().authkey`. В противном случае используется *authkey*, который должен быть строкой.969970#### `[multiprocessing.managers.BaseManager.start]start()`971972Запускает подпроцесс для запуска менеджера.973974#### `[multiprocessing.managers.BaseManager.serve_forever]serve_forever()`975976Запускает сервер в текущем процессе.977978#### `[multiprocessing.managers.BaseManager.from_address]from_address(address, authkey)`979980Метод класса, который создаёт объект менеджера, ссылающийся на уже существующий процесс сервера, использующий заданный адрес и ключ аутентификации.981982#### `[multiprocessing.managers.BaseManager.get_server]get_server()`983984Возвращает объект `Server`, который представляет фактический сервер под управлением Manager. Объект `Server` поддерживает метод [`serve_forever()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.managers.BaseManager.serve_forever):985986```python987>>> from multiprocessing.managers import BaseManager988>>> m = BaseManager(address=('', 50000), authkey='abc'))989>>> server = m.get_server()990>>> s.serve_forever()991```992993`Server` дополнительно имеет атрибут [`address`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.managers.BaseManager.address).994995#### `[multiprocessing.managers.BaseManager.connect]connect()`996997Подключает локальный объект менеджера к удаленному процессу менеджера:998999```python1000>>> from multiprocessing.managers import BaseManager1001>>> m = BaseManager(address='127.0.0.1', authkey='abc))>>> 1002m.connect()1003```10041005#### `[multiprocessing.managers.BaseManager.shutdown]shutdown()`10061007Останавливает процесс, используемый менеджером. Это доступно только в том случае, если [`start()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.managers.BaseManager.start) был использован для запуска серверного процесса.10081009Этот метод можно вызывать несколько раз.10101011#### `[multiprocessing.managers.BaseManager.register]register(typeid[, callable[, proxytype[, exposed[, method_to_typeid[, create_method]]]]])`10121013Метод класса, который можно использовать для регистрации типа или вызываемого объекта в классе менеджера.10141015*typeid* – это «идентификатор типа», который используется для идентификации определенного типа разделяемого объекта. Должен быть строкой.10161017*callable* – это вызываемый объект, используемый для создания объектов для данного идентификатора типа. Если экземпляр менеджера будет создаваться через метод класса [`from_address()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.managers.BaseManager.from_address) или если аргумент *create\_method* равен `False`, то это можно оставить как `None`.10181019*proxytype* – это подкласс [`BaseProxy`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.managers.BaseProxy), который используется для создания прокси для разделяемых объектов с данным *typeid*. Если `None`, то класс прокси создается автоматически.10201021*exposed* используется для указания последовательности имён методов, к которым прокси для этого typeid должны иметь доступ через `BaseProxy._callMethod()`. (Если *exposed* равен `None`, то используется `proxytype._exposed_`, если он существует.) В случае, когда список exposed не указан, все «открытые методы» разделяемого объекта будут доступны. (Здесь «открытый метод» означает любой атрибут, который имеет метод [`__call__()`](https://python-all.ru/3.0/reference/datamodel.html#object.__call__) и чьё имя не начинается с `'_'`.)10221023*method\_to\_typeid* – это отображение, используемое для указания возвращаемого типа тех открытых методов, которые должны возвращать прокси. Оно сопоставляет имена методов со строками typeid. (Если *method\_to\_typeid* равен `None`, то вместо него используется `proxytype._method_to_typeid_`, если он существует.) Если имя метода не является ключом в этом отображении или если отображение равно `None`, то объект, возвращаемый методом, будет скопирован по значению.10241025*create\_method* определяет, должен ли быть создан метод с именем *typeid*, который можно использовать, чтобы сообщить серверному процессу о создании нового разделяемого объекта и возврате прокси для него. По умолчанию это `True`.10261027Экземпляры [`BaseManager`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.managers.BaseManager) также имеют одно свойство только для чтения:10281029#### `[multiprocessing.managers.BaseManager.address]address`10301031Адрес, используемый менеджером.10321033#### `[multiprocessing.managers.SyncManager]class multiprocessing.managers.SyncManager`10341035Подкласс [`BaseManager`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.managers.BaseManager), который можно использовать для синхронизации процессов. Объекты этого типа возвращаются `multiprocessing.Manager()`.10361037Он также поддерживает создание разделяемых списков и словарей.10381039#### `[multiprocessing.managers.SyncManager.BoundedSemaphore]BoundedSemaphore([value])`10401041Создаёт разделяемый объект10421043[`threading.BoundedSemaphore`](https://python-all.ru/3.0/library/threading.html#threading.BoundedSemaphore)10441045и возвращает прокси для него.10461047#### `[multiprocessing.managers.SyncManager.Condition]Condition([lock])`10481049Создаёт разделяемый объект [`threading.Condition`](https://python-all.ru/3.0/library/threading.html#threading.Condition) и возвращает прокси для него.10501051Если *блокировка* указан, то он должен быть прокси для объекта [`threading.Lock`](https://python-all.ru/3.0/library/threading.html#threading.Lock) или [`threading.RLock`](https://python-all.ru/3.0/library/threading.html#threading.RLock).10521053#### `[multiprocessing.managers.SyncManager.Event]Event()`10541055Создаёт разделяемый объект10561057[`threading.Event`](https://python-all.ru/3.0/library/threading.html#threading.Event)10581059и возвращает для него прокси.10601061#### `[multiprocessing.managers.SyncManager.Lock]Lock()`10621063Создаёт разделяемый объект10641065[`threading.Lock`](https://python-all.ru/3.0/library/threading.html#threading.Lock)10661067и возвращает для него прокси.10681069#### `[multiprocessing.managers.SyncManager.Namespace]Namespace()`10701071Создаёт разделяемый объект10721073[`Namespace`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.managers.SyncManager.Namespace)10741075и возвращает для него прокси.10761077#### `[multiprocessing.managers.SyncManager.Queue]Queue([maxsize])`10781079Создаёт разделяемый объект10801081[`queue.Queue`](https://python-all.ru/3.0/library/queue.html#queue.Queue)10821083и возвращает для него прокси.10841085#### `[multiprocessing.managers.SyncManager.RLock]RLock()`10861087Создаёт разделяемый объект10881089[`threading.RLock`](https://python-all.ru/3.0/library/threading.html#threading.RLock)10901091и возвращает для него прокси.10921093#### `[multiprocessing.managers.SyncManager.Semaphore]Semaphore([value])`10941095Создаёт разделяемый объект10961097[`threading.Semaphore`](https://python-all.ru/3.0/library/threading.html#threading.Semaphore)10981099и возвращает прокси для него.11001101#### `[multiprocessing.managers.SyncManager.Array]Array(typecode, sequence)`11021103Создать массив и вернуть для него прокси.11041105#### `[multiprocessing.managers.SyncManager.Value]Value(typecode, value)`11061107Создаёт объект с атрибутом11081109`value`11101111, доступным для записи, и возвращает прокси для него.11121113#### `[multiprocessing.managers.SyncManager.dict]dict()`11141115#### `dict(mapping)`11161117#### `dict(sequence)`11181119Создаёт разделяемый объект11201121`dict`11221123и возвращает для него прокси.11241125#### `[multiprocessing.managers.SyncManager.list]list()`11261127#### `list(sequence)`11281129Создаёт разделяемый объект11301131`list`11321133и возвращает для него прокси.11341135#### объекты Namespace11361137Объект пространства имён не имеет открытых методов, но имеет изменяемые атрибуты. Его представление показывает значения его атрибутов.11381139Однако при использовании прокси для объекта пространства имён атрибут, начинающийся с `'_'`, будет атрибутом прокси, а не атрибутом референта:11401141```python1142>>> manager = multiprocessing.Manager()1143>>> Global = manager.Namespace()1144>>> Global.x = 101145>>> Global.y = 'hello'1146>>> Global._z = 12.3    # это атрибут прокси1147>>> print(Global)1148Namespace(x=10, y='hello')1149```11501151#### Настраиваемые менеджеры11521153Чтобы создать собственный менеджер, нужно создать подкласс [`BaseManager`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.managers.BaseManager) и использовать метод класса [`register()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.managers.BaseManager.register) для регистрации новых типов или вызываемых объектов в классе менеджера. Например:11541155```python1156from multiprocessing.managers import BaseManager11571158class MathsClass(object):1159    def add(self, x, y):1160        return x + y1161    def mul(self, x, y):1162        return x * y11631164class MyManager(BaseManager):1165    pass11661167MyManager.register('Maths', MathsClass)11681169if __name__ == '__main__':1170    manager = MyManager()1171    manager.start()1172    maths = manager.Maths()1173    print(maths.add(4, 3))         # выводит 71174    print(maths.mul(7, 8))         # выводит 561175```11761177#### Использование удалённого менеджера11781179Можно запустить сервер менеджера на одной машине и позволить клиентам использовать его с других машин (при условии, что соответствующие брандмауэры это разрешают).11801181Выполнение следующих команд создаёт сервер для одной общей очереди, к которой удалённые клиенты могут обращаться:11821183```python1184>>> from multiprocessing.managers import BaseManager1185>>> import queue1186>>> queue = queue.Queue()1187>>> class QueueManager(BaseManager): pass1188...1189>>> QueueManager.register('get_queue', callable=lambda:queue)1190>>> m = QueueManager(address=('', 50000), authkey='abracadabra')1191>>> s = m.get_server()1192>>> s.serveForever()1193```11941195Один клиент может получить доступ к серверу следующим образом:11961197```python1198>>> from multiprocessing.managers import BaseManager1199>>> class QueueManager(BaseManager): pass1200...1201>>> QueueManager.register('get_queue')1202>>> m = QueueManager(address=('foo.bar.org', 50000), authkey='abracadabra')1203>>> m.connect()1204>>> queue = m.get_queue()1205>>> queue.put('hello')1206```12071208Другой клиент также может использовать его:12091210```python1211>>> from multiprocessing.managers import BaseManager1212>>> class QueueManager(BaseManager): pass1213...1214>>> QueueManager.register('getQueue')1215>>> m = QueueManager.from_address(address=('foo.bar.org', 50000), authkey='abracadabra')1216>>> queue = m.getQueue()1217>>> queue.get()1218'hello'1219```12201221Локальные процессы также могут получить доступ к этой очереди, используя приведённый выше код на клиенте для удалённого доступа к ней:12221223```python1224>>> from multiprocessing import Process, Queue1225>>> from multiprocessing.managers import BaseManager1226>>> class Worker(Process):1227...     def __init__(self, q):1228...         self.q = q1229...         super(Worker, self).__init__()1230...     def run(self):1231...         self.q.put('local hello')1232...1233>>> queue = Queue()1234>>> w = Worker(queue)1235>>> w.start()1236>>> class QueueManager(BaseManager): pass1237...1238>>> QueueManager.register('get_queue', callable=lambda: queue)1239>>> m = QueueManager(address=('', 50000), authkey='abracadabra')1240>>> s = m.get_server()1241>>> s.serve_forever()1242```12431244### Объекты-прокси12451246Прокси – это объект, который *ссылается* на общий объект, находящийся (предположительно) в другом процессе. Общий объект называется *референтом* этого прокси. Несколько объектов-прокси могут иметь один и тот же референт.12471248Прокси-объект имеет методы, которые вызывают соответствующие методы его референта (хотя не каждый метод референта обязательно будет доступен через прокси). Прокси обычно можно использовать почти так же, как и его референт:12491250```python1251>>> from multiprocessing import Manager1252>>> manager = Manager()1253>>> l = manager.list([i*i for i in range(10)])1254>>> print(l)1255[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]1256>>> print(repr(l))1257<ListProxy object, typeid 'list' at 0xb799974c>1258>>> l[4]1259161260>>> l[2:5]1261[4, 9, 16]1262```12631264Обратите внимание, что применение [`str()`](https://python-all.ru/3.0/library/functions.html#str) к прокси вернёт представление референта, тогда как применение [`repr()`](https://python-all.ru/3.0/library/functions.html#repr) вернёт представление прокси.12651266Важная особенность прокси-объектов в том, что они могут быть сериализованы (picklable) и поэтому могут передаваться между процессами. Однако обратите внимание: если прокси отправляется в процесс соответствующего менеджера, то при десериализации будет получен сам референт. Это означает, например, что один общий объект может содержать другой:12671268```python1269>>> a = manager.list()1270>>> b = manager.list()1271>>> a.append(b)         # референт a теперь содержит референт b1272>>> print(a, b)1273[[]] []1274>>> b.append('hello')1275>>> print(a, b)1276[['hello']] ['hello']1277```12781279> **Примечание**1280>1281> Типы прокси в `multiprocessing` не поддерживают сравнение по значению. Так, например,1282>1283> ```python1284> manager.list([1,2,3]) == [1,2,3]1285> ```1286>1287> вернёт `False`. Вместо этого для сравнений следует использовать копию объекта, на который ссылается прокси.12881289#### `[multiprocessing.managers.BaseProxy]class multiprocessing.managers.BaseProxy`12901291Объекты прокси являются экземплярами подклассов [`BaseProxy`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.managers.BaseProxy).12921293#### `[multiprocessing.managers.BaseProxy._callmethod]_callmethod(methodname[, args[, kwds]])`12941295Вызывает метод референта прокси и возвращает результат.12961297Если `proxy` является прокси, чей референт – `obj`, то выражение12981299```python1300proxy._callmethod(methodname, args, kwds)1301```13021303вычислит выражение13041305```python1306getattr(obj, methodname)(*args, **kwds)1307```13081309в процессе менеджера.13101311Возвращаемое значение будет копией результата вызова или прокси на новый разделяемый объект – см. документацию по аргументу *method\_to\_typeid* метода [`BaseManager.register()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.managers.BaseManager.register).13121313Если вызов вызывает исключение, то оно повторно возбуждается методом [`_callmethod()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.managers.BaseProxy._callmethod). Если в процессе менеджера возникает другое исключение, оно преобразуется в исключение `RemoteError` и возбуждается методом [`_callmethod()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.managers.BaseProxy._callmethod).13141315Обратите внимание: если *methodname* не был *опубликован*, будет возбуждено исключение.13161317Пример использования [`_callmethod()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.managers.BaseProxy._callmethod):13181319```python1320>>> l = manager.list(range(10))1321>>> l._callmethod('__len__')1322101323>>> l._callmethod('__getslice__', (2, 7))   # эквивалентно `l[2:7]`1324[2, 3, 4, 5, 6]1325>>> l._callmethod('__getitem__', (20,))     # эквивалентно `l[20]`1326...1327IndexError: list index out of range1328```13291330#### `[multiprocessing.managers.BaseProxy._getvalue]_getvalue()`13311332Возвращает копию референта.13331334Если референт не может быть сериализован (unpicklable), то это вызовет исключение.13351336#### `[multiprocessing.managers.BaseProxy.__repr__]__repr__()`13371338Возвращает представление прокси-объекта.13391340#### `[multiprocessing.managers.BaseProxy.__str__]__str__()`13411342Возвращает представление референта.13431344#### Очистка13451346Прокси-объект использует колбэк слабой ссылки: когда он собирается сборщиком мусора, он отменяет свою регистрацию у менеджера, которому принадлежит его референт.13471348Разделяемый объект удаляется из процесса менеджера, когда на него больше не ссылается ни один прокси.13491350### Пулы процессов13511352Можно создать пул процессов, которые будут выполнять отправленные им задачи с помощью класса `Pool`.13531354#### `[multiprocessing.pool.multiprocessing.Pool]class multiprocessing.Pool([processes[, initializer[, initargs]]])`13551356Объект пула процессов, который управляет пулом рабочих процессов, которым можно отправлять задачи. Он поддерживает асинхронные результаты с тайм-аутами и колбэками, а также имеет параллельную реализацию map.13571358*processes* – количество используемых рабочих процессов. Если *processes* равно `None`, используется число, возвращаемое `cpu_count()`. Если *initializer* не `None`, каждый рабочий процесс при запуске вызовет `initializer(*initargs)`.13591360#### `[multiprocessing.pool.multiprocessing.Pool.apply]apply(func[, args[, kwds]])`13611362Вызывает13631364*func*13651366с аргументами13671368*args*13691370и именованными аргументами13711372*kwds*13731374. Блокируется до тех пор, пока результат не будет готов.13751376#### `[multiprocessing.pool.multiprocessing.Pool.apply_async]apply_async(func[, args[, kwds[, callback]]])`13771378Вариант метода [`apply()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.pool.multiprocessing.Pool.apply), который возвращает объект результата.13791380Если указан *колбэк*, это должен быть вызываемый объект, принимающий один аргумент. Когда результат готов, к нему применяется *колбэк* (если вызов не завершился ошибкой). *Колбэк* должен выполняться немедленно, иначе поток, обрабатывающий результаты, будет заблокирован.13811382#### `[multiprocessing.pool.multiprocessing.Pool.map]map(func, iterable[, chunksize])`13831384Параллельный аналог встроенной функции [`map()`](https://python-all.ru/3.0/library/functions.html#map), собирающий результат в список. Блокируется до тех пор, пока не будет готов весь результат.13851386Этот метод разбивает итерируемый объект на несколько частей (chunks), которые отправляет в пул процессов как отдельные задачи. Приблизительный размер этих частей можно задать, установив *chunksize* в положительное целое число.13871388#### `[multiprocessing.pool.multiprocessing.Pool.map_async]map_async(func, iterable[, chunksize[, callback]])`13891390Вариант метода [`map()`](https://python-all.ru/3.0/library/functions.html#map), который возвращает объект результата.13911392Если указан *колбэк*, это должен быть вызываемый объект, принимающий один аргумент. Когда результат готов, к нему применяется *колбэк* (если вызов не завершился ошибкой). *Колбэк* должен выполняться немедленно, иначе поток, обрабатывающий результаты, будет заблокирован.13931394#### `[multiprocessing.pool.multiprocessing.Pool.imap]imap(func, iterable[, chunksize])`13951396Более ленивая версия [`map()`](https://python-all.ru/3.0/library/functions.html#map).13971398Аргумент *chunksize* аналогичен тому, что используется в методе [`map()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.pool.multiprocessing.Pool.map). Для очень длинных итерируемых объектов использование большого значения *chunksize* может сделать выполнение задачи **значительно** быстрее, чем использование значения по умолчанию `1`.13991400Кроме того, если *chunksize* равен `1`, то метод [`next()`](https://python-all.ru/3.0/library/functions.html#next) итератора, возвращаемого методом [`imap()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.pool.multiprocessing.Pool.imap), имеет необязательный параметр *timeout*: `next(timeout)` вызовет `multiprocessing.TimeoutError`, если результат не может быть возвращён в течение *timeout* секунд.14011402#### `[multiprocessing.pool.multiprocessing.Pool.imap_unordered]imap_unordered(func, iterable[, chunksize])`14031404То же, что и14051406[`imap()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.pool.multiprocessing.Pool.imap)14071408, но порядок результатов из возвращаемого итератора может быть произвольным. (Только при наличии одного рабочего процесса порядок гарантированно «правильный».)14091410#### `[multiprocessing.pool.multiprocessing.Pool.close]close()`14111412Предотвращает отправку новых задач в пул. После завершения всех задач рабочие процессы завершаются.14131414#### `[multiprocessing.pool.multiprocessing.Pool.terminate]terminate()`14151416Немедленно останавливает рабочие процессы, не завершая незавершённую работу. Когда объект пула собирается сборщиком мусора,14171418[`terminate()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.pool.multiprocessing.Pool.terminate)14191420будет вызван немедленно.14211422#### `[multiprocessing.pool.multiprocessing.Pool.join]join()`14231424Ожидает завершения рабочих процессов. Перед вызовом14251426[`join()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.pool.multiprocessing.Pool.join)14271428необходимо вызвать14291430[`close()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.pool.multiprocessing.Pool.close)14311432или14331434[`terminate()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.pool.multiprocessing.Pool.terminate)14351436.14371438#### `[multiprocessing.pool.AsyncResult]class multiprocessing.pool.AsyncResult`14391440Класс результата, возвращаемого методами `Pool.apply_async()` и `Pool.map_async()`.14411442#### `[multiprocessing.pool.AsyncResult.get]get([timeout])`14431444Возвращает результат, когда он будет получен. Если14451446*timeout*14471448не равно14491450`None`14511452и результат не получен в течение14531454*timeout*14551456секунд, то возбуждается14571458`multiprocessing.TimeoutError`14591460. Если удалённый вызов возбудил исключение, то это исключение будет повторно возбуждено методом14611462[`get()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.pool.AsyncResult.get)14631464.14651466#### `[multiprocessing.pool.AsyncResult.wait]wait([timeout])`14671468Ожидает, пока результат не станет доступен или пока не пройдёт14691470*timeout*14711472секунд.14731474#### `[multiprocessing.pool.AsyncResult.ready]ready()`14751476Возвращает, завершён ли вызов.14771478#### `[multiprocessing.pool.AsyncResult.successful]successful()`14791480Возвращает, завершился ли вызов без возбуждения исключения. Возбуждает14811482[`AssertionError`](https://python-all.ru/3.0/library/exceptions.html#exceptions.AssertionError)14831484, если результат ещё не готов.14851486Следующий пример демонстрирует использование пула:14871488```python1489from multiprocessing import Pool14901491def f(x):1492    return x*x14931494if __name__ == '__main__':1495    pool = Pool(processes=4)              # запустить 4 рабочих процесса14961497    result = pool.apply_async(f, (10,))   # асинхронно вычислить "f(10)"1498    print(result.get(timeout=1))          # выводит "100", если компьютер не *очень* медленный14991500    print(pool.map(f, range(10)))         # выводит "[0, 1, 4,..., 81]"15011502    it = pool.imap(f, range(10))1503    print(next(it))                       # выводит "0"1504    print(next(it))                       # выводит "1"1505    print(it.next(timeout=1))             # выводит "4", если компьютер не *очень* медленный15061507    import time1508    result = pool.apply_async(time.sleep, (10,))1509    print(result.get(timeout=1))          # вызывает TimeoutError1510```15111512### Слушатели и клиенты15131514Обычно обмен сообщениями между процессами осуществляется с помощью очередей или с использованием объектов `Connection`, возвращаемых `Pipe()`.15151516Однако модуль `multiprocessing.connection` предоставляет дополнительную гибкость. По сути, он предлагает высокоуровневый API, ориентированный на сообщения, для работы с сокетами или именованными каналами Windows, а также поддерживает *аутентификацию с помощью дайджеста* с использованием модуля [`hmac`](https://python-all.ru/3.0/library/hmac.html#module-hmac).15171518#### `[multiprocessing.connection.deliver_challenge]multiprocessing.connection.deliver_challenge(connection, authkey)`15191520Отправляет случайно сгенерированное сообщение на другой конец соединения и ожидает ответа.15211522Если ответ совпадает с дайджестом сообщения, используя *authkey* в качестве ключа, то на другой конец соединения отправляется приветственное сообщение. В противном случае возбуждается [`AuthenticationError`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.connection.AuthenticationError).15231524#### `[multiprocessing.connection.answerChallenge]multiprocessing.connection.answerChallenge(connection, authkey)`15251526Принимает сообщение, вычисляет дайджест сообщения, используя *authkey* в качестве ключа, а затем отправляет дайджест обратно.15271528Если приветственное сообщение не получено, то возникает исключение [`AuthenticationError`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.connection.AuthenticationError).15291530#### `[multiprocessing.connection.Client]multiprocessing.connection.Client(address[, family[, authenticate[, authkey]]])`15311532Пытается установить соединение с прослушивателем, который использует адрес *address*, и возвращает объект [`Connection`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Connection).15331534Тип соединения определяется аргументом *family*, но его обычно можно опустить, поскольку он обычно определяется по формату *address*. (См. [*Address Formats*](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing-address-formats))15351536Если *authentication* равно `True` или *authkey* является строкой, используется digest-аутентификация. Ключ аутентификации будет либо *authkey*, либо `current_process().authkey)`, если *authkey* равно `None`. Если аутентификация не удалась, возбуждается [`AuthenticationError`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.connection.AuthenticationError). См. [*Ключи аутентификации*](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing-auth-keys).15371538#### `[multiprocessing.connection.Listener]class multiprocessing.connection.Listener([address[, family[, backlog[, authenticate[, authkey]]]]])`15391540Обёртка для привязанного сокета или именованного канала Windows, который «прослушивает» подключения.15411542*address* – это адрес, который будет использоваться привязанным сокетом или именованным каналом объекта слушателя.15431544*family* – тип используемого сокета (или именованного канала). Может быть одной из строк `'AF_INET'` (для TCP-сокета), `'AF_UNIX'` (для сокета домена Unix) или `'AF_PIPE'` (для именованного канала Windows). Из них только первый гарантированно доступен. Если *family* равен `None`, то семейство определяется по формату *address*. Если *address* также равен `None`, то выбирается значение по умолчанию. По умолчанию выбирается семейство, которое считается самым быстрым из доступных. См. [*Address Formats*](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing-address-formats). Обратите внимание: если *family* равен `'AF_UNIX'`, а address равен `None`, то сокет будет создан в частной временной директории, созданной с помощью [`tempfile.mkstemp()`](https://python-all.ru/3.0/library/tempfile.html#tempfile.mkstemp).15451546Если объект прослушивателя использует сокет, то *backlog* (по умолчанию 1) передаётся методу `listen()` сокета после его привязки.15471548Если *authenticate* равен `True` (по умолчанию `False`) или *authkey* не равен `None`, то используется дайджест-аутентификация.15491550Если *authkey* – строка, то она будет использоваться как ключ аутентификации; в противном случае должно быть *None*.15511552Если *authkey* равно `None`, и *authenticate* равно `True`, то `current_process().authkey` используется в качестве ключа аутентификации. Если *authkey* равно `None` и *authentication* равно `False`, то аутентификация не выполняется. Если аутентификация не удалась, возбуждается [`AuthenticationError`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.connection.AuthenticationError). См. [*Ключи аутентификации*](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing-auth-keys).15531554#### `[multiprocessing.connection.Listener.accept]accept()`15551556Принимает соединение на привязанном сокете или именованном канале объекта прослушивателя и возвращает объект15571558`Connection`15591560. Если аутентификация запрошена и не удалась, то возникает исключение15611562[`AuthenticationError`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.connection.AuthenticationError)15631564.15651566#### `[multiprocessing.connection.Listener.close]close()`15671568Закрывает привязанный сокет или именованный канал объекта listener. Этот метод вызывается автоматически при сборке мусора listener'а. Однако рекомендуется вызывать его явно.15691570Объекты Listener имеют следующие свойства только для чтения:15711572#### `[multiprocessing.connection.Listener.address]address`15731574Адрес, используемый объектом Listener.15751576#### `[multiprocessing.connection.Listener.last_accepted]last_accepted`15771578Адрес, с которого поступило последнее принятое соединение. Если он недоступен, то15791580`None`15811582.15831584Модуль определяет два исключения:15851586#### `[multiprocessing.connection.AuthenticationError]exception multiprocessing.connection.AuthenticationError`15871588Исключение, возникающее при ошибке аутентификации.15891590**Примеры**15911592Следующий серверный код создаёт listener, который использует `'secret password'` в качестве ключа аутентификации. Затем он ожидает соединения и отправляет некоторые данные клиенту:15931594```python1595from multiprocessing.connection import Listener1596from array import array15971598address = ('localhost', 6000)     # семейство определяется как 'AF_INET'1599listener = Listener(address, authkey='secret password')16001601conn = listener.accept()1602print('connection accepted from', listener.last_accepted)16031604conn.send([2.25, None, 'junk', float])16051606conn.send_bytes('hello')16071608conn.send_bytes(array('i', [42, 1729]))16091610conn.close()1611listener.close()1612```16131614Следующий код подключается к серверу и получает от него некоторые данные:16151616```python1617from multiprocessing.connection import Client1618from array import array16191620address = ('localhost', 6000)1621conn = Client(address, authkey='secret password')16221623print(conn.recv())                  # => [2.25, None, 'junk', float]16241625print(conn.recv_bytes())            # => 'hello'16261627arr = array('i', [0, 0, 0, 0, 0])1628print(conn.recv_bytes_into(arr))    # => 81629print(arr)                          # => array('i', [42, 1729, 0, 0, 0])16301631conn.close()1632```16331634#### Форматы адресов16351636- Адрес `'AF_INET'` – это кортеж вида `(hostname, port)`, где *hostname* – строка, а *port* – целое число.1637- Адрес `'AF_UNIX'` – это строка, представляющая имя файла в файловой системе.1638- **Адрес `'AF_PIPE'` – это строка вида**16391640  `r'\\.\pipe\PipeName'`. Чтобы использовать [`Client()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.connection.Client) для подключения к именованному каналу на удалённом компьютере с именем *ServerName*, следует использовать адрес вида `r'\\ServerName\pipe\PipeName'` вместо.16411642Обратите внимание, что любая строка, начинающаяся с двух обратных слешей, по умолчанию считается адресом `'AF_PIPE'`, а не адресом `'AF_UNIX'`.16431644### Ключи аутентификации16451646При использовании `Connection.recv()` полученные данные автоматически распаковываются. К сожалению, распаковка данных из ненадёжного источника представляет угрозу безопасности. Поэтому [`Listener`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.connection.Listener) и [`Client()`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.connection.Client) используют модуль [`hmac`](https://python-all.ru/3.0/library/hmac.html#module-hmac) для аутентификации по дайджесту.16471648Ключ аутентификации – это строка, которую можно рассматривать как пароль: как только соединение установлено, обе стороны требуют подтверждения, что другая сторона знает ключ аутентификации. (Доказательство того, что обе стороны используют один и тот же ключ, **не** подразумевает отправку ключа по соединению.)16491650Если аутентификация запрошена, но ключ аутентификации не указан, используется возвращаемое значение `current_process().authkey` (см. [`Process`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Process)). Это значение автоматически наследуется любым объектом [`Process`](https://python-all.ru/3.0/library/multiprocessing.html#multiprocessing.Process), создаваемым текущим процессом. Это означает, что (по умолчанию) все процессы многопроцессной программы будут использовать единый ключ аутентификации, который можно использовать при настройке соединений между ними.16511652Подходящие ключи аутентификации также можно сгенерировать с помощью [`os.urandom()`](https://python-all.ru/3.0/library/os.html#os.urandom).16531654### Журналирование16551656Доступна некоторая поддержка логирования. Однако обратите внимание, что пакет [`logging`](https://python-all.ru/3.0/library/logging.html#module-logging) не использует блокировки, разделяемые между процессами, поэтому возможно (в зависимости от типа обработчика) перемешивание сообщений от разных процессов.16571658#### `[multiprocessing.get_logger]multiprocessing.get_logger()`16591660Возвращает регистратор (logger), используемый модулем `multiprocessing`. При необходимости будет создан новый.16611662При первом создании логгер имеет уровень `logging.NOTSET` и обработчик, который отправляет вывод в [`sys.stderr`](https://python-all.ru/3.0/library/sys.html#sys.stderr), используя формат `'[%(levelname)s/%(processName)s] %(message)s'`. (Логгер допускает использование нестандартного формата `'%(processName)s'`.) Сообщения, отправленные этому логгеру, по умолчанию не распространяются на корневой логгер.16631664Обратите внимание, что в Windows дочерние процессы наследуют только уровень логгера родительского процесса – любые другие настройки логгера не наследуются.16651666Ниже приведён пример сеанса с включённым журналированием:16671668```python1669>>> import multiprocessing, logging1670>>> logger = multiprocessing.get_logger()1671>>> logger.setLevel(logging.INFO)1672>>> logger.warning('doomed')1673[WARNING/MainProcess] doomed1674>>> m = multiprocessing.Manager()1675[INFO/SyncManager-1] child process calling self.run()1676[INFO/SyncManager-1] manager bound to '\\\\.\\pipe\\pyc-2776-0-lj0tfa'1677>>> del m1678[INFO/MainProcess] sending shutdown message to manager1679[INFO/SyncManager-1] manager exiting with exitcode 01680```16811682### Модуль `multiprocessing.dummy`16831684`multiprocessing.dummy` повторяет API модуля `multiprocessing`, но является не более чем обёрткой вокруг модуля [`threading`](https://python-all.ru/3.0/library/threading.html#module-threading).16851686## Рекомендации по программированию16871688При использовании `multiprocessing` следует придерживаться определённых правил и идиом.16891690### Все платформы16911692Избегайте разделяемого состояния16931694> По возможности следует стараться избегать передачи больших объёмов данных между процессами.1695>1696> Вероятно, лучше использовать очереди или каналы для связи между процессами, чем низкоуровневые примитивы синхронизации из модуля [`threading`](https://python-all.ru/3.0/library/threading.html#module-threading).16971698Сериализуемость16991700> Убедитесь, что аргументы методов прокси-объектов сериализуемы.17011702Потокобезопасность прокси17031704> Не используйте прокси-объект из более чем одного потока, если не защищаете его блокировкой.1705>1706> (С разными процессами, использующими *один и тот же* прокси, проблем не возникает.)17071708Присоединение процессов-зомби17091710> В Unix, когда процесс завершается, но к нему не был присоединён вызовом join, он становится зомби. Их не должно быть много, потому что каждый раз при запуске нового процесса (или вызове1711>1712> `active_children()`1713>1714> ) все завершённые процессы, к которым ещё не был применён join, будут присоединены. Также вызов1715>1716> `Process.is_alive()`1717>1718> для завершённого процесса присоединит его. Тем не менее, считается хорошей практикой явно присоединять все запущенные процессы.17191720Лучше наследовать, чем упаковывать/распаковывать17211722> В Windows многие типы из1723>1724> `multiprocessing`1725>1726> должны быть сериализуемыми (picklable), чтобы дочерние процессы могли их использовать. Однако, как правило, следует избегать отправки разделяемых объектов другим процессам через каналы (pipes) или очереди (queues). Вместо этого программу следует организовать так, чтобы процесс, которому нужен доступ к разделяемому ресурсу, созданному в другом месте, мог унаследовать его от родительского процесса.17271728Избегайте завершения процессов17291730> Использование метода `Process.terminate()` для остановки процесса может привести к тому, что любые разделяемые ресурсы (такие как блокировки, семафоры, каналы и очереди), используемые в данный момент процессом, станут повреждёнными или недоступными для других процессов.1731>1732> Поэтому, вероятно, лучше рассматривать использование `Process.terminate()` только для процессов, которые никогда не используют разделяемые ресурсы.17331734Присоединение процессов, использующих очереди17351736> Имейте в виду, что процесс, поместивший элементы в очередь, будет ждать перед завершением, пока все буферизованные элементы не будут переданы потоком «feeder» в нижележащий канал. (Дочерний процесс может вызвать метод очереди `Queue.cancel_join_thread()`, чтобы избежать такого поведения.)1737>1738> Это означает, что при использовании очереди нужно убедиться, что все элементы, помещенные в очередь, в конечном итоге будут удалены до того, как процесс будет присоединен (joined). В противном случае нельзя быть уверенным, что процессы, которые помещали элементы в очередь, завершатся. Также помните, что недемонические процессы будут присоединены автоматически.1739>1740> Пример, который приведёт к взаимоблокировке:1741>1742> ```python1743> from multiprocessing import Process, Queue1744>1745> def f(q):1746>     q.put('X' * 1000000)1747>1748> if __name__ == '__main__':1749>     queue = Queue()1750>     p = Process(target=f, args=(queue,))1751>     p.start()1752>     p.join()                    # это приводит к взаимоблокировке1753>     obj = queue.get()1754> ```1755>1756> Исправлением здесь была бы перестановка двух последних строк (или просто удаление строки с `p.join()`).17571758Явная передача ресурсов дочерним процессам17591760> В Unix дочерний процесс может использовать общий ресурс, созданный в родительском процессе, с помощью глобального ресурса. Однако лучше передавать объект в качестве аргумента конструктору дочернего процесса.1761>1762> Помимо обеспечения (потенциальной) совместимости с Windows, это также гарантирует, что пока дочерний процесс жив, объект не будет собран сборщиком мусора в родительском процессе. Это может быть важно, если какой-либо ресурс освобождается при сборке мусора в родительском процессе.1763>1764> Так, например1765>1766> ```python1767> from multiprocessing import Process, Lock1768>1769> def f():1770>     ... do something using "lock" ...1771>1772> if __name__ == '__main__':1773>    lock = Lock()1774>    for i in range(10):1775>         Process(target=f).start()1776> ```1777>1778> следует переписать как1779>1780> ```python1781> from multiprocessing import Process, Lock1782>1783> def f(l):1784>     ... do something using "l" ...1785>1786> if __name__ == '__main__':1787>    lock = Lock()1788>    for i in range(10):1789>         Process(target=f, args=(lock,)).start()1790> ```17911792### Windows17931794Поскольку в Windows нет [`os.fork()`](https://python-all.ru/3.0/library/os.html#os.fork), существуют несколько дополнительных ограничений:17951796Упаковываемость17971798> Убедитесь, что все аргументы `Process.__init__()` являются сериализуемыми (picklable). Это означает, в частности, что связанные или несвязанные методы нельзя использовать напрямую в качестве аргумента `target` в Windows – просто определите функцию и используйте ее вместо этого.1799>1800> Кроме того, если вы создаёте подкласс `Process`, убедитесь, что экземпляры будут сериализуемы с помощью pickle при вызове метода `Process.start()`.18011802Глобальные переменные18031804> Имейте в виду, что если код, выполняемый в дочернем процессе, пытается получить доступ к глобальной переменной, то значение, которое он видит (если оно есть), может не совпадать со значением в родительском процессе на момент вызова `Process.start()`.1805>1806> Однако глобальные переменные, которые представляют собой просто константы уровня модуля, не вызывают проблем.18071808Безопасный импорт главного модуля18091810> Необходимо убедиться, что главный модуль может быть безопасно импортирован новым интерпретатором Python без непреднамеренных побочных эффектов (например, запуска нового процесса).1811>1812> Например, в Windows запуск следующего модуля завершится ошибкой [`RuntimeError`](https://python-all.ru/3.0/library/exceptions.html#exceptions.RuntimeError):1813>1814> ```python1815> from multiprocessing import Process1816>1817> def foo():1818>     print('hello')1819>1820> p = Process(target=foo)1821> p.start()1822> ```1823>1824> Вместо этого следует защитить «точку входа» программы, используя конструкцию `if __name__ == '__main__':` следующим образом:1825>1826> ```python1827> from multiprocessing import Process, freeze_support1828>1829> def foo():1830>     print('hello')1831>1832> if __name__ == '__main__':1833>     freeze_support()1834>     p = Process(target=foo)1835>     p.start()1836> ```1837>1838> (Строку `freeze_support()` можно опустить, если программа будет запускаться обычным образом, а не в замороженном виде.)1839>1840> Это позволяет новому порождённому интерпретатору Python безопасно импортировать модуль, а затем выполнить функцию `foo()` этого модуля.1841>1842> Аналогичные ограничения действуют, если пул или менеджер создаются в главном модуле.18431844## Примеры18451846Демонстрация создания и использования настраиваемых менеджеров и прокси:18471848```python1849#1850# This module shows how to use arbitrary callables with a subclass of1851# `BaseManager`.1852#1853# Copyright (c) 2006-2008, R Oudkerk1854# All rights reserved.1855#18561857from multiprocessing import freeze_support1858from multiprocessing.managers import BaseManager, BaseProxy1859import operator18601861##18621863class Foo(object):1864    def f(self):1865        print('you called Foo.f()')1866    def g(self):1867        print('you called Foo.g()')1868    def _h(self):1869        print('you called Foo._h()')18701871# A simple generator function1872def baz():1873    for i in range(10):1874        yield i*i18751876# Proxy type for generator objects1877class GeneratorProxy(BaseProxy):1878    _exposed_ = ('next', '__next__')1879    def __iter__(self):1880        return self1881    def __next__(self):1882        return self._callmethod('next')1883    def __next__(self):1884        return self._callmethod('__next__')18851886# Function to return the operator module1887def get_operator_module():1888    return operator18891890##18911892class MyManager(BaseManager):1893    pass18941895# register the Foo class; make `f()` and `g()` accessible via proxy1896MyManager.register('Foo1', Foo)18971898# register the Foo class; make `g()` and `_h()` accessible via proxy1899MyManager.register('Foo2', Foo, exposed=('g', '_h'))19001901# register the generator function baz; use `GeneratorProxy` to make proxies1902MyManager.register('baz', baz, proxytype=GeneratorProxy)19031904# register get_operator_module(); make public functions accessible via proxy1905MyManager.register('operator', get_operator_module)19061907##19081909def test():1910    manager = MyManager()1911    manager.start()19121913    print('-' * 20)19141915    f1 = manager.Foo1()1916    f1.f()1917    f1.g()1918    assert not hasattr(f1, '_h')1919    assert sorted(f1._exposed_) == sorted(['f', 'g'])19201921    print('-' * 20)19221923    f2 = manager.Foo2()1924    f2.g()1925    f2._h()1926    assert not hasattr(f2, 'f')1927    assert sorted(f2._exposed_) == sorted(['g', '_h'])19281929    print('-' * 20)19301931    it = manager.baz()1932    for i in it:1933        print('<%d>' % i, end=' ')1934    print()19351936    print('-' * 20)19371938    op = manager.operator()1939    print('op.add(23, 45) =', op.add(23, 45))1940    print('op.pow(2, 94) =', op.pow(2, 94))1941    print('op.getslice(range(10), 2, 6) =', op.getslice(list(range(10)), 2, 6))1942    print('op.repeat(range(5), 3) =', op.repeat(list(range(5)), 3))1943    print('op._exposed_ =', op._exposed_)19441945##19461947if __name__ == '__main__':1948    freeze_support()1949    test()1950```19511952Использование `Pool`:19531954```python1955#1956# A test of `multiprocessing.Pool` class1957#1958# Copyright (c) 2006-2008, R Oudkerk1959# All rights reserved.1960#19611962import multiprocessing1963import time1964import random1965import sys19661967#1968# Functions used by test code1969#19701971def calculate(func, args):1972    result = func(*args)1973    return '%s says that %s%s = %s' % (1974        multiprocessing.current_process().name,1975        func.__name__, args, result1976        )19771978def calculatestar(args):1979    return calculate(*args)19801981def mul(a, b):1982    time.sleep(0.5*random.random())1983    return a * b19841985def plus(a, b):1986    time.sleep(0.5*random.random())1987    return a + b19881989def f(x):1990    return 1.0 / (x-5.0)19911992def pow3(x):1993    return x**319941995def noop(x):1996    pass19971998#1999# Test code2000#20012002def test():2003    print('cpu_count() = %d\n' % multiprocessing.cpu_count())20042005    #2006    # Create pool2007    #20082009    PROCESSES = 42010    print('Creating pool with %d processes\n' % PROCESSES)2011    pool = multiprocessing.Pool(PROCESSES)2012    print('pool = %s' % pool)2013    print()20142015    #2016    # Tests2017    #20182019    TASKS = [(mul, (i, 7)) for i in range(10)] + \2020            [(plus, (i, 8)) for i in range(10)]20212022    results = [pool.apply_async(calculate, t) for t in TASKS]2023    imap_it = pool.imap(calculatestar, TASKS)2024    imap_unordered_it = pool.imap_unordered(calculatestar, TASKS)20252026    print('Ordered results using pool.apply_async():')2027    for r in results:2028        print('\t', r.get())2029    print()20302031    print('Ordered results using pool.imap():')2032    for x in imap_it:2033        print('\t', x)2034    print()20352036    print('Unordered results using pool.imap_unordered():')2037    for x in imap_unordered_it:2038        print('\t', x)2039    print()20402041    print('Ordered results using pool.map() --- will block till complete:')2042    for x in pool.map(calculatestar, TASKS):2043        print('\t', x)2044    print()20452046    #2047    # Simple benchmarks2048    #20492050    N = 1000002051    print('def pow3(x): return x**3')20522053    t = time.time()2054    A = list(map(pow3, range(N)))2055    print('\tmap(pow3, xrange(%d)):\n\t\t%s seconds' % \2056          (N, time.time() - t))20572058    t = time.time()2059    B = pool.map(pow3, range(N))2060    print('\tpool.map(pow3, xrange(%d)):\n\t\t%s seconds' % \2061          (N, time.time() - t))20622063    t = time.time()2064    C = list(pool.imap(pow3, range(N), chunksize=N//8))2065    print('\tlist(pool.imap(pow3, xrange(%d), chunksize=%d)):\n\t\t%s' \2066          ' seconds' % (N, N//8, time.time() - t))20672068    assert A == B == C, (len(A), len(B), len(C))2069    print()20702071    L = [None] * 10000002072    print('def noop(x): pass')2073    print('L = [None] * 1000000')20742075    t = time.time()2076    A = list(map(noop, L))2077    print('\tmap(noop, L):\n\t\t%s seconds' % \2078          (time.time() - t))20792080    t = time.time()2081    B = pool.map(noop, L)2082    print('\tpool.map(noop, L):\n\t\t%s seconds' % \2083          (time.time() - t))20842085    t = time.time()2086    C = list(pool.imap(noop, L, chunksize=len(L)//8))2087    print('\tlist(pool.imap(noop, L, chunksize=%d)):\n\t\t%s seconds' % \2088          (len(L)//8, time.time() - t))20892090    assert A == B == C, (len(A), len(B), len(C))2091    print()20922093    del A, B, C, L20942095    #2096    # Test error handling2097    #20982099    print('Testing error handling:')21002101    try:2102        print(pool.apply(f, (5,)))2103    except ZeroDivisionError:2104        print('\tGot ZeroDivisionError as expected from pool.apply()')2105    else:2106        raise AssertionError('expected ZeroDivisionError')21072108    try:2109        print(pool.map(f, list(range(10))))2110    except ZeroDivisionError:2111        print('\tGot ZeroDivisionError as expected from pool.map()')2112    else:2113        raise AssertionError('expected ZeroDivisionError')21142115    try:2116        print(list(pool.imap(f, list(range(10)))))2117    except ZeroDivisionError:2118        print('\tGot ZeroDivisionError as expected from list(pool.imap())')2119    else:2120        raise AssertionError('expected ZeroDivisionError')21212122    it = pool.imap(f, list(range(10)))2123    for i in range(10):2124        try:2125            x = next(it)2126        except ZeroDivisionError:2127            if i == 5:2128                pass2129        except StopIteration:2130            break2131        else:2132            if i == 5:2133                raise AssertionError('expected ZeroDivisionError')21342135    assert i == 92136    print('\tGot ZeroDivisionError as expected from IMapIterator.next()')2137    print()21382139    #2140    # Testing timeouts2141    #21422143    print('Testing ApplyResult.get() with timeout:', end=' ')2144    res = pool.apply_async(calculate, TASKS[0])2145    while 1:2146        sys.stdout.flush()2147        try:2148            sys.stdout.write('\n\t%s' % res.get(0.02))2149            break2150        except multiprocessing.TimeoutError:2151            sys.stdout.write('.')2152    print()2153    print()21542155    print('Testing IMapIterator.next() with timeout:', end=' ')2156    it = pool.imap(calculatestar, TASKS)2157    while 1:2158        sys.stdout.flush()2159        try:2160            sys.stdout.write('\n\t%s' % it.next(0.02))2161        except StopIteration:2162            break2163        except multiprocessing.TimeoutError:2164            sys.stdout.write('.')2165    print()2166    print()21672168    #2169    # Testing callback2170    #21712172    print('Testing callback:')21732174    A = []2175    B = [56, 0, 1, 8, 27, 64, 125, 216, 343, 512, 729]21762177    r = pool.apply_async(mul, (7, 8), callback=A.append)2178    r.wait()21792180    r = pool.map_async(pow3, list(range(10)), callback=A.extend)2181    r.wait()21822183    if A == B:2184        print('\tcallbacks succeeded\n')2185    else:2186        print('\t*** callbacks failed\n\t\t%s != %s\n' % (A, B))21872188    #2189    # Check there are no outstanding tasks2190    #21912192    assert not pool._cache, 'cache = %r' % pool._cache21932194    #2195    # Check close() methods2196    #21972198    print('Testing close():')21992200    for worker in pool._pool:2201        assert worker.is_alive()22022203    result = pool.apply_async(time.sleep, [0.5])2204    pool.close()2205    pool.join()22062207    assert result.get() is None22082209    for worker in pool._pool:2210        assert not worker.is_alive()22112212    print('\tclose() succeeded\n')22132214    #2215    # Check terminate() method2216    #22172218    print('Testing terminate():')22192220    pool = multiprocessing.Pool(2)2221    DELTA = 0.12222    ignore = pool.apply(pow3, [2])2223    results = [pool.apply_async(time.sleep, [DELTA]) for i in range(100)]2224    pool.terminate()2225    pool.join()22262227    for worker in pool._pool:2228        assert not worker.is_alive()22292230    print('\tterminate() succeeded\n')22312232    #2233    # Check garbage collection2234    #22352236    print('Testing garbage collection:')22372238    pool = multiprocessing.Pool(2)2239    DELTA = 0.12240    processes = pool._pool2241    ignore = pool.apply(pow3, [2])2242    results = [pool.apply_async(time.sleep, [DELTA]) for i in range(100)]22432244    results = pool = None22452246    time.sleep(DELTA * 2)22472248    for worker in processes:2249        assert not worker.is_alive()22502251    print('\tgarbage collection succeeded\n')22522253if __name__ == '__main__':2254    multiprocessing.freeze_support()22552256    assert len(sys.argv) in (1, 2)22572258    if len(sys.argv) == 1 or sys.argv[1] == 'processes':2259        print(' Using processes '.center(79, '-'))2260    elif sys.argv[1] == 'threads':2261        print(' Using threads '.center(79, '-'))2262        import multiprocessing.dummy as multiprocessing2263    else:2264        print('Usage:\n\t%s [processes | threads]' % sys.argv[0])2265        raise SystemExit(2)22662267    test()2268```22692270Типы синхронизации, такие как блокировки, условия и очереди:22712272```python2273#2274# A test file for the `multiprocessing` package2275#2276# Copyright (c) 2006-2008, R Oudkerk2277# All rights reserved.2278#22792280import time, sys, random2281from queue import Empty22822283import multiprocessing               # may get overwritten22842285#### TEST_VALUE22862287def value_func(running, mutex):2288    random.seed()2289    time.sleep(random.random()*4)22902291    mutex.acquire()2292    print('\n\t\t\t' + str(multiprocessing.current_process()) + ' has finished')2293    running.value -= 12294    mutex.release()22952296def test_value():2297    TASKS = 102298    running = multiprocessing.Value('i', TASKS)2299    mutex = multiprocessing.Lock()23002301    for i in range(TASKS):2302        p = multiprocessing.Process(target=value_func, args=(running, mutex))2303        p.start()23042305    while running.value > 0:2306        time.sleep(0.08)2307        mutex.acquire()2308        print(running.value, end=' ')2309        sys.stdout.flush()2310        mutex.release()23112312    print()2313    print('No more running processes')23142315#### TEST_QUEUE23162317def queue_func(queue):2318    for i in range(30):2319        time.sleep(0.5 * random.random())2320        queue.put(i*i)2321    queue.put('STOP')23222323def test_queue():2324    q = multiprocessing.Queue()23252326    p = multiprocessing.Process(target=queue_func, args=(q,))2327    p.start()23282329    o = None2330    while o != 'STOP':2331        try:2332            o = q.get(timeout=0.3)2333            print(o, end=' ')2334            sys.stdout.flush()2335        except Empty:2336            print('TIMEOUT')23372338    print()23392340#### TEST_CONDITION23412342def condition_func(cond):2343    cond.acquire()2344    print('\t' + str(cond))2345    time.sleep(2)2346    print('\tchild is notifying')2347    print('\t' + str(cond))2348    cond.notify()2349    cond.release()23502351def test_condition():2352    cond = multiprocessing.Condition()23532354    p = multiprocessing.Process(target=condition_func, args=(cond,))2355    print(cond)23562357    cond.acquire()2358    print(cond)2359    cond.acquire()2360    print(cond)23612362    p.start()23632364    print('main is waiting')2365    cond.wait()2366    print('main has woken up')23672368    print(cond)2369    cond.release()2370    print(cond)2371    cond.release()23722373    p.join()2374    print(cond)23752376#### TEST_SEMAPHORE23772378def semaphore_func(sema, mutex, running):2379    sema.acquire()23802381    mutex.acquire()2382    running.value += 12383    print(running.value, 'tasks are running')2384    mutex.release()23852386    random.seed()2387    time.sleep(random.random()*2)23882389    mutex.acquire()2390    running.value -= 12391    print('%s has finished' % multiprocessing.current_process())2392    mutex.release()23932394    sema.release()23952396def test_semaphore():2397    sema = multiprocessing.Semaphore(3)2398    mutex = multiprocessing.RLock()2399    running = multiprocessing.Value('i', 0)24002401    processes = [2402        multiprocessing.Process(target=semaphore_func,2403                                args=(sema, mutex, running))2404        for i in range(10)2405        ]24062407    for p in processes:2408        p.start()24092410    for p in processes:2411        p.join()24122413#### TEST_JOIN_TIMEOUT24142415def join_timeout_func():2416    print('\tchild sleeping')2417    time.sleep(5.5)2418    print('\n\tchild terminating')24192420def test_join_timeout():2421    p = multiprocessing.Process(target=join_timeout_func)2422    p.start()24232424    print('waiting for process to finish')24252426    while 1:2427        p.join(timeout=1)2428        if not p.is_alive():2429            break2430        print('.', end=' ')2431        sys.stdout.flush()24322433#### TEST_EVENT24342435def event_func(event):2436    print('\t%r is waiting' % multiprocessing.current_process())2437    event.wait()2438    print('\t%r has woken up' % multiprocessing.current_process())24392440def test_event():2441    event = multiprocessing.Event()24422443    processes = [multiprocessing.Process(target=event_func, args=(event,))2444                 for i in range(5)]24452446    for p in processes:2447        p.start()24482449    print('main is sleeping')2450    time.sleep(2)24512452    print('main is setting event')2453    event.set()24542455    for p in processes:2456        p.join()24572458#### TEST_SHAREDVALUES24592460def sharedvalues_func(values, arrays, shared_values, shared_arrays):2461    for i in range(len(values)):2462        v = values[i][1]2463        sv = shared_values[i].value2464        assert v == sv24652466    for i in range(len(values)):2467        a = arrays[i][1]2468        sa = list(shared_arrays[i][:])2469        assert a == sa24702471    print('Tests passed')24722473def test_sharedvalues():2474    values = [2475        ('i', 10),2476        ('h', -2),2477        ('d', 1.25)2478        ]2479    arrays = [2480        ('i', list(range(100))),2481        ('d', [0.25 * i for i in range(100)]),2482        ('H', list(range(1000)))2483        ]24842485    shared_values = [multiprocessing.Value(id, v) for id, v in values]2486    shared_arrays = [multiprocessing.Array(id, a) for id, a in arrays]24872488    p = multiprocessing.Process(2489        target=sharedvalues_func,2490        args=(values, arrays, shared_values, shared_arrays)2491        )2492    p.start()2493    p.join()24942495    assert p.exitcode == 024962497####24982499def test(namespace=multiprocessing):2500    global multiprocessing25012502    multiprocessing = namespace25032504    for func in [ test_value, test_queue, test_condition,2505                  test_semaphore, test_join_timeout, test_event,2506                  test_sharedvalues ]:25072508        print('\n\t######## %s\n' % func.__name__)2509        func()25102511    ignore = multiprocessing.active_children()      # cleanup any old processes2512    if hasattr(multiprocessing, '_debug_info'):2513        info = multiprocessing._debug_info()2514        if info:2515            print(info)2516            raise ValueError('there should be no positive refcounts left')25172518if __name__ == '__main__':2519    multiprocessing.freeze_support()25202521    assert len(sys.argv) in (1, 2)25222523    if len(sys.argv) == 1 or sys.argv[1] == 'processes':2524        print(' Using processes '.center(79, '-'))2525        namespace = multiprocessing2526    elif sys.argv[1] == 'manager':2527        print(' Using processes and a manager '.center(79, '-'))2528        namespace = multiprocessing.Manager()2529        namespace.Process = multiprocessing.Process2530        namespace.current_process = multiprocessing.current_process2531        namespace.active_children = multiprocessing.active_children2532    elif sys.argv[1] == 'threads':2533        print(' Using threads '.center(79, '-'))2534        import multiprocessing.dummy as namespace2535    else:2536        print('Usage:\n\t%s [processes | manager | threads]' % sys.argv[0])2537        raise SystemExit(2)25382539    test(namespace)2540```25412542Пример, показывающий, как использовать очереди для передачи задач набору рабочих процессов и сбора результатов:25432544```python2545#2546# Простой пример использования пула рабочих процессов для выполнения задач.2547#2548# Обратите внимание, что результаты, скорее всего, не будут получены из выходной2549# очереди в том же порядке, в котором соответствующие задачи были2550# помещены во входную очередь. Если важно получить результаты2551# в исходном порядке, рассмотрите использование `Pool.map()` или2552# `Pool.imap()` (что в любом случае сократит объём необходимого кода).2553#2554# Авторское право (c) 2006-2008, R Oudkerk2555# Все права защищены.2556#25572558import time2559import random25602561from multiprocessing import Process, Queue, current_process, freeze_support25622563#2564# Функция, выполняемая рабочими процессами2565#25662567def worker(input, output):2568    for func, args in iter(input.get, 'STOP'):2569        result = calculate(func, args)2570        output.put(result)25712572#2573# Функция, используемая для вычисления результата2574#25752576def calculate(func, args):2577    result = func(*args)2578    return '%s says that %s%s = %s' % \2579        (current_process().name, func.__name__, args, result)25802581#2582# Функции, на которые ссылаются задачи2583#25842585def mul(a, b):2586    time.sleep(0.5*random.random())2587    return a * b25882589def plus(a, b):2590    time.sleep(0.5*random.random())2591    return a + b25922593#2594#2595#25962597def test():2598    NUMBER_OF_PROCESSES = 42599    TASKS1 = [(mul, (i, 7)) for i in range(20)]2600    TASKS2 = [(plus, (i, 8)) for i in range(10)]26012602    # Создание очередей2603    task_queue = Queue()2604    done_queue = Queue()26052606    # Отправка задач2607    for task in TASKS1:2608        task_queue.put(task)26092610    # Запуск рабочих процессов2611    for i in range(NUMBER_OF_PROCESSES):2612        Process(target=worker, args=(task_queue, done_queue)).start()26132614    # Получение и вывод результатов2615    print('Unordered results:')2616    for i in range(len(TASKS1)):2617        print('\t', done_queue.get())26182619    # Добавить ещё задач с помощью `put()`2620    for task in TASKS2:2621        task_queue.put(task)26222623    # Получить и вывести ещё несколько результатов2624    for i in range(len(TASKS2)):2625        print('\t', done_queue.get())26262627    # Сообщить дочерним процессам об остановке2628    for i in range(NUMBER_OF_PROCESSES):2629        task_queue.put('STOP')26302631if __name__ == '__main__':2632    freeze_support()2633    test()2634```26352636Пример того, как пул рабочих процессов может каждый запустить экземпляр `SimpleHTTPServer.HttpServer`, разделяя один прослушивающий сокет.26372638```python2639#2640# Пример, в котором пул HTTP-серверов использует один общий слушающий сокет2641#2642# В Windows этот модуль зависит от возможности сериализации сокета с помощью pickle2643# объекта, чтобы рабочие процессы могли унаследовать копию сервера2644# объект. (Мы импортируем `multiprocessing.reduction`, чтобы включить эту сериализацию.)2645#2646# Не уверены, нужно ли синхронизировать доступ к методу `socket.accept()` с помощью2647# блокировки, разделяемой между процессами – похоже, это не обязательно.2648#2649# Авторское право (c) 2006-2008, R Oudkerk2650# Все права защищены.2651#26522653import os2654import sys26552656from multiprocessing import Process, current_process, freeze_support2657from http.server import HTTPServer2658from http.server import SimpleHTTPRequestHandler26592660if sys.platform == 'win32':2661    import multiprocessing.reduction    # сделать сокеты сериализуемыми/наследуемыми26622663def note(format, *args):2664    sys.stderr.write('[%s]\t%s\n' % (current_process().name, format%args))26652666class RequestHandler(SimpleHTTPRequestHandler):2667    # мы переопределяем log_message(), чтобы показывать, какой процесс обрабатывает запрос2668    def log_message(self, format, *args):2669        note(format, *args)26702671def serve_forever(server):2672    note('starting server')2673    try:2674        server.serve_forever()2675    except KeyboardInterrupt:2676        pass26772678def runpool(address, number_of_processes):2679    # создать единственный объект сервера – каждый дочерний процесс унаследует его копию2680    server = HTTPServer(address, RequestHandler)26812682    # создать дочерние процессы, которые будут работать как воркеры2683    for i in range(number_of_processes-1):2684        Process(target=serve_forever, args=(server,)).start()26852686    # главный процесс также выступает в роли воркера2687    serve_forever(server)26882689def test():2690    DIR = os.path.join(os.path.dirname(__file__), '..')2691    ADDRESS = ('localhost', 8000)2692    NUMBER_OF_PROCESSES = 426932694    print('Serving at http://%s:%d using %d worker processes' % \2695          (ADDRESS[0], ADDRESS[1], NUMBER_OF_PROCESSES))2696    print('To exit press Ctrl-' + ['C', 'Break'][sys.platform=='win32'])26972698    os.chdir(DIR)2699    runpool(ADDRESS, NUMBER_OF_PROCESSES)27002701if __name__ == '__main__':2702    freeze_support()2703    test()2704```27052706Некоторые простые тесты производительности, сравнивающие `multiprocessing` с [`threading`](https://python-all.ru/3.0/library/threading.html#module-threading):27072708```python2709#2710# Простые тесты производительности для пакета multiprocessing2711#2712# Авторское право (c) 2006-2008, R Oudkerk2713# Все права защищены.2714#27152716import time, sys, multiprocessing, threading, queue, gc27172718if sys.platform == 'win32':2719    _timer = time.clock2720else:2721    _timer = time.time27222723delta = 127242725#### TEST_QUEUESPEED27262727def queuespeed_func(q, c, iterations):2728    a = '0' * 2562729    c.acquire()2730    c.notify()2731    c.release()27322733    for i in range(iterations):2734        q.put(a)27352736    q.put('STOP')27372738def test_queuespeed(Process, q, c):2739    elapsed = 02740    iterations = 127412742    while elapsed < delta:2743        iterations *= 227442745        p = Process(target=queuespeed_func, args=(q, c, iterations))2746        c.acquire()2747        p.start()2748        c.wait()2749        c.release()27502751        result = None2752        t = _timer()27532754        while result != 'STOP':2755            result = q.get()27562757        elapsed = _timer() - t27582759        p.join()27602761    print(iterations, 'objects passed through the queue in', elapsed, 'seconds')2762    print('average number/sec:', iterations/elapsed)27632764#### TEST_PIPESPEED27652766def pipe_func(c, cond, iterations):2767    a = '0' * 2562768    cond.acquire()2769    cond.notify()2770    cond.release()27712772    for i in range(iterations):2773        c.send(a)27742775    c.send('STOP')27762777def test_pipespeed():2778    c, d = multiprocessing.Pipe()2779    cond = multiprocessing.Condition()2780    elapsed = 02781    iterations = 127822783    while elapsed < delta:2784        iterations *= 227852786        p = multiprocessing.Process(target=pipe_func,2787                                    args=(d, cond, iterations))2788        cond.acquire()2789        p.start()2790        cond.wait()2791        cond.release()27922793        result = None2794        t = _timer()27952796        while result != 'STOP':2797            result = c.recv()27982799        elapsed = _timer() - t2800        p.join()28012802    print(iterations, 'objects passed through connection in',elapsed,'seconds')2803    print('average number/sec:', iterations/elapsed)28042805#### TEST_SEQSPEED28062807def test_seqspeed(seq):2808    elapsed = 02809    iterations = 128102811    while elapsed < delta:2812        iterations *= 228132814        t = _timer()28152816        for i in range(iterations):2817            a = seq[5]28182819        elapsed = _timer()-t28202821    print(iterations, 'iterations in', elapsed, 'seconds')2822    print('average number/sec:', iterations/elapsed)28232824#### TEST_LOCK28252826def test_lockspeed(l):2827    elapsed = 02828    iterations = 128292830    while elapsed < delta:2831        iterations *= 228322833        t = _timer()28342835        for i in range(iterations):2836            l.acquire()2837            l.release()28382839        elapsed = _timer()-t28402841    print(iterations, 'iterations in', elapsed, 'seconds')2842    print('average number/sec:', iterations/elapsed)28432844#### TEST_CONDITION28452846def conditionspeed_func(c, N):2847    c.acquire()2848    c.notify()28492850    for i in range(N):2851        c.wait()2852        c.notify()28532854    c.release()28552856def test_conditionspeed(Process, c):2857    elapsed = 02858    iterations = 128592860    while elapsed < delta:2861        iterations *= 228622863        c.acquire()2864        p = Process(target=conditionspeed_func, args=(c, iterations))2865        p.start()28662867        c.wait()28682869        t = _timer()28702871        for i in range(iterations):2872            c.notify()2873            c.wait()28742875        elapsed = _timer()-t28762877        c.release()2878        p.join()28792880    print(iterations * 2, 'waits in', elapsed, 'seconds')2881    print('average number/sec:', iterations * 2 / elapsed)28822883####28842885def test():2886    manager = multiprocessing.Manager()28872888    gc.disable()28892890    print('\n\t######## testing Queue.Queue\n')2891    test_queuespeed(threading.Thread, queue.Queue(),2892                    threading.Condition())2893    print('\n\t######## testing multiprocessing.Queue\n')2894    test_queuespeed(multiprocessing.Process, multiprocessing.Queue(),2895                    multiprocessing.Condition())2896    print('\n\t######## testing Queue managed by server process\n')2897    test_queuespeed(multiprocessing.Process, manager.Queue(),2898                    manager.Condition())2899    print('\n\t######## testing multiprocessing.Pipe\n')2900    test_pipespeed()29012902    print()29032904    print('\n\t######## testing list\n')2905    test_seqspeed(list(range(10)))2906    print('\n\t######## testing list managed by server process\n')2907    test_seqspeed(manager.list(list(range(10))))2908    print('\n\t######## testing Array("i", ..., lock=False)\n')2909    test_seqspeed(multiprocessing.Array('i', list(range(10)), lock=False))2910    print('\n\t######## testing Array("i", ..., lock=True)\n')2911    test_seqspeed(multiprocessing.Array('i', list(range(10)), lock=True))29122913    print()29142915    print('\n\t######## testing threading.Lock\n')2916    test_lockspeed(threading.Lock())2917    print('\n\t######## testing threading.RLock\n')2918    test_lockspeed(threading.RLock())2919    print('\n\t######## testing multiprocessing.Lock\n')2920    test_lockspeed(multiprocessing.Lock())2921    print('\n\t######## testing multiprocessing.RLock\n')2922    test_lockspeed(multiprocessing.RLock())2923    print('\n\t######## testing lock managed by server process\n')2924    test_lockspeed(manager.Lock())2925    print('\n\t######## testing rlock managed by server process\n')2926    test_lockspeed(manager.RLock())29272928    print()29292930    print('\n\t######## testing threading.Condition\n')2931    test_conditionspeed(threading.Thread, threading.Condition())2932    print('\n\t######## testing multiprocessing.Condition\n')2933    test_conditionspeed(multiprocessing.Process, multiprocessing.Condition())2934    print('\n\t######## testing condition managed by a server process\n')2935    test_conditionspeed(multiprocessing.Process, manager.Condition())29362937    gc.enable()29382939if __name__ == '__main__':2940    multiprocessing.freeze_support()2941    test()2942```29432944Пример/демонстрация того, как использовать `managers.SyncManager`, `Process` и другие для построения системы, которая может распределять процессы и работу через распределенную очередь на «кластер» машин в сети, доступный по SSH. Для этого потребуется настроить аутентификацию по закрытому ключу для всех хостов.29452946```python2947#2948# Модуль для запуска процессов на удалённом хосте2949#2950# Зависит от пакета `multiprocessing` – тестировалось с `processing-0.60`2951#2952# Авторское право (c) 2006-2008, R Oudkerk2953# Все права защищены.2954#29552956__all__ = ['Cluster', 'Host', 'get_logger', 'current_process']29572958#2959# Импорты2960#29612962import sys2963import os2964import tarfile2965import shutil2966import subprocess2967import logging2968import itertools2969import queue29702971try:2972    import pickle as pickle2973except ImportError:2974    import pickle29752976from multiprocessing import Process, current_process, cpu_count2977from multiprocessing import util, managers, connection, forking, pool29782979#2980# Логирование2981#29822983def get_logger():2984    return _logger29852986_logger = logging.getLogger('distributing')2987_logger.propogate = 029882989_formatter = logging.Formatter(util.DEFAULT_LOGGING_FORMAT)2990_handler = logging.StreamHandler()2991_handler.setFormatter(_formatter)2992_logger.addHandler(_handler)29932994info = _logger.info2995debug = _logger.debug29962997#2998# Получить количество процессоров2999#30003001try:3002    slot_count = cpu_count()3003except NotImplemented:3004    slot_count = 130053006#3007# Тип менеджера, который порождает подпроцессы3008#30093010class HostManager(managers.SyncManager):3011    '''3012    Тип менеджера, используемый для запуска процессов на (предположительно) удалённом хосте3013    '''3014    def __init__(self, address, authkey):3015        managers.SyncManager.__init__(self, address, authkey)3016        self._name = 'Host-unknown'30173018    def Process(self, group=None, target=None, name=None, args=(), kwargs={}):3019        if hasattr(sys.modules['__main__'], '__file__'):3020            main_path = os.path.basename(sys.modules['__main__'].__file__)3021        else:3022            main_path = None3023        data = pickle.dumps((target, args, kwargs))3024        p = self._RemoteProcess(data, main_path)3025        if name is None:3026            temp = self._name.split('Host-')[-1] + '/Process-%s'3027            name = temp % ':'.join(map(str, p.get_identity()))3028        p.set_name(name)3029        return p30303031    @classmethod3032    def from_address(cls, address, authkey):3033        manager = cls(address, authkey)3034        managers.transact(address, authkey, 'dummy')3035        manager._state.value = managers.State.STARTED3036        manager._name = 'Host-%s:%s' % manager.address3037        manager.shutdown = util.Finalize(3038            manager, HostManager._finalize_host,3039            args=(manager._address, manager._authkey, manager._name),3040            exitpriority=-103041            )3042        return manager30433044    @staticmethod3045    def _finalize_host(address, authkey, name):3046        managers.transact(address, authkey, 'shutdown')30473048    def __repr__(self):3049        return '<Host(%s)>' % self._name30503051#3052# Подкласс Process, представляющий процесс на (возможно) удалённой машине3053#30543055class RemoteProcess(Process):3056    '''3057    Представляет процесс, запущенный на удалённом хосте3058    '''3059    def __init__(self, data, main_path):3060        assert not main_path or os.path.basename(main_path) == main_path3061        Process.__init__(self)3062        self._data = data3063        self._main_path = main_path30643065    def _bootstrap(self):3066        forking.prepare({'main_path': self._main_path})3067        self._target, self._args, self._kwargs = pickle.loads(self._data)3068        return Process._bootstrap(self)30693070    def get_identity(self):3071        return self._identity30723073HostManager.register('_RemoteProcess', RemoteProcess)30743075#3076# Класс Pool, использующий кластер3077#30783079class DistributedPool(pool.Pool):30803081    def __init__(self, cluster, processes=None, initializer=None, initargs=()):3082        self._cluster = cluster3083        self.Process = cluster.Process3084        pool.Pool.__init__(self, processes or len(cluster),3085                           initializer, initargs)30863087    def _setup_queues(self):3088        self._inqueue = self._cluster._SettableQueue()3089        self._outqueue = self._cluster._SettableQueue()3090        self._quick_put = self._inqueue.put3091        self._quick_get = self._outqueue.get30923093    @staticmethod3094    def _help_stuff_finish(inqueue, task_handler, size):3095        inqueue.set_contents([None] * size)30963097#3098# Тип менеджера, который запускает менеджеры хостов на других машинах3099#31003101def LocalProcess(**kwds):3102    p = Process(**kwds)3103    p.set_name('localhost/' + p.name)3104    return p31053106class Cluster(managers.SyncManager):3107    '''3108    Представляет набор слотов, работающих на разных хостах31093110    `Cluster` является подклассом `SyncManager`, поэтому он позволяет создавать3111    различные типы разделяемых объектов.3112    '''3113    def __init__(self, hostlist, modules):3114        managers.SyncManager.__init__(self, address=('localhost', 0))3115        self._hostlist = hostlist3116        self._modules = modules3117        if __name__ not in modules:3118            modules.append(__name__)3119        files = [sys.modules[name].__file__ for name in modules]3120        for i, file in enumerate(files):3121            if file.endswith('.pyc') or file.endswith('.pyo'):3122                files[i] = file[:-4] + '.py'3123        self._files = [os.path.abspath(file) for file in files]31243125    def start(self):3126        managers.SyncManager.start(self)31273128        l = connection.Listener(family='AF_INET', authkey=self._authkey)31293130        for i, host in enumerate(self._hostlist):3131            host._start_manager(i, self._authkey, l.address, self._files)31323133        for host in self._hostlist:3134            if host.hostname != 'localhost':3135                conn = l.accept()3136                i, address, cpus = conn.recv()3137                conn.close()3138                other_host = self._hostlist[i]3139                other_host.manager = HostManager.from_address(address,3140                                                              self._authkey)3141                other_host.slots = other_host.slots or cpus3142                other_host.Process = other_host.manager.Process3143            else:3144                host.slots = host.slots or slot_count3145                host.Process = LocalProcess31463147        self._slotlist = [3148            Slot(host) for host in self._hostlist for i in range(host.slots)3149            ]3150        self._slot_iterator = itertools.cycle(self._slotlist)3151        self._base_shutdown = self.shutdown3152        del self.shutdown31533154    def shutdown(self):3155        for host in self._hostlist:3156            if host.hostname != 'localhost':3157                host.manager.shutdown()3158        self._base_shutdown()31593160    def Process(self, group=None, target=None, name=None, args=(), kwargs={}):3161        slot = next(self._slot_iterator)3162        return slot.Process(3163            group=group, target=target, name=name, args=args, kwargs=kwargs3164            )31653166    def Pool(self, processes=None, initializer=None, initargs=()):3167        return DistributedPool(self, processes, initializer, initargs)31683169    def __getitem__(self, i):3170        return self._slotlist[i]31713172    def __len__(self):3173        return len(self._slotlist)31743175    def __iter__(self):3176        return iter(self._slotlist)31773178#3179# Подкласс Queue, используемый распределённым пулом3180#31813182class SettableQueue(queue.Queue):3183    def empty(self):3184        return not self.queue3185    def full(self):3186        return self.maxsize > 0 and len(self.queue) == self.maxsize3187    def set_contents(self, contents):3188        # длина содержимого должна быть не меньше количества3189        # потоков, которые потенциально вызвали get()3190        self.not_empty.acquire()3191        try:3192            self.queue.clear()3193            self.queue.extend(contents)3194            self.not_empty.notifyAll()3195        finally:3196            self.not_empty.release()31973198Cluster.register('_SettableQueue', SettableQueue)31993200#3201# Класс, представляющий логический процессор в кластере3202#32033204class Slot(object):3205    def __init__(self, host):3206        self.host = host3207        self.Process = host.Process32083209#3210# Хост3211#32123213class Host(object):3214    '''3215    Представляет хост, используемый как узел в кластере.32163217    `hostname` задаёт имя хоста. Если hostname не3218    то для входа на хост используется ssh. Для входа под3219    от имени другого пользователя используйте имя хоста вида3220    "username@somewhere.org"32213222    `slots` используется для указания количества слотов для процессов на3223    хосте. Это влияет на то, как часто процессы будут распределяться по3224    этому хосту. Обычно это должно быть равно количеству процессоров на3225    том хосте.3226    '''3227    def __init__(self, hostname, slots=None):3228        self.hostname = hostname3229        self.slots = slots32303231    def _start_manager(self, index, authkey, address, files):3232        if self.hostname != 'localhost':3233            tempdir = copy_to_remote_temporary_directory(self.hostname, files)3234            debug('startup files copied to %s:%s', self.hostname, tempdir)3235            p = subprocess.Popen(3236                ['ssh', self.hostname, 'python', '-c',3237                 '"import os; os.chdir(%r); '3238                 'from distributing import main; main()"' % tempdir],3239                stdin=subprocess.PIPE3240                )3241            data = dict(3242                name='BoostrappingHost', index=index,3243                dist_log_level=_logger.getEffectiveLevel(),3244                dir=tempdir, authkey=str(authkey), parent_address=address3245                )3246            pickle.dump(data, p.stdin, pickle.HIGHEST_PROTOCOL)3247            p.stdin.close()32483249#3250# Копирует файлы в удалённый каталог, возвращая имя каталога.3251#32523253unzip_code = '''"3254import tempfile, os, sys, tarfile3255tempdir = tempfile.mkdtemp(prefix='distrib-')3256os.chdir(tempdir)3257tf = tarfile.open(fileobj=sys.stdin, mode='r|gz')3258for ti in tf:3259    tf.extract(ti)3260print tempdir3261"'''32623263def copy_to_remote_temporary_directory(host, files):3264    p = subprocess.Popen(3265        ['ssh', host, 'python', '-c', unzip_code],3266        stdout=subprocess.PIPE, stdin=subprocess.PIPE3267        )3268    tf = tarfile.open(fileobj=p.stdin, mode='w|gz')3269    for name in files:3270        tf.add(name, os.path.basename(name))3271    tf.close()3272    p.stdin.close()3273    return p.stdout.read().rstrip()32743275#3276# Код, который запускает менеджер хостов.3277#32783279def main():3280    # Получает данные от родительского процесса через stdin.3281    data = pickle.load(sys.stdin)3282    sys.stdin.close()32833284    # Устанавливает некоторые параметры.3285    _logger.setLevel(data['dist_log_level'])3286    forking.prepare(data)32873288    # Создаёт сервер для объекта `HostManager`.3289    server = managers.Server(HostManager._registry, ('', 0), data['authkey'])3290    current_process()._server = server32913292    # Сообщает родительскому процессу адрес сервера и количество процессоров.3293    conn = connection.Client(data['parent_address'], authkey=data['authkey'])3294    conn.send((data['index'], server.address, slot_count))3295    conn.close()32963297    # Устанавливает имя и т.д.3298    current_process().set_name('Host-%s:%s' % server.address)3299    util._run_after_forkers()33003301    # Регистрирует функцию очистки.3302    def cleanup(directory):3303        debug('removing directory %s', directory)3304        shutil.rmtree(directory)3305        debug('shutting down host manager')3306    util.Finalize(None, cleanup, args=[data['dir']], exitpriority=0)33073308    # Запускает менеджер хостов.3309    debug('remote host manager starting in %s', data['dir'])3310    server.serve_forever()3311```3312