Как использовать несколько ядер процессора python

Как использовать все процессоры при записи данных в файл используя python?

У меня есть функция код, который по определенной логике записывает данные в Базу данных. Я считываю данные с 2-х файлов построчно и на основании этих данных создаю запись в Базе Данных. Как БД использую ESRI geodatabase. Но, проблема с том, что для работы мы используем 1 ядро и тратим очень много времени на выполнения этого кода. У меня же 96 ядер и я хочу сэкономить время при использовании всех ядер. Это можно сделать через распаралеливание процессов. Вот мой код: Создаю БД и таблицу в ней:

 arcpy.CreateFileGDB_management("C:\Users\ivan\CellRebell\ESRI_New_Zeland", "%s.gdb"%Island) result = arcpy.management.CreateFeatureclass( "C:\Users\ivan\CellRebell\ESRI_New_Zeland/%s.gdb"%Island, "esri_square", "POLYGON", spatial_reference=4326) feature_class = result[0] arcpy.AddField_management(feature_class, 'ID', 'TEXT') 

Записываю данные в таблицу:

with arcpy.da.InsertCursor(feature_class, ['ID','SHAPE@']) as cursor: with open('%s_long_double.txt'%Island, 'r') as long: for i in long: i_1, i_2 = i.split() with open('%s_short_double.txt'%Island, 'r') as short: for k in short: k_1, k_2 = k.split() coord = [[float(i_1),float(k_1)],[float(i_2),float(k_1)],[float(i_2),float(k_2)],[float(i_1),float(k_2)]] count+=1 row = ['%s_%s'%(Island, count), coord] cursor.insertRow(row) del cursor 

Как использовать все ядра для записи данных в файл? Скрипт написан на Python 2.7. Основная задача, которую я хочу решить — это ускорить процесс записи данных в таблицу через использования всех процессоров. Спасибо

Как работать с процессами и потоками в Python

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

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

Давайте возьмем за отправную точку ситуацию, когда у нас есть приложение, которое работает по стандартной схеме клиент – сервер:

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

Как мы можем помочь пользователю продолжить взаимодействие с нашим приложением, пока формируется отчет? Мы можем создать отдельный процесс, отдельный поток, и выполнять код асинхронно.

Рассмотрим каждое понятие отдельно.

Процессы

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

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

Небольшой пример создания процессов:

from multiprocessing import Process def print_word(word): print('hello,', word) if __name__ == '__main__': p1 = Process(target=print_word, args=('bob',), daemon=True) p2 = Process(target=print_word, args=('alice',), daemon=True) p1.start() p2.start() p1.join() p2.join()

Процессы представлены как экземпляр класса Process из встроенной библиотеки multiprocessing.

У нас есть функция, которая принимает 1 параметр и печатает приветствие с переданным параметром. Внутри конструкции if мы создаем два процесса p1 и p2 в качестве параметров, то есть мы передаем:

target – с названием выполняемой функции,

args – параметры для функции, которую мы будем вызывать,

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

Для того чтобы процесс стартовал, мы вызываем у каждого метод .start().

Но ниже мы вызываем еще и метод .join().

Для чего нужен join() и что такое daemon? Или основные и фоновые процессы

У нас есть основной (главный) процесс, который содержит весь код нашей программы, и два дополнительных (фоновых) p1, p2. Их мы создаем, когда мы прописываем параметр daemon=True. Так мы как раз и указываем, что эти два процесса будут второстепенными. Если мы не вызовем метод join у фонового процесса, то наша программа завершит свое выполнение, не дожидаясь выполнения p1 и p2.

Немного теории о процессах

Процессы не могут работать параллельно на одноядерной машине.

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

Познакомимся с еще одним термином — вытесняющая многозадачность.

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

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

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

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

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

Потоки

Первое, о чем хотим сказать про потоки — интерфейсы работы с процессами и потоками в Python очень похожи.

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

GIL (Global interpreter lock) следит за тем, чтобы в один момент времени работал лишь один поток. Механизм похож на то, как процессы конкурируют за ядро. Но в отличие от процессов GIL освобождается при вызове блокирующей функции операций ввода/вывода. Другой механизм его освобождения – time.sleep(). Об этом позже.

import threading def greet(name): print('hello: ', name) if __name__ == '__main__': t1 = threading.Thread(target=greet, args=('bob',), daemon=True) t2 = threading.Thread(target=greet, args=('alice',), daemon=True) t1.start() t2.start() t1.join() t2.join()

Как видно, процесс создания потоков идентичен алгоритму формирования процессов.

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

Первая проблема – Race Condition или состояние гонки.

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

Слева вы видите первый запуск программы. Первый поток берет значение из глобальной переменной x, прибавляет 1 и записывает в x результат = 3. Затем второй поток начинает работу. Он берет из переменной x значение 3, умножает на 2 и записывает результат = 6.

На правой схеме – второй запуск программы, где сперва в работу вступает поток 2, он выполняет те же операции, берет x = 2, умножает на 2 и фиксирует результат 4. Затем вступает поток 1, читает 4 из x, увеличивает на единицу и записывает 5.

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

Но давайте посмотрим на такой поток выполнения:

Поток 1 вступает в работу, читает переменную x и переключает контекст на поток 2 (context switch). Затем поток 2 берет значение из x = 2, умножает на 2 и записывает в x = 4. Процессор переключает контекст на поток 1, а в потоке 1, как мы помним, сохранено значение x = 2. В итоге он увеличивает значение на единицу и записывает в x = 3, а значит, на выходе мы получаем 3.

Один поток обогнал другой при переключении контекста, и мы получили непредсказуемый результат. Такое событие называется Race condition. Как тогда быть уверенным в том, что поток, взявший в работу какие-то данные, выполнит свою работу, перед тем как переключит свой контекст на другой потоку?

