StreamingResponse with async generator get stuck indefinitely

[iudeen]

The starting point for issues should usually be a discussion...

https://github.com/encode/starlette/discussions

Possible bugs may be raised as a "Potential Issue" discussion, feature requests may be raised as an "Ideas" discussion. We can then determine if the discussion needs to be escalated into an "Issue" or not.

This will help us ensure that the "Issues" list properly reflects ongoing or needed work on the project.

---

• Initially raised as discussion in FastAPI StreamingResponse with async generator sometimes causes socket.send() exception indefinitely tiangolo/fastapi#5183

import asyncio

from starlette.applications import Starlette
from starlette.responses import StreamingResponse
from starlette.routing import Route


def sync_streamer():
    try:
        while True:
            yield b"--boundary\r\nContent-Type: text/plain\r\nContent-Length: 1\r\n\r\n1\r\n"
    except asyncio.CancelledError:
        print("caught cancelled error")
        raise GeneratorExit


async def async_streamer():
    try:
        while True:
            yield b"--boundary\r\nContent-Type: text/plain\r\nContent-Length: 1\r\n\r\n1\r\n"
    except asyncio.CancelledError:
        print("caught cancelled error")
        raise GeneratorExit


async def sync_endpoint(request):
    try:
        return StreamingResponse(
            sync_streamer(),
            media_type="multipart/x-mixed-replace; boundary=boundary",
        )
    except asyncio.CancelledError:
        print("caught cancelled error")
        raise GeneratorExit


async def async_endpoint(request):
    try:
        return StreamingResponse(
            async_streamer(),
            media_type="multipart/x-mixed-replace; boundary=boundary",
        )
    except KeyboardInterrupt:
        exit()


routes = [
    Route('/sync', sync_endpoint),
    Route('/async', async_endpoint),
]

app = Starlette(routes=routes)

CASE 1: When /async (Asynchronous Generator) is hit

>uvicorn stream:app
INFO:     Started server process [23288]
INFO:     Waiting for application startup.                               
INFO:     Application startup complete.                                  
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     127.0.0.1:65107 - "GET /async HTTP/1.1" 200 OK

# stuck indefinitely after client cancels the request.

CASE 2: When /sync (Synchronous Generator) is hit

>uvicorn stream:app
INFO:     Started server process [24968]
INFO:     Waiting for application startup.                               
INFO:     Application startup complete.                                  
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     127.0.0.1:55582 - "GET /sync HTTP/1.1" 200 OK
Exception in callback _ProactorBasePipeTransport._call_connection_lost(None)
handle: <Handle _ProactorBasePipeTransport._call_connection_lost(None)>     
Traceback (most recent call last):                                          
  File "...\lib\asyncio\events.py", line 80, in _run
    self._context.run(self._callback, *self._args)
  File "...\lib\asyncio\proactor_events.py", line 162, in _call_connection_lost
    self._sock.shutdown(socket.SHUT_RDWR)
ConnectionResetError: [WinError 10054] An existing connection was forcibly closed by the remote host

When CASE 1, even `Keyboard Interrupts are ignored. I had to kill Python process forcefully.

[iudeen]

Did some quick debugging, seems like its not reaching this part of the code when Cancelled, when Async generator is used.

    async def listen_for_disconnect(self, receive: Receive) -> None:
        while True:
            message = await receive()
            if message["type"] == "http.disconnect":
                break

[Kludex]

If you add an await asyncio.sleep(0), does it solve the problem?

async def async_streamer():
    try:
        while True:
            yield b"--boundary\r\nContent-Type: text/plain\r\nContent-Length: 1\r\n\r\n1\r\n"
            await asyncio.sleep(0)  # <--
    except asyncio.CancelledError:
        print("caught cancelled error")
        raise GeneratorExit

[iudeen]

Yes, it does! Now exceptions are caught, we get GeneratorExit.

What's happening here?

[Kludex]

Other tasks are not able to run, as it's stuck on the async_streamer, because it doesn't have any await statements. 👀

[iudeen]

Oh! Is it something the developer is expected to handle (not much aware of async generators yet) or, it'd be a good scenario starlette can handle?

[Kludex]

Yes. Nothing we can do about it.

[vishal09vns]

How to solve this issue.
StreamingResponse starts sending response once all the elements of async_streamer function are generated.