WSGI and uvloop

[dalf]

uvloop.install() in a WSGI application crashes granian with this error

thread 'tokio-runtime-worker' panicked at 'called `Result::unwrap()` on an `Err` value: PyErr { type: <class 'RuntimeError'>, value: RuntimeError("Task <Task pending name='Task-1' coro=<Event.wait() running at /home/alexandre/.pyenv/versions/3.9.16/lib/python3.9/asyncio/locks.py:226> cb=[<builtins.PyTaskCompleter object at 0x7f4fd0c56cd0>()]> got Future <Future pending> attached to a different loop"), traceback: Some(<traceback object at 0x7f4fd0b73b00>) }', src/wsgi/serve.rs:21:5
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

There is no error with uwsgi for example.

Minimal code:

from flask import Flask

app = Flask(__name__)

try:
    import uvloop
except ImportError:
    pass
else:
    uvloop.install()

@app.route('/')
def hello_world():
    return 'Hello, World!'

Use case: an URL endpoint sends multiple HTTP requests, and then aggregate the results. To avoid too many threads, the code uses the async version of httpx.
I did not open an issue because I feel this is an edge case.

[gi0baro]

🤔 why would you call uvloop.install()? The loop is already initialised by Granian itself. Maybe I should add a note in README or docs.

But also: how do you use async code in a sync WSGI app? I'm definitely missing something here..

[dalf]

A more complete example:

import asyncio
import threading
from flask import Flask
from httpx import AsyncClient

app = Flask(__name__)

# start a loop
try:
    import uvloop
except ImportError:
    pass
else:
    uvloop.install()

def loop_thread():
    global LOOP
    LOOP = asyncio.new_event_loop()
    LOOP.run_forever()

thread = threading.Thread(
    target=loop_thread,
    name='asyncio_loop',
    daemon=True,
)
thread.start()

client = AsyncClient()

@app.route('/')
def fetch():
    URLS = [
        'https://bing.com',
        'https://google.com',
        'https://startpage.com',
    ]
    futures = [
        asyncio.run_coroutine_threadsafe(client.get(url), loop=LOOP)
        for url in URLS
    ]
    texts = [
        future.result().text
        for future in futures
    ]
    return texts

if __name__ == '__main__':
    app.run()

Feel free to ignore this usage.
I know in the sync world, the usual pattern is to use one thread per url (as long there are not too many url).

I'm curious why it crashes in the Rust code, but I will take more time to read the code.

[gi0baro]

So, a few things:

• the event loop in Python always run in the main thread
• if you need to access the loop, you should get it with asyncio.get_running_loop()
• I still miss the point of running asynchronous code in a blocking portion of code

On the last point: your code will block on the asyncio.run_coroutine_threadsafe, so it will be the same as using httpx client in a sync context.
If you want to use threads, there's no need to use asyncio, you can run the requests using a threadpool, with concurrent.futures, see https://docs.python.org/3/library/concurrent.futures.html#threadpoolexecutor-example

[dalf]

your code will block on the asyncio.run_coroutine_threadsafe, so it will be the same as using httpx client in a sync context.

As far I understand and see:

• in the code above, asyncio.run_coroutine_threadsafe allows to send the HTTP requests in parallel.
• it works because the fetch function and the loop run in different threads (at least for the default granian configuration)

demo with asyncio.sleep

import asyncio
import threading
from flask import Flask

app = Flask(__name__)
loop = asyncio.get_event_loop()


def printthreadid(prefix):
    print(f"{prefix:20} thread id={threading.get_ident()}")


async def this_is_async(t):
    printthreadid('this_is_async')
    await asyncio.sleep(t)


@app.route('/')
def fetch():
    printthreadid('fetch')
    futures = [
        asyncio.run_coroutine_threadsafe(this_is_async(t), loop=loop)
        for t in [1, 2, 3]
    ]
    for future in futures:
        future.result()
    return 'ok'


printthreadid('main')

if __name__ == '__main__':
    app.run()

$ granian --interface wsgi testasync &
[INFO] Starting granian
[INFO] Listening at: 127.0.0.1:8000
[INFO] Booting worker-1 with pid: 481320
main                 thread id=139719734102912
[INFO] Started worker-1
[INFO] Started worker-1 runtime-1
fetch                thread id=139719689975360
this_is_async        thread id=139719734102912
this_is_async        thread id=139719734102912
this_is_async        thread id=139719734102912

$ # in another terminal
$ time curl http://127.0.0.1:8000
ok
real    0m3.021s
user    0m0.005s
sys     0m0.013s

I still miss the point of running asynchronous code in a blocking portion of code

Sorry, I have skipped the question; I've tried to migrate a sync app with a lot of legacies to an async framework and stopped halfway for various reasons.

if you need to access the loop, you should get it with asyncio.get_running_loop()

Ok !
The line uvloop.install() exists because uwsgi ignores uvloop.