``` from threading import Thread from time import sleep counter = 0 def increase(by): global counter local_counter = counter local_counter += by sleep(0.1) counter = local_counter print(f'') t1 = Thread(target=increase, args=(10,)) t2 = Thread(target=increase, args=(20,)) t1.start() t2.start() t1.join() t2.join() ```

Посмотрим на результат:

``` counter=10 counter=20 ```

Вместо 30 получаем 20.

На помощь нам может прийти такое понятие как Lock.

Lock (замок) – объект, который захватывает поток, и пока поток не освободит (release) Lock, другие потоки не смогут ничего сделать с этими данными, захваченными при помощи замка.

``` from threading import Thread, Lock from time import sleep counter = 0 def increase(by, lock: Lock): global counter lock.acquire() local_counter = counter local_counter += by sleep(0.1) counter = local_counter print(f'') lock.release() lock = Lock() t1 = Thread(target=increase, args=(10, lock,)) t2 = Thread(target=increase, args=(20, lock,)) t1.start() t2.start() t1.join() t2.join() ```

Вот теперь как и должно быть:

``` counter=10 counter=30 ```

Несмотря на то, что Lock помогает решить проблему с Race condition, он может привести к другой сложной ситуации, когда один поток ждет освобождение одного замка, а другой ждет освобождение от первого. Такое ожидание приводит к ситуации взаимного тупика, известного как Deadlock.

``` from threading import Thread, Lock from time import sleep a = 5 b = 10 a_lock = Lock() b_lock = Lock() def function_a(): global a global b a_lock.acquire() print('Функция a, a_lock = заблокирован') sleep(1) b_lock.acquire() print('Функция a, b_lock = заблокирован') sleep(1) a_lock.release() print('Функция a, a_lock = разблокирован') b_lock.release() print('Функция a, b_lock = разблокирован') def function_b(): global a global b b_lock.acquire() print('Функция b, b_lock = заблокирован') a_lock.acquire() print('Функция b, a_lock = заблокирован') sleep(1) b_lock.release() print('Функция b, b_lock = разблокирован') a_lock.release() print('Функция b, a_lock = разблокирован') t1 = Thread(target=function_a) t2 = Thread(target=function_b) t1.start() t2.start() t1.join() t2.join() print('Готово') ```

И теперь посмотрим результат:

``` Функция a, a_lock = заблокирован Функция b, b_lock = заблокирован ```

Наша программа зависает в ожидании разблокировки, которая никогда не произойдет. Так же Deadlock произойдет при попытке заблокировать наш Lock повторно в том же потоке.

Решить проблему с Deadlock могут помочь различные механизмы синхронизации потоков. Разберем один из таких примеров – Semaphore (Семафор).

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

Посмотрим Semaphore на примере реализации очереди из реального кейса.

``` import datetime from threading import Semaphore, Thread from time import sleep s = Semaphore(3) def semaphore_func(payload: int): s.acquire() now = datetime.datetime.now().strftime('%H:%M:%S') print(f', ') sleep(2) s.release() threads = [Thread(target=semaphore_func, args=(i,)) for i in range(7)] for t in threads: t.start() for t in threads: t.join() ``` 

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

``` now='00:49:51', payload=0 now='00:49:51', payload=1 now='00:49:51', payload=2 now='00:49:53', payload=3 now='00:49:53', payload=5 now='00:49:53', payload=4 now='00:49:55', payload=6 ```

Это удобно использовать, например, в таком виде: если база данных может держать не более 30 соединений, то инстанциируем Semaphore со значением 30. Блокируем, когда поднимаем соединение и разблокируем, когда освобождаем.

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

Теперь поговорим об освобождении GIL.

CPython управляет памятью с помощью подсчета ссылок. То есть для каждого объекта Python подсчитывается, сколько на него указывается ссылок с других объектов, использующих его в данный момент. При добавлении ссылки счетчик увеличивается, при удалении ссылки счетчик уменьшается. А когда счетчик ссылок становится 0 — это означает, что объект больше не нужен, и его можно удалить из памяти.

Следовательно, если не будет GIL, который запрещает Python процессу выполнять более одной команды байт-кода в каждый момент времени, то при подсчете ссылок может случиться Race-condition, с подсчетом ссылок на объекты, как это было в примере с переменными выше.

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

Стало быть, если мы сделаем библиотеку, даже с CPU-bound нагрузкой, где мы не взаимодействуем с объектами Python (словарями, списками, целыми числами и т. д.) или большая часть библиотеки не взаимодействует, то мы можем освободить GIL. Например, библиотеки hashlib и NumPy выполняют расчеты на чистом C и освобождают GIL.

time.sleep() — реализация этой функции освобождает GIL и выполняется на уровне системы и работает вне кода Python.

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

Асинхронность

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

Кооперативная многозадачность — это тип многозадачности, при котором фоновые задачи выполняются только во время простоя основного процесса и только в том случае, если на это получено разрешение основного процесса.

То есть время, когда исполняемая программа управляет передачей управления другому процессу и передачей процессорного времени.

Недостатком такого исполнения является то, что если одна задача зависла. Зависает вся система.

А вот преимущества такого решения: разработчик программы отдает управление тогда, когда он посчитает это нужным.

Теперь мы подобрались к понятию асинхронного программирования.

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

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

В реальном программировании, а особенно в web-разработке мы очень часто чего-то ждём и не делаем полезной работы. Вот несколько примеров:

• Отправили запрос на сторонний ресурс и ждем ответа.
• Отправили запрос в базу данных и ждем результата запроса.
• Читаем или записываем файл на диск.
• И так далее.

Получается что мы ждем, ждем и ждем. А в это время наша программа могла бы выполнить множество полезной нагрузки. И мы как разработчики ПО точно знаем, где мы будем ожидать. Ничего не напоминает? Да! Похоже на кооперативную многозадачность, но только не на уровне операционной системы, а на уровне процесса.

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

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

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

