В Кремниевой долине есть очень особенный ресторан фаст-фуда, который всегда открыт. Там имеется один столик, за ним может разместиться лишь один посетитель, которому дадут совершенно фантастический гамбургер. Когда туда приходишь — ждёшь до тех пор, пока не настанет твоя очередь. Потом хозяин ресторана подведёт тебя к столику, и, это же Америка, тебе зададут, кажется, бесконечное количество вопросов о том, как приготовить и как подать твой гамбургер.
Сегодня, правда, мы не собираемся говорить о кулинарных изысках. Мы говорим о системе очередей, которую используют рестораны. Если вам повезло и вы прибыли в ресторан тогда, когда столик пуст, и когда никого в очереди нет, вы можете прямо сразу за него сеть. В противном случае хозяин даст вам специальный пейджер (из бескрайней кучи таких пейджеров!) и вы можете бродить вокруг ресторана до тех пор, пока этот пейджер не подаст сигнал. Дело хозяина ресторана — обеспечить, чтобы посетители попадали бы за столик в порядке их прибытия. Когда настанет ваша очередь, хозяин отправит сигнал на ваш пейджер, а вы вернётесь в ресторан, где сможете усесться за столик.
Если вы передумали — можете вернуть пейджер хозяину, а он, ни слова не сказав, спокойно его заберёт. Если ваш пейджер уже сработал, то хозяин, если на вас очередь не оканчивается, вызовет следующего посетителя. Посетители этого ресторана всегда вежливы, они, получив пейджер, не уходят украдкой, никого не предупредив. А хозяин всегда честен и не усадит за столик кого-то, кто стоит в очереди за вами, даже если вам, чтобы вернуться в ресторан после срабатывания пейджера, нужно некоторое время.
Вышеприведённое описание соответствует описанию класса Lock, представляющего взаимно исключающую блокировку. Когда посетитель приходит в ресторан — это то же самое, что вызов acquire(), а уход из ресторана — это вызов release(). Если посетитель, после того, как стал в очередь, решил уйти — это отмена операции во время ожидания в acquire(). Решить уйти можно до срабатывания пейджера или после этого. То есть — отменить операцию можно до или после того, как блокировка активировала ваш вызов (но до возврата из acquire()).
Однажды ресторан расширится, наймёт ещё помощников шеф-повара, в нём поставят несколько новых столов. У ресторана, как и прежде, будет один хозяин, работа которого, на самом деле, не особо изменится. Но, так как одновременно в ресторане можно будет разместить несколько посетителей, теперь, вместо простой блокировки, придётся использовать семафор — класс Semaphore.
Оказывается, реализация примитивов синхронизации — дело непростое. В случае с библиотекой asyncio это несколько неожиданно, так как единовременно может выполняться только одна задача, а переключение задач выполняется лишь в await. Но в прошлом году под сомнение были поставлены такие вещи, как справедливость распределения ресурсов, обеспечиваемая asyncio, её корректность, семантика и производительность. На самом деле, три последних жалобы появились в прошлом месяце. Мне, последнему, кто занимается задачами с метками expert-asyncio, приходится в спешке разбираться с тем, как лучше всего представить себе семафоры.
Метафора с рестораном оказалась очень полезной. Например, есть разница между количеством свободных столиков и количеством посетителей, которых можно безотлагательно за столики усадить. Эта разница равна количеству посетителей, пейджеры которых сработали, но которые ещё не подошли после этого к хозяину ресторана.
В деле справедливого распределения ресурсов есть одна особая проблема, которая проявляется, когда задача, освободившая семафор и немедленно пытающаяся снова его захватить, может дискриминировать другие задачи. Это — то же самое, как если бы посетитель ресторана встал из-за столика, вышел бы из ресторана, тут же вернулся бы и уселся бы за столик снова, не обращая внимания на очередь из других посетителей.
В библиотеке была ошибка, когда отменённый вызов acquire() мог оставить блокировку в неправильном состоянии. Это — как если бы хозяин ресторана пришёл бы в замешательство, когда посетитель со сработавшим пейджером возвращает пейджер, но отказывается садиться за столик.
Но ресторанная метафора не способна помочь во всех ситуациях. Дело в том, что при отмене acquire() в asyncio происходит сложная последовательность событий. В Python 3.11 мы начали нагружать операцию отмены дополнительными задачами из-за двух новых добавленных нами в систему асинхронных менеджеров контекста:
Класс TaskGroup используется для управления группами родственных задач. Если одна из задач даёт сбой, другие отменяются, а менеджер контекста ожидает выхода всех задач.
Функция timeout() применяется для управления тайм-аутами. Когда срабатывает тайм-аут — текущая задача отменяется.
Вот основные сложности, которые возникают при обработке отмен задач:
В процессе ожидания завершения работы объекта класса Future, представляющего собой конечный результат асинхронной операции, эта операция может быть отменена. После этого операция
awaitзавершается с ошибкой, вызывая исключение CancelledError.Но когда объект
Futureнаходится в состоянии ожидания и вызываетсяCancelledError, нельзя предполагать, что операция этого объекта была отменена! Возможно и такое, что объектFutureуже помечен как имеющий результат (то есть — его больше нельзя отменить), а задача была помечена как готовая к выполнению. Но другая (тоже готовая к выполнению) задача запускается первой и отменяет первую задачу. Выражаю благодарность пользователю Cyker Way за то, что указал на эту исключительную ситуацию.
Это помогает представить себе объект Future как нечто, пребывающее в одном из четырёх состояний:
Ожидание.
Завершение работы, имеется результат.
Завершение работы, имеется исключение.
Завершение работы, но это — следствие отмены операции.
Из состояния ожидания объект Future может перейти в одно из других состояний, после чего снова изменить состояние он не может (тут можно было бы вставить симпатичную картинку с диаграммой состояний :-) ).
Семафор управляет FIFO-очередью задач, ожидающих выполнения. Он не использует состояние, соответствующее выдаче исключения, но три других состояния использует:
Ожидание: посетитель с пейджером, на который ещё не отправили сигнал о готовности блюда.
Владение результатом: посетитель, на пейджер которого отправили уведомление.
Отмена: посетитель, который вернул пейджер до того, как на него было отправлено уведомление.
Справедливое распределение ресурсов должен обеспечивать механизм, в соответствии с логикой которого новый объект Future всегда добавляют в конец очереди. Это происходит, когда вызов acquire() обнаруживает, что семафор заблокирован. При этом, когда очередь не пуста, при вызове release() самый левый (то есть — самый старый) объект Future всегда маркируют как объект, содержащий результат. Ошибка, касающаяся справедливого распределения ресурсов, произошла из-за того, что вызов acquire() работает по сокращённой схеме в том случае, если переменная семафора level (количество свободных столиков) является ненулевой. Этот вызов не должен так поступать в том случае, если в очереди ещё есть объекты Future. Другими словами — ошибка выражалась в том, что мы иногда усаживали только что прибывших посетителей за столики в ситуации, когда у нас были свободные столики, несмотря на то, что в это время кто-то уже ждал в очереди.
Как думаете — что вызывало ошибку, связанную с отменой операции? Речь идёт о сценарии, когда объект Future владеет результатом (посетитель, на пейджер которого поступил сигнал), но задача, ожидающая этот объект Future, отменяется (посетитель отказывается усесться за столик).
Я изо всех сил пытался представить себе состояние семафора, с его переменной level и FIFO-очередью объектов Future, пребывающих в состоянии ожидания. Ещё я пытался найти определение вызова locked(). Если бы переменная level была бы общедоступной, я попытался бы разобраться и с её семантикой. В итоге у меня получились следующие определения:
W— список ожидающих объектовFuture, или[f для f в очереди, если не выполнен вызов f.done()].R— список объектовFuture, владеющих результатом, или[f для f в очереди, если выполнен вызов f.done() и не выполнен вызов f.cancelled()].C— список объектовFuture, операции, соответствующие которым, отменены, или[f для f в очереди, если выполнен вызов f.cancelled()].
А вот — некоторые конструкции, построенные на основе этих определений:
set(W + R + C) == set(queue)— все объектыFuture, которые либо ожидают в очереди, либо имеют результат, либо отменены.level >= len(R)— у нас должно быть как минимум столько свободных столиков, сколько имеется посетителей, на пейджеры которых отправлен сигнал о готовности заказа.define locked() as (len(W) > 0 or len(R) > 0 or level == 0)— мы можем немедленно усадить кого-то за столик при выполнении следующих условий:нет посетителей, ожидающих срабатывания пейджера,
нет посетителей, на пейджеры которых уже отправлен сигнал о готовности заказа,
имеется по крайней мере один свободный столик.
В итоге — вот ссылка на текущий код реализации класса Semaphore.
О, а приходите к нам работать? 🤗 💰
Мы в wunderfund.io занимаемся высокочастотной алготорговлей с 2014 года. Высокочастотная торговля — это непрерывное соревнование лучших программистов и математиков всего мира. Присоединившись к нам, вы станете частью этой увлекательной схватки.
Мы предлагаем интересные и сложные задачи по анализу данных и low latency разработке для увлеченных исследователей и программистов. Гибкий график и никакой бюрократии, решения быстро принимаются и воплощаются в жизнь.
Сейчас мы ищем плюсовиков, питонистов, дата-инженеров и мл-рисерчеров.
