Как я могу упаковать coroutine как нормальную функцию в цикле событий?

Я использую asyncio для сетевой инфраструктуры.

В приведенном ниже коде (low_level - наша низкоуровневая функция, main - это наша программа, user_func - пользовательская функция):

import asyncio

loop = asyncio.get_event_loop()
""":type :asyncio.AbstractEventLoop"""


def low_level():
    yield from asyncio.sleep(2)


def user_func():
    yield from low_level()


if __name__ == '__main__':
    co = user_func()
    loop.run_until_complete(co)

Я хочу обернуть low_level как нормальную функцию, а не coroutine (для compatibility и т.д.), но low_level находится в цикле событий. Как можно обернуть его как нормальную функцию?

Ответ 1

Поскольку low_level является сопрограммой, ее можно использовать только при запуске цикла событий asyncio. Если вы хотите иметь возможность вызывать его из синхронного кода, который не запускает цикл событий, вам необходимо предоставить оболочку, которая фактически запускает цикл событий и запускает сопрограмму до завершения:

def sync_low_level():
    loop = asyncio.get_event_loop()
    loop.run_until_complete(low_level())

Если вы хотите иметь возможность вызывать low_level() из функции, входящей в цикл цикла выполнения, заблокировать ее в течение двух секунд, но не использовать yield from, ответ заключается в том, что вы не можете, Цикл событий является однопоточным; всякий раз, когда выполнение выполняется внутри одной из ваших функций, цикл события блокируется. Никакие другие события или обратные вызовы не могут быть обработаны. Единственные пути для функции, запущенной в цикле событий, чтобы вернуть управление циклу событий, равны 1) return 2), используйте yield from. Вызов asyncio.sleep в low_level никогда не сможет завершить, если вы не выполните одну из двух этих функций.

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

import asyncio

loop = asyncio.get_event_loop()

@asyncio.coroutine
def low_level(loop=None):
    yield from asyncio.sleep(2, loop=loop)


def sync_low_level():
    new_loop = asyncio.new_event_loop()
    new_loop.run_until_complete(low_level(loop=new_loop))

@asyncio.coroutine
def user_func():
    sync_low_level()

if __name__ == "__main__":
    loop.run_until_complete(user_func())

Но я действительно не уверен, почему вы хотите это сделать.

Если вы просто хотите, чтобы low_level действовал как метод, возвращающий Future, поэтому вы можете прикреплять к нему обратные вызовы и т.д., просто оберните его в asyncio.async():

loop = asyncio.get_event_loop()

def sleep_done(fut):
    print("Done sleeping")
    loop.stop()

@asyncio.coroutine
def low_level(loop=None):
    yield from asyncio.sleep(2, loop=loop)

def user_func():
    fut = asyncio.async(low_level())
    fut.add_done_callback(sleep_done)

if __name__ == "__main__":
    loop.call_soon(user_func)
    loop.run_forever()

Вывод:

<2 second delay>
"Done sleeping"

Кроме того, в вашем примере кода вы должны использовать декоратор @asyncio.coroutine для low_level и user_func, как указано в asyncio docs

Корутин - это генератор, который следует определенным соглашениям. Для документации, все сопрограммы должны быть украшены @asyncio.coroutine, но это не может быть строго соблюдено.

Edit:

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

@asyncio.coroutine
def low_level(loop=None):
    yield from asyncio.sleep(2, loop=loop)

def thr_low_level():
   loop = asyncio.new_event_loop()
   t = threading.Thread(target=loop.run_until_complete, args(low_level(loop=loop),))
   t.start()
   t.join()

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