• Реализация на основе коллбэков.
• Реализация на основе корутин.

Оба подхода имеют место. Например, мощный фреймворк TORNADO реализован именно на основе коллбэков.

У этого подхода есть ряд недостатков:

• Код перестает выглядеть как синхронный, что усложняет отладку.
• Ад коллбэков, в котором будет сложно разобраться. Просто погуглите фразу “callback hell”.

Если после этих минусов желание попробовать ещё осталось, то можно в подходе легко разобраться.

А вот подход на основе корутин мы разберем более глубоко. У него также есть ряд преимуществ и недостатков:

Плюсы:

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

Минусы:

• Чуть более сложный подход для понимания.

В Python есть ряд библиотек, которые позволяют работать с асинхронностью:

• asyncio — основная библиотека для работы с асинхронным программированием,
• aiohttp — для асинхронной работы с запросами,
• aiofiles — для работы с файловой системой.

Как вы наверное заметили, у библиотек есть префикс aio (asynchronous input output, асинхронный ввод-вывод). Тут как раз решается проблема ожидания. Такие задачи называют IO bound.

Рассмотрим термины, которые нам помогут во всём разобраться.

Event loop (цикл событий) — ядро каждого приложения asyncio. Циклы событий запускают асинхронные задачи и обратные вызовы, выполняют операции сетевого ввода-вывода и запускают подпроцессы. Официальную документацию можно прочесть тут.

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

Футуры — это определение обычно воспринимается тяжелее всего, но я постараюсь объяснить как можно проще. Это объект, в котором хранится результат и состояние задачи:

То есть в процессе работы мы можем управлять задачами в зависимости от футуры (статус/результат) задачи.

Корутины могут быть реализованы с использованием генераторов или async/await. Мы выбираем второй вариант как более лаконичный.

Посмотрим, как это выглядит в коде.

Создадим первую корутину:

``` import asyncio async def hello(): print('Запуск функции hello') await asyncio.sleep(5) # Отдаем управление обратно в Event loop пока ждем print('Переключение контекста в функцию hello') ```

Теперь у нас есть асинхронная функция. Научимся теперь её запускать. Первое, что хочется сделать — вызвать её как обычную функцию. Давайте попробуем:

``` import asyncio async def hello(): print('Запуск функции hello') await asyncio.sleep(5) # Отдаем управление обратно в Event loop пока ждём print('Переключение контекста в функцию hello') hello() ```

При выполнении ничего не произошло. А вот наш друг интерпретатор выдал предупреждение.

``` RuntimeWarning: coroutine 'hello' was never awaited hello() RuntimeWarning: Enable tracemalloc to get the object allocation traceback ```

Тут из сообщения становится понятно, что при вызове таким образом асинхронной функции она превращается в асинхронную корутину.

Как же можно запустить корутину?

• Из другой корутины.
• Обернуть в задачу.
• Запустить через метод asyncio.run и run_until_complete из цикла событий.

``` import asyncio async def hello(): print('Запуск функции hello') await asyncio.sleep(5) # Отдаем управление обратно в Event loop пока ждём print('Переключение контекста в функцию hello') asyncio.run(hello()) ```

И получили результат, который ожидали.

``` Запуск функции hello Переключение контекста в функцию hello ```

Вызов метода asyncio.run(hello()) принимает корутину, которую необходимо выполнить, открывает цикл событий, выполняет корутину и закрывает цикл событий.

Что делать, если необходимо запустить две задачи конкурентно?

Это поможет нам сделать asyncio.gather, но раз функция asyncio.run принимает только одну корутину, создадим новую корутину, которая будет запускать конкурентно несколько задач.

``` import asyncio async def hello(): print('Запуск функции hello') await asyncio.sleep(5) # Отдаем управление обратно в Event loop пока ждём print('Переключение контекста в функцию hello') async def starter(): await asyncio.gather(hello(), hello()) asyncio.run(starter()) ```

И получаем тот результат, который ожидали.

``` Запуск функции hello Запуск функции hello Переключение контекста в функцию hello Переключение контекста в функцию hello ```

Время выполнения около 5 секунд. Если бы две функции выполнялись синхронно, то время выполнения составило около 10 секунд.

А если нам необходимо выполнить 10 тысяч раз, сколько времени это займёт? Видоизменяем код:

``` import asyncio import time start = time.time() ## точка отсчета времени async def hello(): print('Запуск функции hello') await asyncio.sleep(5) # Отдаем управление обратно в Event loop пока ждём print('Переключение контекста в функцию hello') async def starter(): await asyncio.gather(*[hello() for i in range(10000)]) asyncio.run(starter()) end = time.time() - start print(end) ```

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

``` … Переключение контекста в функцию hello Переключение контекста в функцию hello Переключение контекста в функцию hello 5.27926778793335 ```

Неплохо. Чуть больше тех же самых 5 секунд.

Что же это значит? Представьте, что запрос на сторонний сайт занимает порядка 5 секунд. И нам необходимо получить результат тех же самых 10000 запросов. Используя асинхронное программирование, 10 тысяч запросов сеть будут выполняться чуть больше 5 секунд. Правда, здорово?

Но мы пойдем дальше и будем уже более гибко и детально работать с асинхронным выполнением:

``` import asyncio async def hello(): print('Запуск функции hello') await asyncio.sleep(5) # Отдаем управление обратно в Event loop пока ждём print('Переключение контекста в функцию hello') async def bye(): print('Запуск функции bye') await asyncio.sleep(5) # Отдаем управление обратно в Event loop пока ждём print('Переключение контекста в функцию byе') ioloop = asyncio.get_event_loop() tasks = [ioloop.create_task(hello()), ioloop.create_task(bye())] tasks_for_wait = asyncio.wait(tasks) ioloop.run_until_complete(tasks_for_wait) ioloop.close() ```

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

