Почему await не переключает коррутины?

Question

[Иван Фурсов]

Я так поняла что await переключает внимание цикла event_loop с одной коррутины на следующую, но в моём случае это не работает

import asyncio

async def c():
    pass

async def a():
    print("start a")
    await c()
    print("finish a")

async def b():
    print("start b")
    await c()
    print("finish b")

async def main():
    loop.create_task(a())
    print("create task")
    loop.create_task(b())

loop = asyncio.new_event_loop()
loop.run_until_complete(main())
loop.run_forever()

выводит

create task
start a
finish a
start b
finish b

но я ожидаю вывод

create task
start a
start b
finish a
finish b

Answer 1

[Roman K]

в def c сделай await asyncio.sleep(0.1). Иначе ты просто напарываешься на конечный автомат ивентлупа. И там как раз всё логично и вывод правильный.

Более того, правильно запускать asyncio.run(main()) и явное использование loop — deprecated

Comments

[Иван Фурсов]

ну тут я не очень поняла, я же создаю таски и с помощью await я должна переключаться между тасками

[shurshur]

Бабка Люда, переключение происходит не как попало, а на что-то "следующее" в цикле событий. В данном примере в момент создания таска a() запускается a() и весь код синхронный до await c() в a(); в момент вызова c() запускается именно c() (а не продолжается main(), хотя чисто теоретически реализация могла бы и её запустить без противоречия теории, но говорят, что реально реализация всегда запустит c()) и опять код синхронный до завершения c(), потом опять продолжается a(). Переключение в main() для её продолжения произойдёт только после завершения a().

Все советчики тут советуют asyncio.sleep не просто так: в этом случае и a(), и c() должны будут подождать, и цикл событий точно за время этого ожидания переключится на main() - единственную подходящую для активизации задачу в списке на выполнение.

[Roman K]

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

Держи

import asyncio
import heapq
from asyncio import DefaultEventLoopPolicy, SelectorEventLoop
from asyncio.base_events import MAXIMUM_SELECT_TIMEOUT, _format_handle

from asyncio.log import logger

_MIN_SCHEDULED_TIMER_HANDLES = 100

_MIN_CANCELLED_TIMER_HANDLES_FRACTION = 0.5


class MyEventLoop(SelectorEventLoop):
    def run_forever(self):
        print("run_forever")
        return super().run_forever()

    def _run_once(self):
        """Run one full iteration of the event loop.

        This calls all currently ready callbacks, polls for I/O,
        schedules the resulting callbacks, and finally schedules
        'call_later' callbacks.
        """

        print("-"*40)
        print(f"_run_once. {self._scheduled=}, {self._ready=}")

        sched_count = len(self._scheduled)
        if (
            sched_count > _MIN_SCHEDULED_TIMER_HANDLES
            and self._timer_cancelled_count / sched_count
            > _MIN_CANCELLED_TIMER_HANDLES_FRACTION
        ):
            # Remove delayed calls that were cancelled if their number
            # is too high
            new_scheduled = []
            for handle in self._scheduled:
                if handle._cancelled:
                    handle._scheduled = False
                else:
                    new_scheduled.append(handle)

            heapq.heapify(new_scheduled)
            self._scheduled = new_scheduled
            self._timer_cancelled_count = 0
        else:
            # Remove delayed calls that were cancelled from head of queue.
            while self._scheduled and self._scheduled[0]._cancelled:
                self._timer_cancelled_count -= 1
                handle = heapq.heappop(self._scheduled)
                handle._scheduled = False

        timeout = None
        if self._ready or self._stopping:
            timeout = 0
        elif self._scheduled:
            # Compute the desired timeout.
            when = self._scheduled[0]._when
            timeout = min(max(0, when - self.time()), MAXIMUM_SELECT_TIMEOUT)

        print(f"select timeout {timeout=}")

        event_list = self._selector.select(timeout)
        print(f"got {len(event_list)} events: {event_list=}")
        self._process_events(event_list)

        # Handle 'later' callbacks that are ready.
        end_time = self.time() + self._clock_resolution
        print(f"{self._scheduled=}")
        while self._scheduled:
            handle = self._scheduled[0]
            if handle._when >= end_time:
                break
            handle = heapq.heappop(self._scheduled)
            handle._scheduled = False
            self._ready.append(handle)

        # This is the only place where callbacks are actually *called*.
        # All other places just add them to ready.
        # Note: We run all currently scheduled callbacks, but not any
        # callbacks scheduled by callbacks run this time around --
        # they will be run the next time (after another I/O poll).
        # Use an idiom that is thread-safe without using locks.
        ntodo = len(self._ready)
        print(f"{self._ready=}")
        for i in range(ntodo):
            handle = self._ready.popleft()
            if handle._cancelled:
                continue
            if self._debug:
                try:
                    self._current_handle = handle
                    t0 = self.time()
                    handle._run()
                    dt = self.time() - t0
                    if dt >= self.slow_callback_duration:
                        logger.warning(
                            "Executing %s took %.3f seconds", _format_handle(handle), dt
                        )
                finally:
                    self._current_handle = None
            else:
                handle._run()
        handle = None  # Needed to break cycles when an exception occurs.
        print("_run_once FINISHED")
        print("=" * 40)


