Добрых дней, товарищи хабравчане! Изначально я писал этот текст как конспект по асинхронности в пайтоне для самого себя, потому что вещь эта довольно хитрая и не хотелось бы в будущем растерять и забыть знания о ней, когда буду работать с другими языками программирования и фреймворками. Но потом решил попробовать переделать это в формат статьи и опубликовать, потому что сами знаете — когда кому‑то объясняешь, сам начинаешь лучше понимать. Очень надеюсь на конструктивную критику в комментариях, так как хочу сделать эту статью хорошим пособием для новичков или просто людей, которые хотят повторить и закрепить фундаментальные концепции, потому что без их понимания никуда. Итак, приступим.
В разных языках программирования асинхронность крайне отличается, поэтому давать конкретное определение неправильно. А в пайтоне проще сразу показать, как она работает. Но перед этим всё же напишу небольшое введение.
Есть так называемые I/O‑bound операции, или операции ввода/вывода. Под вводом здесь подразумевается не буквально функция input(), а как правило приём байтов из одного сокета в другой. Например из браузерного (клиентского) сокета в серверный и обработка этого набора байтов, или приём байтов из сокета БД или внешнего API.
Так вот в синхронных программах за раз мы могли бы обработать только 1 такой цикл: в сокет пришли байты → мы обработали эти байты → отправили ответ клиенту (другой набор байтов из серверного сокета в клиентский). Это еще называют «блокировкой». Т.е. поток программы как бы «блокируется», становится недоступен во время цикла обработки соединения. Он может обрабатывать только это единственное соединение, и ничего больше. Собственно так работают синхронные воркеры gunicorn, и если мы хотим обработку множества соединений, то делаем несколько воркеров (просто добавляем мультипроцессорность — потому что для синхронных программ это единственный способ обрабатывать несколько операций за раз — с помощью банального дублирования их в нескольких процессах. Да, есть еще многопоточность, но там заправляет GIL, и если очень кратко, пропуская несколько десятилетий споров и размышлений насчёт его работы и полезности, то это фигня, которая позволяет за раз выполняться только 1 потоку. Это сделано для того, чтобы несколько потоков не пытались одновременно изменить одни и те же данные. Это логично, но это убивает всю идею многопоточности — когда несколько потоков работают параллельно и могут выполнять несколько операций сразу). Тем не менее, это всё равно может быть неэффективно — кол‑во одновременных соединений ограничивается кол‑вом воркеров — 8 воркеров могут обрабатывать только 8 соединений за раз, остальные ждут в очереди.
Итак, возвращаемся к асинхронности. В однопоточных программах асинхронность — это способ выполнения программы так, чтобы её поток не блокировался на время цикла обработки соединения и мог принимать и обслуживать другие соединения. И тут я введу только 2 термина:
1. Корутина — функция, которая умеет приостанавливать и возобновлять своё выполнение. Ключевое слово async говорит интерпретатору, что данная функция возвращает корутину, а await позволяет дождаться выполнения той или иной корутины. Если применить await к корутине А внутри корутины Б, то эта внешняя корутина Б приостановит своё выполнение до тех пор, пока не выполнится корутина А.
2. Событийный цикл — алгоритм (бесконечный цикл), который управляет выполнением корутин. Я раскрою это понятие шире, когда покажу пару примеров.
Я уже много всего наболтал, поэтому перейдём уже к примерам.
Смотрим примеры
Что ж, начнём с того, как вообще писать асинхронные программы. Допустим есть синхронный код:
from time import sleep, perf_countert1 = perf_counter()sleep(1) # допустим, это обработка соединения с юзером. # В дальнейшем любой sleep будет приравниваться к обработке какого-то соединения# (с юзером, бд, апи – неважно), чтобы не засорять примеры техническими деталями,# типа создания сокета и т.д.sleep(2)print(perf_counter() - t1)
Процесс, в котором выполняется эта программа, пусть будет тем самым воркером, который должен держать соединения.
Думаю очевидно, что это выполняется 3 секунды. Теперь новички скажут: чтобы сделать код асинхронным, давайте просто добавим везде async/await. Хорошо, давайте:
async def main(): await asyncio.sleep(1) await asyncio.sleep(2)t1 = perf_counter()asyncio.run(main())print(perf_counter() - t1)
Для тех, кто с нуля, опишу, что я здесь сделал. await должен быть внутри корутины, потому что await останавливает выполнение корутины. Соответственно, если написать await вне корутины, то ему банально будет нечего останавливать — выпрыгнет ошибка. Создать корутину можно с помощью слова async.
asyncio.run() — это создание событийного цикла и запуск корутины в нём. Обратите внимание, я передаю не main, а main(), потому что main — это ссылка на обычную функцию, а её вызов возвращает объект корутины, как раз то, с чем умеет работать событийный цикл.
Это я просто расписал формальности, чтобы был понятен синтаксис и примерная логика выполнения. Суть в том, что код этот тоже выполняется 3 секунды. И теперь ответим на пару вопросов:
1. Код асинхронный? Да, потому что есть синтаксис async/await и потому что под капотом запускается событийный цикл, который выполняет корутину.
2. Получили ли мы какую‑то оптимизацию, сделав код асинхронным? НЕТ. Этот код выполняется столько же, сколько и синхронный, он полностью ему аналогичен, различия лишь в механизмах выполнения.
Теперь перепишем скрипт так, чтобы он не просто был асинхронным, но и работал, как асинхронный (как мы только что выяснили, это разные вещи).
async def my_sleep(delay, name): await asyncio.sleep(delay) print(f'Таск {name} завершился с delay = {delay}')async def main(): task1 = asyncio.create_task(my_sleep(1, 'Таск 1')) task2 = asyncio.create_task(my_sleep(2, 'Таск 2')) await task2t1 = perf_counter()asyncio.run(main())print(perf_counter() - t1)
Снова начну с описания изменений, а потом к сути. Во‑первых, мы добавили логирование к asyncio.sleep (функция my_sleep), чтобы понимать какой таск завершился первым. Во‑вторых, в дополнение к корутинам пришли какие‑то таски. И именно на них строится вся асинхронность в пайтоне, собственно сейчас происходит первая кульминация всего этого моего повествования.
Таск — это обёртка над корутиной, который запускает её в фоне (я специально так постарался над выделением этого слова, потому что я потом отдельно подробно распишу, что это значит). Когда в корутине встречается await, она останавливает своё выполнение, таск уведомляет об этом событийный цикл, и он продолжает искать готовые к выполнению таски.
Этот код выполняется за 2 секунды. Как это происходит:
1. В task1 создаём таск, он запускает сон на 1 секунду.
2. В task2 создаём таск, он запускает сон на 2 секунды.
3. Событийный цикл начинает выполнять task1, там встречается await, корутина останавливается, таск передает управление событийному циклу.
4. Событийный цикл начинает выполнять task2, там встречается await, корутина останавливается, таск передает управление событийному циклу.
5. Готовых к выполнению тасков нет, событийный цикл ожидает.
6. Спустя секунду task1 просыпается. Поскольку событийный цикл свободен (не выполняет другие таски), он начинает выполнять task1 и… выполняет его. Мы видим «Таск Таск 1 завершился с delay = 1».
7. Готовых к выполнению тасков нет, событийный цикл ожидает.
8. Спустя ещё одну секунду task2 просыпается. Поскольку событийный цикл свободен (не выполняет другие таски), он начинает выполнять task2 и выполняет его. Мы видим «Таск Таск 2 завершился с delay = 2».
Вот это и есть настоящая асинхронность. Теперь добавлю немного конкретики.
Во‑первых, для начала раскроем понятие событийного цикла. Я уже сказал, что это бесконечный цикл, который управляет тасками и корутинами, но как? Если вы когда то видели реализацию асинхронного сокета без asyncio, а на селекторах, то здесь вам это поможет (если не видели — забейте, щас разжую).
На самом деле все работает на селекторах. Это такой системный вызов (тут универсально расписать не получится, так как на разных ос работает по разному, да и низкоуровневые детали тут особо не нужны, но суть +‑ такая), который возвращает список файловых дескрипторов (которые в том числе могут представлять собой сокеты), готовых для чтения и для записи. Это низкоуровневая сущность ос, и asyncio её не реализовывает, а использует для чтения этих самых готовых дескрипторов. Но работает он не напрямую с дескрипторами, а с дескрипторами, обернутыми в объекты Future (именно он позволяет корутине «останавливать» выполнение, именно он реализует метод __ await__) и Task. Так вот когда корутина останаливается, таск извлекает себя из очереди доступных тасков (передавая коллбэк объекту футуры), файловый дескриптор также переходит в неактивное состояние. Когда корутина возобновляется, селектор будит не таск напрямую, а объект Future; именно Future хранит колбэк, который кладёт таск обратно в очередь готовых тасков.
Во‑вторых, когда готовых тасков нет, событийный цикл ничего не делает (ожидает) — эта фраза значит, что алгоритм внутри цикла физически ждет результата системного вызова select. И начинает работать, когда выполнился п.1 и в списке тасков появился новый таск.
В‑третьих, что значит то самое «в фоне». Эта фраза по сути означает комбинацию п.1 и п.2. Т.е. неактивный таск висит в памяти процесса и ждет, пока дескриптор активируется (станет доступен для чтения/записи), футура вернет таск в список тасков, и событийный цикл его выполнит. Вот это нахождение в памяти и ожидание — и есть «в фоне». В это время событийный цикл выполняет другие готовые таски. Дальше я буду использовать именно эту фразу, потому что она короткая, интуитивно понятная и, на мой взгляд, подходит больше всего для названия вышеописанного механизма.
К делу не относится, но просто интересный факт — asyncio.sleep не создает ни сокетов, ни файловых дескрипторов, тогда как он работает? asyncio ставит таймаут на селектор, то есть ждет готовых дескрипторов столько, сколько мы указали в asyncio.sleep, и, когда не дожидается, просто прерывает это ожидание и продолжает выполнять другие таски.
Пишем асинхронный сервер
Теперь вернемся к нашей главной проблеме — эффективной обработке соединений. Идея в том, чтобы обрабатывать каждое соединение в отдельном таске — это позволит нам держать висящие, неиспользуемые соединения в фоне (имеются в виду соединения, в которые в данную конкретную секунду не поступает никаких сообщений от юзера), а активные будут обрабатываться событийным циклом. Напоминаю, процесс, в котором выполняется наша программа пусть будет называться воркером — теперь уже не синхронным, а асинхронным, как у ювикорна.
Итак, вот код сервера:
Смотреть код
async def accept_connections(server_socket: socket, loop: asyncio.AbstractEventLoop): while True: client_socket, address = await loop.sock_accept(server_socket) print('Приняли соединение') asyncio.create_task(handle_request(client_socket, loop))async def handle_request(client_socket, loop): print('Заходим в цикл обработки соединения') with client_socket as client: while True: data = await loop.sock_recv(client, 1024) await loop.sock_sendall(client, 'Message accepted'.encode('utf-8')) decoded_data = data.decode('utf-8') print('Данные', decoded_data) # Допустим мы как то обрабатываем эти данныеasync def main(): server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_socket.setblocking(False) server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server_socket.bind(('127.0.0.1', 8000)) server_socket.listen() loop = asyncio.get_running_loop() print('Принимаю соединения...') await accept_connections(server_socket, loop)asyncio.run(main())
Я сделал код максимально коротким и простым как репа — никакой обработки ошибок, зависших соединений и всего прочего. Рассказывать про методы сокетов здесь не вижу смысла — мы просто создаем tcp ipv4 сокет, привязываем его к ip + порту в main(). Но кое‑что тут все же важно:
server_socket.setblocking(False) — это делает сокет неблокирующим. Это значит, что после вызова accept (принятие соединения от клиента) программа продолжит выполняться, а не заморозится в ожидании входящего соединения.
Дальше выполняется accept_connection. Он в бесконечном цикле принимает новые соединения и оборачивает их обработку в таск. Если бы мы делали какой‑то мониторинг, healthcheck, обработку ошибок, то стоило бы их положить в какой‑то список, который хранился бы в рамках процесса программы, но опять же этого нет простоты ради.
В handle_request мы опять же в бесконечном цикле общаемся с клиентом, получая от него данные и отправляя их ему.
Таким образом, у нас получается 2 вложенных цикла: 1ый принимает соединения и работает в рамках всего процесса, 2ой обрабатывает само соединение и существует в рамках таска. Благодаря await event loop переключается между этими циклами: когда сервер ждёт данные от клиента, цикл принимает новых клиентов или обслуживает других. При этом в рамках обработки соединения (внутри handle_request) у нас нет никаких тасков, так как там как раз все должно быть последовательно: получил байт → потом обработал → потом получил результат → и только после этого отправил клиенту. Но если бы нам нужно было сходить в БД и во внешнее API, не дожидаясь ответа друг от друга, то это ещё несколько i/o‑bound задач и для них нужно было бы создать таски внутри handle_request, а клиенту отправить уведомление, мол, жди ответа. Но это уже другая история.
Отдельное внимание надо обратить на то, что мы используем не socket.send/socket.recv, а loop.sendall/loop.recv. Первые 2 синхронные и заблокировали бы событийный цикл, а вторые возвращают корутины при вызове и могут быть приостановлены, а значит таск, в котором они были приостановлены, сможет отдать управление событийному циклу. Под капотом loop.sock_recv не просто «ждёт данные». Он регистрирует файловый дескриптор клиентского сокета в селекторе и возвращает управление event loop. А когда селектор сообщает, что сокет готов к чтению — корутина продолжается. Это та самая механика Future и селектора, которую мы разбирали выше.
Кстати, вот код клиента, он синхронный:
Смотреть код
import socketclient_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)with client_socket as client: client.connect(('127.0.0.1', 8000)) while True: data = input('Что отправляем: ') client.send(data.encode('utf-8')) response = client.recv(1024) print('Ответ:', response.decode('utf-8'))
Копируйте себе, меняйте, экспериментируйте, пробуйте. Собственно на этом про асинхронность всё, осталось лишь рассказать про момент, где её нельзя, а точнее полностью бесполезно использовать.
CPU‑bound задачи
Это алгоритмы, где буквально происходят вычисления засчет процессора: итерация по огромному массиву данных, обучение llm, пандас проходит по экселю или csv весом в 50Мб. Проблема в том, что там нет никакого i/o, нет ожидания каких‑то данных, только их непрерывное вычисление. Эти алгоритмы ничего не знают ни про файловые дескрипторы (в контексте сокетов), ни про сокеты, ни про сеть или всякие очереди, им вообще плевать на все то, о чем я распинался выше. Они по своей природе будут выполняться синхронно.
Возьмем такой пример:
def cpu_bound(): return sum(i for i in range(30_000_000))t1 = perf_counter()cpu_bound()cpu_bound()print(perf_counter() - t1)
У меня это выполняется в среднем где то 3 секунды. А теперь прикрутим сюда асинхронщину:
async def async_cpu_bound(): return sum(i for i in range(30_000_000))async def main(): task1 = asyncio.create_task(async_cpu_bound()) task2 = asyncio.create_task(async_cpu_bound())t1 = perf_counter()asyncio.run(main())print(perf_counter() - t1)
Те же +-3 секунды. Тут даже await task1 или await task2 не нужен — cpu_bound задача просто заморозила поток программы и монотонно выполняла вычисление. Событийный цикл хоть и создал таски, но сначала полностью выполнил один, потом второй, потому что они не останавливались из‑за непрерывных, монотонных вычислений.
Решение здесь — это мультипроцессорность. Т.е. забываем про asyncio в контексте создания фоновых тасков и делаем так, чтобы функция cpu_bound физически выполнялась в другом процессе, а не в том, в котором выполняется программа. Перепишем скрипт:
Смотреть код
from concurrent.futures import ProcessPoolExecutordef cpu_bound(): return sum(i for i in range(30_000_000))async def main(): loop = asyncio.get_running_loop() with ProcessPoolExecutor() as pool: t1 = perf_counter() tasks = [ loop.run_in_executor(pool, cpu_bound), loop.run_in_executor(pool, cpu_bound), ] await asyncio.gather(*tasks) print(perf_counter() - t1) if name == '__main__': asyncio.run(main())
Выполнилось за 1.6–1.8 секунды. Т.е. прирост производительности где‑то в 1.7 раза просто засчет того, что мы разбили выполнение по процессам.
Во‑первых, тут важно сказать, что необходим
if name == '__main__',
иначе вылезет ошибка concurrent.futures.process.BrokenProcessPool: A process in the process pool was terminated abruptly while the future was running or pending. Дело в том, что без этой проверки новосозданные процессы будут запускать asyncio.run(main()) — а там опять создание процессов. В итоге они будут плодиться рекурсивно, поэтому мы запрещаем создавать новые процессы в копиях родительского процесса.
Во‑вторых, прирост в скорости будет, только если вы считаете конкретное большое число, а не структуру данных. Если бы cpu_bound возвращал бы список 30млн интов, то было бы даже медленнее, чем когда мы запускали это синхронно. У меня 2 таких вызова выполнялись 7–8 секунд против 4сек на синхронке. Это связано с передачей большого объёма данных между процессами, потому что 30млн интов — это много.
Заключение
Завершу статью аналогией из жизни: вы поставили жариться чебупели на 10 минут. Зачем тупо стоять и залипать на сковородку (вынуждая свой мозг простаивать), когда можно эти 10 минут писать статью про асинхронность в пайтоне? Вы запустили на фон таск — жарку чебупелей. Когда таймер запищит, вы приостановите написание статьи и пойдете заниматься чебупелями.
Но если вы доказываете какую‑то сложную теорему из матанализа, то никак не сможете совместить это со статьёй: вы будете вынуждены полностью погрузиться в решение, потому что мозг и так плавится от условия, тут не до статей (это ваша cpu_bound задача). Однако вы можете заставить решать задачу младшего брата (делегировать cpu_bound задачу другому процессу), а самому пойти дописывать статью.
Если вам понравилась статья, я сделаю вторую часть про внутреннее устройство uvicorn, там мы слегка пройдёмся по его исходникам, составим список сущностей, на которых он работает и разберём, где именно кроется его асинхронность.
ссылка на оригинал статьи https://habr.com/ru/articles/1058626/