Теперь давайте попробуем управлять выполнениями задач и рассмотрим код ниже:

``` import asyncio async def hello(): print('Запуск функции hello') await asyncio.sleep(5) # Отдаем управление обратно в Event loop пока ждём print('Переключение контекста в функцию hello') return 'Выполнена функция hello' async def bye(): print('Запуск функции bye') await asyncio.sleep(2) # Отдаем управление обратно в Event loop пока ждём print('Переключение контекста в функцию byе') return 'Выполнена функция bye' async def starter(ioloop): tasks = [ioloop.create_task(hello()), ioloop.create_task(bye())] done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) result = done.pop().result() for pending_future in pending: pending_future.cancel() print(result) ioloop = asyncio.get_event_loop() ioloop.run_until_complete(starter(ioloop)) ioloop.close() ```

Результат будет таким:

``` Запуск функции hello Запуск функции bye Переключение контекста в функцию byе Выполнена функция bye ```

Теперь только представьте, какие возможности у нас открылись! Например, мы можем запрашивать курсы валют сразу с нескольких ресурсов, и принимать результат того, который быстрее ответит. Чувствуете, как растет скорость и устойчивость приложения?

Или ещё такой пример. Мы можем динамически добавлять новые задачи, когда одна из задач выполнена. Например, парсить сайт в 20 задач. Только в этом случае добавляем к футурам в статусе pending новую задачу.

А самое приятное — наши асинхронные задачи выглядят как синхронные:

• Работая в один поток, можно делать больше работы;
• Удобная отладка;
• Нет проблем с блокировками;
• Можем использовать обратные вызовы (коллбэки) и отложенные обратные вызовы вдобавок к нашему асинхронному коду. Для этого посмотрите на методы цикла событий call_soon, call_later, call_at.

Для работы с конкурентностью есть различные библиотеки, которые решают самые востребованные задачи IO:

• aiohttp — работа с HTTP запросами;
• aiofiles — работа с файлами.

Мы рассмотрели темы асинхронного и параллельного программирования. Теперь осталось дело за малым, опробовать всё это на практике.

Итого

Отдельные процессы

Плюсы:

+ Используют все ресурсы ядра процессора.

+ Можно загрузить все ядра процессора.

+ Независимые системные процессы.

+ Подходит для CPU bound операций.

Минусы:

• Если необходимо использовать общую память, то необходимо синхронизировать, так как нет общих переменных.
• Требуют больших ресурсов, так как запускают отдельный интерпретатор.

Используем там, где обрабатываемые данные не зависят от других процессов и данных. Например:

+ Расчет нейронных сетей.

+ Обработка изолированных фотографий.

+ Архивирование изолированных файлов.

+ Конвертация форматов файлов.

Отдельные потоки

Плюсы:

+ Используют немного памяти.

Минусы:

• Одновременный доступ к памяти может приводить к конфликтам.
• Сложный код.

Используем там, где код много раз ожидает, пока выполнится задача. Например:

Асинхронность

Плюсы:

+ Работает в одном процессе и в одном потоке.

+ Экономное использование памяти.

+ Подходит для I/O bound операций.

Минусы:

• Сложность отладки.
• CPU bound операции блокируют все задачи.

Используем там, где код много раз ожидает. Например:

+ Работа с файловой системой.

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

Спасибо за внимание! Надеемся, что этот материал был полезен для вас.

Авторские материалы для разработчиков мы также публикуем в наших соцсетях – ВК и Telegram.

• python
• многопоточность
• многопоточное программирование
• асинхронность
• асинхронное программирование

• Блог компании SimbirSoft
• Python
• Программирование
• Параллельное программирование

Многопоточность в Python. Библиотеки threading и multiprocessing.

Процесс — исполняемый экземпляр какой-либо программы. Каждый процесс состоит из следующих элементов:

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

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

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

При работе с потоками стоит учесть несколько моментов:

• одно ядро процессора в один момент может исполнять только один поток;
• потоки одного процесса могут исполняться физически одновременно (на разных ядрах);
• бессмысленно порождать потоков больше, чем у вас есть ядер.

Потоки имеют несколько применений. Первое — ускорение работы программы. Ускорение достигается за счет параллельного выполнения независимых друг от друга вычислений. Например, при численном интегрировании область интегрирования может быть разбита на 3 участка. На каждый участок создается свой поток, в котором численно вычислется интеграл для конкретного участка. Второе — независимое исполнение операций. Отличие этого случая от первого хорошо видно на следующем примере. Пусть есть приложение с графическим интерфейсом, где весь код выполняется в одном потоке. При выполнении какой-нибудь долгой операции (например, копирование файла) интерфейс приложения просто перестанет отвечать до тех пор, пока долгий процесс не завершится. В таком случае в один поток помещается работа графического интерфейса, в другой — остальные вычисления. В таком случае интерфейс позволит проводить другие операции даже во время выполнения долгой операции в другом потоке (например, заполнение прогресс бара в процессе копирования файла).

threading

В Python работа с потоками осуществляется при помощи стандартной библиотеки threading. В библиотеке представлен класс Thread для создания потока выполнения. Задание исполняемого кода в отдельном потоке возможно двумя способами:

• передача исполняемого объекта (функции) в конструктор класса;
• переопределение функции run() в классе-наследнике.

После того, как объект создан, поток запускается путем вызова метода start(). Рассмотрим простой пример:

import threading import sys def thread_job(number): print('Hello <>'.format(number)) sys.stdout.flush() def run_threads(count): threads = [ threading.Thread(target=thread_job, args=(i,)) for i in range(0, count) ] for thread in threads: thread.start() # каждый поток должен быть запущен for thread in threads: thread.join() # дожидаемся исполнения всех потоков run_threads(4) print(finish) 

