Модуль multiprocessing в Python, параллельная обработка без GIL.

Параллельная обработка данных на разных ядрах процессора в Python.

Пакет multiprocessing поддерживает порождение процессов с использованием API, аналогичного модулю threading .

Модуль многопроцессорной обработки данных предлагает как локальную, так и удаленную параллельную обработку данных, эффективно обходя GIL (глобальную блокировку интерпретатора) и используя ядра процессора вместо потоков. Благодаря этому, этот модуль позволяет программисту полностью использовать несколько процессоров на данной машине. Он работает как под Unix, так и под Windows.

Содержание:

Сходство и различия API модулей multiprocessing и threading .

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

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

Аналогичный пример с использованием API, аналогичного модулю threading :

Контексты и методы запуска процессов на разных ядрах.

В зависимости от платформы модуль multiprocessing поддерживает три способа запуска процесса.

spawn :

Родительский процесс запускает новый процесс интерпретатора Python. Дочерний процесс унаследует только те ресурсы, которые необходимы для запуска метода Process.run() объекта multiprocessing.Process . В частности, ненужные файловые дескрипторы и дескрипторы родительского процесса не будут унаследованы. Запуск процесса с использованием этого метода довольно медленный по сравнению с использованием fork или forkserver .

Изменено в Python 3.8: В macOS метод запуска spawn теперь используется по умолчанию. Метод запуска fork следует считать небезопасным, так как он может привести к сбоям подпроцесса.

Доступно в Unix и Windows. По умолчанию в Windows и macOS.

Родительский процесс использует os.fork() для разветвления интерпретатора Python. Дочерний процесс, когда он начинается, фактически идентичен родительскому процессу. Все ресурсы родительского процесса наследуются дочерним процессом. Обратите внимание, что безопасное разветвление многопоточного процесса проблематично.

Доступно только в Unix. По умолчанию в Unix.

forkserver :

Когда программа запускается и выбирает метод запуска forkserver , запускается процесс сервера. С этого момента всякий раз, когда программе требуется новый процесс, родительский процесс подключается к серверу и запрашивает его разветвление для нового процесса. Процесс сервера является однопоточным, поэтому использование os.fork() безопасно. Никакие ненужные ресурсы не наследуются.

Доступно на платформах Unix, которые поддерживают передачу дескрипторов файлов по каналам Unix.

В Unix использование методов запуска spawn или forkserver также запускает процесс отслеживания ресурсов, который отслеживает несвязанные именованные системные ресурсы (такие как именованные семафоры или объекты разделяемой памяти), созданные процессами программы. Когда все процессы завершены, трекер ресурсов отсоединяет все оставшиеся отслеживаемые объекты. Обычно их не должно быть, но если процесс был остановлен сигналом, могут быть «утечки» ресурсов. Ни семафоры, ни сегменты разделяемой памяти не будут автоматически разъединены до следующей перезагрузки. Это проблематично для обоих объектов, поскольку система допускает только ограниченное количество именованных семафоров, а сегменты разделяемой памяти занимают некоторое пространство в основной памяти.

Чтобы выбрать метод запуска, используете функцию модуля multiprocessing.set_start_method() в предложении if __name__ == ‘__main__’ основного модуля. Функция multiprocessing.set_start_method() не должна использоваться в программе более одного раза.

В качестве альтернативы можно использовать функцию multiprocessing.get_context() для получения объекта контекста. Объекты контекста имеют тот же API, что и модуль multiprocessing , и позволяют использовать несколько методов запуска в одной программе.

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

Библиотека, которая хочет использовать определенный метод запуска, вероятно, должна использовать get_context() , чтобы не мешать выбору пользователя библиотеки.

Предупреждение В настоящее время методы запуска spawn и forkserver не могут использоваться с «замороженными» исполняемыми файлами. То есть с двоичными файлами, созданными такими пакетами, как pyInstaller и cx_Freeze в Unix. Метод запуска fork работает с такими файлами нормально.

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

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

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

— Очереди Queues , в собственной реализации.

Класс multiprocessing.Queue является почти клоном класса queue.Queue . Очереди безопасны для потоков в разных ядрах процессора.

— Каналы Pipes .

Класс multiprocessing.Pipe() возвращает пару объектов, соединенных каналом, которые по умолчанию является duplex двусторонним.

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

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

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

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

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

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

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

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

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

— Использование общей памяти Shared memory.

Данные могут быть сохранены на карте общей памяти с помощью multiprocessing.Value или multiprocessing.Array .

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

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

— Использование серверного процесса Server process.

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

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

Главные принципы программирования для модуля multiprocessing .

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

Следующие принципы относится ко всем методам запуска.

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

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

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

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

Лучше наследовать, чем pickle / unpickle .
При использовании методов запуска spawn или forkserver многие типы из multiprocessing должны быть упакованы модулем pickle , чтобы дочерние процессы могли их использовать. Обычно следует избегать отправки общих объектов другим процессам с использованием каналов или очередей.

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

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

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

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

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

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

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

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

следует переписать как:

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

Следующие принципы относится к методы запуска spawn и forkserver .

Есть несколько дополнительных ограничений, которые не применяются к методу запуска fork .

Больше picklability .
Убедитесь, что все аргументы конструктора Process.__init__() являются picklable . Кроме того, если создается подкласс multiprocessing.Process() , необходимо убедится, что экземпляры будут picklable при вызове метода Process.start() .

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

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

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