class X(DefaultEventLoopPolicy):
    def new_event_loop(self):
        return MyEventLoop()


asyncio.set_event_loop_policy(X())


async def c():
    pass


async def a():
    print("start a")
    await c()
    print("finish a")


async def b():
    print("start b")
    await c()
    print("finish b")


async def main():
    asyncio.create_task(a())
    print("create task")
    asyncio.create_task(b())


if __name__ == "__main__":
    asyncio.run(main())

Не обращай внимания на несколько run_forever, хоть это и контринтуитивно. Всё потому что asyncio.run выполняет очистку

def run(main, *, debug=None):
    if events._get_running_loop() is not None:
        raise RuntimeError(
            "asyncio.run() cannot be called from a running event loop")

    if not coroutines.iscoroutine(main):
        raise ValueError("a coroutine was expected, got {!r}".format(main))

    loop = events.new_event_loop()
    try:
        events.set_event_loop(loop)
        if debug is not None:
            loop.set_debug(debug)
        return loop.run_until_complete(main)
    finally:
        try:
            _cancel_all_tasks(loop)  # Здесь
            loop.run_until_complete(loop.shutdown_asyncgens())  # Здесь
            loop.run_until_complete(loop.shutdown_default_executor())  # И здесь
        finally:
            events.set_event_loop(None)
            loop.close()

[Иван Фурсов]

а возможно зделать так что бы на функции с() цикл переключался на след. задачу как с asyncio.sleep()
но при этом не добавляя в c() asyncio.sleep()?

[Roman K]

Бабка Люда, не добавляя — нет. Только asyncio.sleep(0) (у нуля есть своя кастомная реализация, которая просто пропускает один цикл ивентлупа).

И ещё раз повторю: в этом нет никакого смысла.

[Roman K]

Бабка Люда, и ещё замечу, что ты в коде создаёшь таски, но нигде их не await'ишь. И тот факт, что они успевают завершиться — случайно совпавшие факторы. В следующий раз такого может и не быть. Поэтому таски НУЖНО await'ить.

[Иван Фурсов]

но зачем это делать, ведь эти функции должны были выполняться

[Иван Фурсов]

асинхронно

[Roman K]

Бабка Люда, в чём и к чему вопрос?

[shurshur]

Бабка Люда, тут ошибка в том, что функции лучше понимать не как выполняющие действия, а как возвращающие результат в математическом смысле (или в смысле подхода из функционального программирования). То есть если результат функции не нужен для достижения окончательного результата, то её и запускать не нужно, даже если она описана в коде. В реальности, конечно, asyncio не совсем чистое функциональное программирование, но ошибку function was never awaited любой новичок наверняка видел прям в первом же своём эксперименте (фактически она говорит "результат этой функции как конкретное вычисленное значение ни разу не был затребован при выполнении программы / вычислении Главной Функции main").

Answer 2

[shurshur]

asyncio вообще никак не гарантирует порядок, в котором будут выполняться функции и выполняться переключение на следующую. В этом примере, например, будет запущена a() и ничего про запуск b() не будет ещё известно среде выполнения. Затем будет запущена c(), быстро завершится и завершит a().

Смысла в таком тесте мало - тут всё очень быстро происходит. Я бы рекомендовал для экспериментов вставить asyncio.sleep в функцию c, чтобы она не завершилась так уж быстро, а также во время этого ожидания гарантировано выбралась бы следующая задача - в данном случае продолжилось выполнение main().

Comments

[Roman K]

Строго говоря, гарантирует. Это можно увидеть в методе _run_once. Но просчитать это тяжело и расчитывать на это нельзя.

[shurshur]

Roman Kitaev, ну, скажем так, текущая имплементация перебирает цикл определённым образом, который можно посмотреть в сырцах, но это не гарантирует, что это не будет изменено. Да и на более сложных примерах уже ничего не будет так однозначно. Особенно, когда это будет завязано на реальный ввод-вывод - на что, собственно, asyncio и проектировали.

[Roman K]

shurshur, твоя правда. Я даже не уверен, что это работает хотя бы примерно так в uvloop

Answer 3

[baynic]

Переключение происходит тогда, когда происходит событие, требующее ожидание. В функции c() замените pass на await asyncio.sleep(1)

Comments

[Иван Фурсов]

но если в функции c будет вечный цикл то переключение не произойдёт

[shurshur]

Бабка Люда, всё так, вечный цикл - вечно синхронный код.

[Иван Фурсов]

shurshur, я имела в виду что

async def c():
    while True:
         pass

то выведет

create_task
start a

но если бы await переключил бы то вывелось бы

create_task
start a
start b

не так ли?

[shurshur]

Бабка Люда, да, всё так, цикл while True: pass является синхронным и никогда не позволит перейти в eventloop, а значит и что-то переключить.

[Иван Фурсов]

когда я создала соппрограмму на основе генератора всё заработало как мне нужно

@asyncio.coroutine
def c():
    yield