Конструктор класса Thread имеет следующие аргументы:

• group должно быть None; зарезервировано для будующих реализаций Python 3;
• target является исполняемым объектом (по умолчанию равен None, ничего не исполняется);
• name обозначет имя потока (по умолчанию имя генерируется автоматически);
• args — кортеж аргументов для исполняемого объекта;
• kwargs — словарь именованных аргументов для исполняемого объекта;
• daemon равное True обозначет служебный поток (служебные потоки завершаются принудительно при завершении процесса); по умолчанию False.

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

import threading import sys import time def thread_job(number): time.sleep(2) # "усыпляем" поток на 2 сек print('Hello <>'.format(number)) sys.stdout.flush() def run_threads(count): threads = [ threading.Thread(target=thread_job, args=(i,)) for i in range(1, count) ] for thread in threads: thread.start() # каждый поток должен быть запущен run_threads(1) print(finish) 

Как можно увидеть, программа завершается без ошибок (с кодом 0), но теперь строка «finish» печатается раньше строки «Hello 0», т.к. главный поток теперь не ждет завершения работы других потоков. Метод join() используется для блокирования исполнения родительского потока до тех пор, пока созданный поток не завершится. Это нужно в случаях, когда для работы потока-родителя необходим результат работы потока-потомка. Вспомним пример с численным интегрированием. Вычисление итогового значения интеграла выполняется в главном потоке, но это возможно только после завершения вычислений в побочных потоках. В таком случае главный поток нужно просто приостановить до тех пор, пока не завершатся все побочные потоки. Метод join() может принимать один аргумент — таймаут в секундах. Если таймаут задан, join() бликирует работу на указанное время. Если по истечении времени ожидаемый поток не будет завершен, join() все равно разблокирует работу потока, вызвашего его. Проверить, исполняется ли поток можно методом is_alive(). Подробнее ознакомиться с функционалом библиотеки можно в официальной документации по threading.

Упражнение №1

Запустите следующий код. В чем проблема данного кода? Всегда ли counter = 10 после исполнения кода программы?

import threading import sys def thread_job(): global counter old_counter = counter counter = old_counter + 1 print('<> '.format(counter), end='') sys.stdout.flush() counter = 0 threads = [threading.Thread(target=thread_job) for _ in range(10)] for thread in threads: thread.start() for thread in threads: thread.join() print(counter) 

Демонстрация «проблемности» кода:

import threading import random import time import sys def thread_job(): global counter old_counter = counter time.sleep(random.randint(0, 1)) counter = old_counter + 1 print('<> '.format(counter), end='') sys.stdout.flush() counter = 0 threads = [threading.Thread(target=thread_job) for _ in range(10)] for thread in threads: thread.start() for thread in threads: thread.join() print(counter) 

Почему так происходит? Есть несколько возможных решений этой проблемы.

import threading import random import time import sys def thread_job(): lock.acquire() # mutex global counter old_counter = counter time.sleep(random.randint(0, 1)) counter = old_counter + 1 print('<> '.format(counter), end='') sys.stdout.flush() lock.release() lock = threading.Lock() counter = 0 threads = [threading.Thread(target=thread_job) for _ in range(10)] for thread in threads: thread.start() for thread in threads: thread.join() print(counter) 

import threading import random import time import sys def thread_job(): with lock: global counter old_counter = counter time.sleep(random.randint(0, 1)) counter = old_counter + 1 print('<> '.format(counter), end='') sys.stdout.flush() lock = threading.Lock() counter = 0 threads = [threading.Thread(target=thread_job) for _ in range(10)] for thread in threads: thread.start() for thread in threads: thread.join() print(counter) 

Вариант с контекстным менеджером более предпочтителен. Вспомните работу с файлами при помощи with. По завершении with файл автоматически закрывался. В данном случае похожая ситуация. Для того, чтобы запретить нескольким потокам параллельно выполнять некоторые участки кода, мы используем Lock (в UNIX системах более известен как мьютекс (mutex)). Мьютекс может быть в двух состояниях: свободен и заблокирован. Если какой-либо поток пытается заблокировать уже заблокированный мьютекс, то поток блокируется до тех пор, пока мьютекс не освободится. Причем если несколько потоков претендует на блокирование мьютекса, то потоки просто выстраиваются в очередь. Главная проблема — не освобожденный мьютекс. Отсутствие строчки lock.release() может повесить остальные потоки в бесконечное ожидание. Контекстный менеджер позволит избежать этой проблемы. Как только он закончится, все захваченные им ресурсы будут освобождены, в том числе мьютекс.

Упражнение №2

Иногда бывает нужно узнать доступность набора ip адресов. Неэффективный вариант представлен ниже.

Реализуйте то же самое, но используя threading.

import os, re received_packages = re.compile(r"(\d) received") status = ("no response", "alive but losses", "alive") for suffix in range(20, 30): ip = "192.168.178." + str(suffix) ping_out = os.popen("ping -q -c2 " + ip, "r") # получение вердикта print(". pinging ", ip) while True: line = ping_out.readline() if not line: break n_received = received_packages.findall(line) if n_received: print(ip + ": " + status[int(n_received[0])]) 

Global Interpreter Lock (GIL)

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

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

Упражнение №3

Написать программу, которая будет находить сумму чисел массива с использованием N потоков. Запустить с разным параметром N. Убедиться, что несмотря на увеличение N, ускорения подсчета не происходит. Причина этому — GIL. В Python вычисления распараллеливать бессмысленно. Замерить время работы можно с помощью библиотеки time (ответ в секундах):

start = time.time() # код, время работы которого надо замерить print(time.time() - start) 

Упражнение №4

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