Например, при использовании метода запуска spawn или forkserver , выполняющего следующий модуль, произойдет сбой с исключением RuntimeError :

Вместо этого следует защитить точку входа программы, используя if __name__ == ‘__main__’ :

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

Это позволяет вновь созданному интерпретатору Python безопасно импортировать модуль и затем запускать функцию модуля worker() .

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

Модуль multiprocessing на примерах

Модуль multiprocessing был добавлен в Python версии 2.6. Изначально он был определен в PEP 371 Джесси Ноллером и Ричардом Одкерком. Модуль multiprocessing позволяет вам создавать процессы таким же образом, как при создании потоков при помощи модуля threading. Суть в том, что, в связи с тем, что мы теперь создаем процессы, вы можете обойти GIL (Global Interpreter Lock) и воспользоваться возможностью использования нескольких процессоров на компьютере. Пакет multiprocessing также включает ряд API, которых вообще нет в модуле threading. Например, есть очень удобный класс Pool, который вы можете использовать для параллельного выполнения функции между несколькими входами. Мы рассмотрим Pool немного позже. Мы начнем с класса Process модуля multiprocessing.

Приступим к работе с Multiprocessing

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

Для этого примера мы импортируем Process и создаем функцию doubler. Внутри функции, мы дублируем число, которое мы ей передали. Мы также используем модуль os, чтобы получить ID нынешнего процесса. Это скажет нам, какой именно процесс вызывает функцию. Далее, в нижнем блоке кода, мы создаем несколько Процессов и начинаем их.

Самый последний цикл только вызывает метод join() для каждого из процессов, что говорит Python подождать, пока процесс завершится. Если вам нужно остановить процесс, вы можете вызвать метод terminate(). Когда вы запустите этот код, вы получите выдачу, на подобие этой:

Все же иногда приятно иметь читабельное название процессов. К счастью, класс Process дает возможность вам получить доступ к названию вашего процесса. Давайте посмотрим:

На этот раз мы импортируем кое-что дополнительно: current_process. Это примерно то же самое, что и current_thread модуля threading. Мы используем его для того, чтобы получить имя потока, который вызывает нашу функцию. Обратите внимание на то, что мы не указывали название первых пяти процессов. И только шестой мы назвали Test. Давайте посмотрим, какую выдачу мы получим:

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

Замки (Locks)

Модуль multiprocessing поддерживает замки так же, как и модуль threading. Все что вам нужно, это импортировать Lock, повесить его, сделать что-нибудь и снять его. Давайте посмотрим:

Здесь мы создали простую функцию вывода, которая выводит все, что вы ей передаете. Чтобы не дать процессам конфликтовать друг с другом, мы используем объект Lock. Этот код зациклится над нашим списком трех объектов и создаст процесс для каждого из них. Каждый процесс будет вызывать нашу функцию, и передавать её одному из объектов. Так как мы используем замки, следующий процесс в строке будет ждать, пока замок не снимается, после чего он сможет продолжить.

Логирование (Logging)

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

Простейший способ вести журнал, это отправить все на stderr. Мы можем сделать это, вызвав функцию log_to_stderr(). Далее мы вызываем функцию get_logger для получения доступа к логгеру и настраиваем его уровень логгинга на INFO. Остальная часть кода остается такой же, какой и была. Обратите внимание на то, что я не вызываю метод join() здесь. Вместо этого, поток parent (другими словами, ваш скрипт) вызовет join() лично. Когда вы сделаете это, вы получите что-то на подобие:

Давайте пойдем дальше, и рассмотрим класс Pool поближе

Класс Pool

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

Здесь мы создали экземпляр Pool и указали ему создать три рабочих процесса. Далее мы используем метод map для отображения функции для каждого процесса. Наконец мы выводим результат, что в нашем случае является списком: [10, 20, 40]. Вы также можете получить результат вашего процесса в пуле, используя метод apply_async:

Так мы можем запросить результат процесса. В этом суть работы функции get. Она пытается получить наши результаты. Обратите внимание на то, что мы также настроили обратный отсчет, на тот случай, если что-нибудь произойдет с вызываемой нами функцией. Мы не хотим, чтобы она была заблокирована.

Связь между процессами

Когда речь заходит о связи между процессами, модули нашего multiprocessing включают в себя два главных метода: Queue и Pipe. Работа Queue защищена как от процессов, так и от потоков. Давайте взглянем на достаточно простой пример:

Здесь нам только и нужно, что импортировать Process и Queue. Далее мы создаем две функции, одна для создания данных и добавления их в очередь, и вторая для использования данных и обработки их. Добавление данных в Queue выполняется при помощи метода put(), в то время как получение данных из Queue выполняется через метод get. Последний кусок кода только создает объект Queue и несколько экземпляров Process, после чего возвращает их. Обратите внимание на то, что мы вызываем join() в наших объектах process больше, чем Queue.

Подведем итоги

Здесь мы прошли через достаточно большое количество материала. Вы узнали много чего нового о модуле multiprocessing для направления обычных функций, связи между процессами при помощи Queue, наименований потоков и многого другого. Разумеется, в документации Python предоставлено намного больше развернутой информации, которую я даже не начинал затрагивать в данной статье, так что настоятельно рекомендую с ней ознакомиться. Тем не менее, вы все-таки узнали много чего о том, как усилить мощность обработки вашего компьютера при помощи Python!