import urllib.request import time urls = [ 'https://www.yandex.ru', 'https://www.google.com', 'https://habrahabr.ru', 'https://www.python.org', 'https://isocpp.org', ] def read_url(url): with urllib.request.urlopen(url) as u: return u.read() start = time.time() for url in urls: read_url(url) print(time.time() - start) 

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

multiprocessing

Библиотека multiprocessing позволяет организовать параллелизм вычислений за счет создания подпроцессов. Т.к. каждый процесс выполняется независимо от других, этот метод параллелизма позволяет избежать проблем с GIL. Предоставляемый библиотекой API схож с тем, что есть в threading, хотя есть уникальные вещи. Создание процесса происходит поутем создания объекта класса Process. Аргументы конструктора аналогичны тем, что есть в конструкторе Thread. В том числе аргумент daemon позволяет создавать служебные процессы. Служебные процессы завершаются вместе с родительским процессом и не могут порождать свои подпроцессы.

Простой пример работы с библиотекой:

from multiprocessing import Process def f(name): print('hello', name) if __name__ == '__main__': p = Process(target=f, args=('bob',)) p.start() p.join() 

Чтобы убедить, что каждый процесс имеет свой ID, запустите пример:

from multiprocessing import Process import os def info(title): print(title) print('module name:', __name__) print('parent process:', os.getppid()) print('process id:', os.getpid()) def f(name): info('function f') print('hello', name) if __name__ == '__main__': info('main line') p = Process(target=f, args=('bob',)) p.start() p.join() 

Старайтесь не забывать про конструкцию __name__ == ‘__main__’ . Это надо для того, чтобы ваш модуль можно было безопасно подключать в другие модули и при этом не создавались новые процессы без вашего ведома.

Упражнение №5

Запустите код. Попробуйте объяснить, почему LIST — пуст.

import multiprocessing def worker(): LIST.append('item') LIST = [] if __name__ == "__main__": processes = [ multiprocessing.Process(target=worker) for _ in range(5) ] for p in processes: p.start() for p in processes: p.join() print(LIST) 

Общение между процессами

multiprocessing предоставляет два вида межпроцессного обмена данными: очереди и каналы данных (pipe).

Очереди (класс Queue) аналогичны структуре данных «очередь», рассмотренной вами в курсе алгоритмов.

from multiprocessing import Process, Queue def f(q): q.put([42, None, 'hello']) if __name__ == '__main__': q = Queue() p = Process(target=f, args=(q,)) p.start() print(q.get()) # выводит "[42, None, 'hello']" p.join() 

Класс Pipe отвечает за канал обмена данными (по умолчанию, двунаправленный), представленный двумя концами, объектами класса Connection. С одним концом канала работает родительский процесс, а с другим концом — подпроцесс.

from multiprocessing import Process, Pipe def f(conn): conn.send([42, None, 'hello']) conn.close() if __name__ == '__main__': parent_conn, child_conn = Pipe() p = Process(target=f, args=(child_conn,)) p.start() print(parent_conn.recv()) # выводит "[42, None, 'hello']" p.join() 

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

from multiprocessing import Process, Lock def f(l, i): l.acquire() try: print('hello world', i) finally: l.release() if __name__ == '__main__': lock = Lock() for num in range(10): Process(target=f, args=(lock, num)).start() 

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

Класс Pool в multiprocessing

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

Наиболее интересные функции: Pool.apply, Pool.map, Pool.apply_async, Pool.map_async.

apply, map работают аналогично питоновским built-in apply, map.

Как работает Pool можно понять на примере:

from multiprocessing import Pool def cube(x): return x**3 if __name__ == "__main__": pool = Pool(processes=4) # создаем пул из 4 процессов # в apply можно передать несколько аргументов results = [pool.apply(cube, args=(x,)) for x in range(1,7)] # раскидываем числа от 1 до 7 по 4 процессам print(results) pool = Pool(processes=4) # то же самое, но с map. разбивает итерируемый объект (range(1,7)) на chunks и раскидывает аргументы по процессам results = pool.map(cube, range(1,7)) print(results) 

map, apply — блокирующие вызовы. Главная программа будет заблокирована, пока процесс не выполнит работу.

map_async, apply_async — неблокирующие. При их вызове, они сразу возвращают управление в главную программу (возвращают ApplyResult как результат). Метод get() объекта ApplyResult блокирует основной поток, пока функция не будет выполнена.

pool = mp.Pool(processes=4) results = [pool.apply_async(cube, args=(x,)) for x in range(1,7)] output = [p.get() for p in results] print(output) 

Упражение №6*

Для этого упражнения скачайте архив viterbi_mp.zip с кодом и необходимыми данными.

Рассмотрим следующую задачу. Положение мобильного робота на двумерной карте может быть представлено тремя числами: x, y и направлением θ. Точное положение робота нам не известно. В связи с этим мы строим N гипотез о его пложении, сумма их вероятностей равна 1. В процессе движения робота некоторые гипотезы исчезали, а некоторые порождали новые. Однако в каждый момент времени количество гипотез — константа. Известно, какая гипотеза из какой была порождена.

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

Задача: необходимо восстановить траекторию движения робота. Есть несколько способов приближенно решить данную задачу. Один из способов — восстановить наиболее вероятную траекторию. Для этого воспользуемся алгоритмом Витерби, одним из алгоритмов динамического программирования.

Пусть у нас было T моментов времени. На каждом моменте времени t мы для каждой частицы, существующей в момент времени t, выбираем наиболее вероятный переход из какой-нибудь частицы с момента времени t-1. Тогда ответом будет — argmax по вероятности среди всех частиц в последний момент времени. Однако, сам алгоритм довольно медленный. Его асимптотика O(T * N^2) .

В архиве вам предоставлен код в файле generate_viterbi_trajectory.py . Однако, он написан без распараллеливания. Ваша задача — распараллелить код, используя multiprocessing. Файл graph.ldj представляет собой текстовый файл, где каждая строка в формате JSON. Каждая строка представляет собой один момент времени. В этом задании вам предлагаются первые 10 моментов времени движения робота. В каждый момент времени количество частиц N = 2000 . Файл localization_config.json — файл конфигурации, содержащий параметры с которыми происходила генерация графа. Файл true_trajectory.json содержит массив троек чисел (x, y, θ), построенный нераспараллеленым алгоритмом. Вам надо будет сравнить полученную вами траекторию с данной при помощи скрипта correspond_trajectories.py . Для тех, кто хочет попробовать свой код на больших данных, используйте файл full_graph.ldj , который содержит порядка 1700 строк. Архив с файлом.

Не забудьте замерить время работы. Примерное время работы на моем компьютере для 10 строк в 1 процесс — 300 сек.

Сайт построен с использованием Pelican. За основу оформления взята тема от Smashing Magazine. Исходные тексты программ, приведённые на этом сайте, распространяются под лицензией GPLv3, все остальные материалы сайта распространяются под лицензией CC-BY.

Способы реализации параллельных вычислений в программах на Python

Параллелизм дает возможность работать над несколькими вычислениями одновременно в одной программе. Такого поведения в Python можно добиться несколькими способами:

• Используя многопоточность threading , позволяя нескольким потокам работать по очереди.
• Используя несколько ядер процессора multiprocessing . Делать сразу несколько вычислений, используя несколько ядер процессора. Это и называется параллелизмом.
• Используя асинхронный ввод-вывод с модулем asyncio . Запуская какую то задачу, продолжать делать другие вычисления, вместо ожидания ответа от сетевого подключения или от операций чтения/записи.

Разница между потоками и процессами.

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

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

Асинхронный ввод-вывод не является ни потоковым ( threading ), ни многопроцессорным ( multiprocessing ). По сути, это однопоточная, однопроцессная парадигма и не относится к параллельным вычислениям.

У Python есть одна особенность, которая усложняет параллельное выполнение кода. Она называется GIL, сокращенно от Global Interpreter Lock. GIL гарантирует, что в любой момент времени работает только один поток. Из этого следует, что с потоками невозможно использовать несколько ядер процессора.

GIL был введен в Python потому, что управление памятью CPython не является потокобезопасным. Имея такую блокировку Python может быть уверен, что никогда не будет условий гонки.

Что такое условия гонки и потокобезопасность?

• Состояние гонки возникает, когда несколько потоков могут одновременно получать доступ к общей структуре данных или местоположению в памяти и изменять их, вследствии чего могут произойти непредсказуемые вещи. Пример из жизни: если два пользователя одновременно редактируют один и тот же документ онлайн и второй пользователь сохранит данные в базу, то перезапишет работу первого пользователя. Чтобы избежать условий гонки, необходимо заставить второго пользователя ждать, пока первый закончит работу с документом и только после этого разрешить второму пользователю открыть и начать редактировать документ.
• Потокобезопасность работает путем создания копии локального хранилища в каждом потоке, чтобы данные не сталкивались с другим потоком.

Алгоритм планирования доступа потоков к общим данным.

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

Например, есть общая переменная a = 2 . Теперь предположим, что есть два потока, thread_one и thread_two . Они выполняют следующие операции:

a = 2 # функция 1 потока def thread_one(): global a a = a + 2 # функция 2 потока def thread_two(): global a a = a * 3 

Если поток thread_one получит доступ к общей переменной a первым и thread_two вторым, то результат будет 12:

или наоборот, сначала запустится thread_two , а затем thread_one , то мы получим другой результат:

Таким образом очевидно, что порядок выполнения операций потоками имеет значение

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

Есть еще худший вариант развития событий, который может произойти без встроенной в Python блокировки потоков GIL . Например, если оба потока начинают читать глобальную переменную a одновременно, оба потока увидят, что a = 2 , а дальше, в зависимости от того какой поток произведет вычисления последним, в конечном итоге и будет равна переменная a (4 или 6). Не то, что ожидалось!

Исследование разных подходов к параллельным вычислениям в Python.

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

def heavy(n): for x in range(1, n): for y in range(1, n): x**y 

Функция heavy() представляет собой вложенный цикл, который выполняет возведение в степень. Это функция связана со скоростью ядра процессора производить математические вычисления. Если понаблюдать за операционной системой во время выполнения функции, то можно увидеть загрузку ЦП близкую к 100%.

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

Однопоточный режим работы.

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

import time def heavy(n): for x in range(1, n): for y in range(1, n): x**y def sequential(n): for i in range(n): heavy(500) print(f"n> циклов вычислений закончены") if __name__ == "__main__": start = time.time() sequential(80) end = time.time() print("Общее время работы: ", end - start) # 80 циклов вычислений закончены # Общее время работы: 23.573118925094604 

Использование потоков threading .

В следующем примере будем использовать несколько потоков для выполнения функции heavy() . Также произведем 80 циклов вычислений. Для этого разделим вычисления на 4 потока, в каждом из которых запустим 20 циклов:

import threading import time def heavy(n, i, thead): for x in range(1, n): for y in range(1, n): x**y print(f"Цикл № i>. Поток thead>") def sequential(calc, thead): print(f"Запускаем поток № thead>") for i in range(calc): heavy(500, i, thead) print(f"calc> циклов вычислений закончены. Поток № thead>") def threaded(theads, calc): # theads - количество потоков # calc - количество операций на поток threads = [] # делим вычисления на `theads` потоков for thead in range(theads): t = threading.Thread(target=sequential, args=(calc, thead)) threads.append(t) t.start() # Подождем, пока все потоки # завершат свою работу. for t in threads: t.join() if __name__ == "__main__": start = time.time() # разделим вычисления на 4 потока # в каждом из которых по 20 циклов threaded(4, 20) end = time.time() print("Общее время работы: ", end - start) # Показано часть вывода # . # . # . # Общее время работы: 43.33752250671387 

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

Если бы у Python не было GIL, то вычисления функции heavy() происходили быстрее, а общее время выполнения программы стремилось к времени выполнения однопоточной программы. Причина, по которой многопоточный режим в данном примере не будет работать быстрее однопоточного — это вычисления, связанные с процессором и заключаются в GIL!

Если бы функция heavy() имела много блокирующих операций, таких как сетевые вызовы или операции с файловой системой, то применение многопоточного режима работы было бы оправдано и дало огромное увеличение скорости!

Это утверждение можно проверить смоделировав операции ввода-вывода при помощи функции time.sleep() .

import threading import time def heavy(): # имитации операций ввода-вывода time.sleep(2) def threaded(theads): threads = [] # делим операции на `theads` потоков for thead in range(theads): t = threading.Thread(target=heavy) threads.append(t) t.start() # Подождем, пока все потоки # завершат свою работу. for t in threads: t.join() print(f"theads> циклов имитации операций ввода-вывода закончены") if __name__ == "__main__": start = time.time() # 80 потоков - это неправильно и показано # чисто в демонстрационных целях threaded(80) end = time.time() print("Общее время работы: ", end - start) # 80 циклов имитации операций ввода-вывода закончены # Общее время работы: 2.008725881576538 

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

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

Использование многопроцессорной обработки multiprocessing .

Теперь попробуем настоящую параллельную обработку с использованием модуля multiprocessing . Модуль multiprocessing во многом повторяет API модуля threading , поэтому изменения в коде будут незначительны.

Для того, чтобы произвести 80 циклов вычислений функции heavy() , узнаем сколько процессор имеет ядер, а потом поделим циклы вычислений на количество ядер.

import multiprocessing import time def heavy(n, i, proc): for x in range(1, n): for y in range(1, n): x**y print(f"Цикл № i> ядро proc>") def sequential(calc, proc): print(f"Запускаем поток № proc>") for i in range(calc): heavy(500, i, proc) print(f"calc> циклов вычислений закончены. Процессор № proc>") def processesed(procs, calc): # procs - количество ядер # calc - количество операций на ядро processes = [] # делим вычисления на количество ядер for proc in range(procs): p = multiprocessing.Process(target=sequential, args=(calc, proc)) processes.append(p) p.start() # Ждем, пока все ядра # завершат свою работу. for p in processes: p.join() if __name__ == "__main__": start = time.time() # узнаем количество ядер у процессора n_proc = multiprocessing.cpu_count() # вычисляем сколько циклов вычислений будет приходится # на 1 ядро, что бы в сумме получилось 80 или чуть больше calc = 80 // n_proc + 1 processesed(n_proc, calc) end = time.time() print(f"Всего n_proc> ядер в процессоре") print(f"На каждом ядре произведено calc> циклов вычислений") print(f"Итого n_proc*calc> циклов за: ", end - start) # Весь вывод показывать не будем # . # . # . # Всего 6 ядер в процессоре # На каждом ядре произведено 14 циклов вычислений # Итого 84 циклов вычислений за: 5.0251686573028564 

Код выполнился почти в 5 раз быстрее. Это прекрасно демонстрирует линейное увеличение скорости вычислений от количества ядер процессора.

Использование многопроцессорной обработки с пулом.

Можно сделать предыдущую версию программы немного более элегантной, используя multiprocessing.Pool() . Объект пула, управляет пулом рабочих процессов, в который могут быть отправлены задания. Используя метод Pool.starmap() , можно произвести инициализацию функции sequential () для каждого процесса.

В целях эксперимента в функции запуска пула процессов pooled(core) предусмотрено ручное указание количества ядер процессора. Если не указывать значение core , то по умолчанию будет использоваться количество ядер процессора вашей системы, что является разумным выбором:

import multiprocessing import time def heavy(n, i, proc): for x in range(1, n): for y in range(1, n): x**y print(f"Вычисление № i> процессор proc>") def sequential(calc, proc): print(f"Запускаем поток № proc>") for i in range(calc): heavy(500, i, proc) print(f"calc> циклов вычислений закончены. Процессор № proc>") def pooled(core=None): # вычисляем количество ядер процессора n_proc = multiprocessing.cpu_count() if core is None else core # вычисляем количество операций на процесс calc = int(80 / n_proc) if 80 % n_proc == 0 else int(80 // n_proc + 1) # создаем список инициализации функции # sequential(calc, proc) для каждого процесса init = map(lambda x: (calc, x), range(n_proc)) with multiprocessing.Pool() as pool: pool.starmap(sequential, init) print (calc, n_proc, core) return (calc, n_proc, core) if __name__ == "__main__": start = time.time() # в целях эксперемента, укажем количество # ядер больше чем есть на самом деле calc, n_proc, n = pooled(20) end = time.time() text = '' if n is None else 'задано ' print(f"Всего text>n_proc> ядер процессора") print(f"На каждом ядре произведено calc> циклов вычислений") print(f"Итого n_proc*calc> циклов за: ", end - start) # Весь вывод показывать не будем # . # . # . # Всего задано 20 ядер процессора # На каждом ядре произведено 4 циклов вычислений # Итого 80 циклов за: 5.422096252441406 

Из результатов работы видно, что время работы незначительно увеличилось.

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

Выводы:

• Используйте модули threading или asyncio для программ, связанных с сетевым вводом-выводом, чтобы значительно повысить производительность.
• Используйте модуль multiprocessing для решения проблем, связанных с операциями ЦП. Этот модуль использует весь потенциал всех ядер в процессоре.

• КРАТКИЙ ОБЗОР МАТЕРИАЛА.
• Global Interpreter Lock (GIL)