Bidirectional streaming

[benblack769]

I am trying to create a web service with bidirectional streaming, i.e. the server should concurrently be receiving new data chunks at the same time it is sending back chunks. This is a key feature of the http/2.0 protocol. Here is my minimal reproducible example:

import asyncio

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


async def simple_make_generator(stream):
    buffer = asyncio.Queue(maxsize=8)
    loop = asyncio.get_running_loop()

    async def write_to_buffer(stream):
        async for chunk in stream:
            await buffer.put(chunk)
        await buffer.put(b"")
        await buffer.join()

    write_task = loop.create_task(write_to_buffer(stream))
    while True:
        chunk = await buffer.get()
        yield chunk
        buffer.task_done()
        if not chunk:
            break
    await write_task


async def main_page(request: Request):
    body = request.stream()
    gen = simple_make_generator(body)
    return StreamingResponse(gen)


app = Starlette( routes=[
    Route('/bidirectional', main_page, methods=['POST']),
])


if __name__ == "__main__":
    from hypercorn.config import Config
    from hypercorn.asyncio import serve
    import asyncio
    asyncio.run(serve(app, Config()))

And here is a way of calling it with a large dataset, generated with python -c 'print("data"*1000*1000*8)' > massive_request.txt

curl -v --http2  --http2-prior-knowledge http://127.0.0.1:8000/bidirectional -X POST --data-binary @massive_request.txt -H "Content-Length:" > arg.txt

This call stalls after uploading ~10% of the data, and makes no further progress.

Now I know this should be possible because I managed to get an identical server running with the Quart framework:

from quart import Quart, render_template, websocket, request, stream_with_context, make_response
import asyncio
app = Quart(__name__)


@app.post("/bidirectional")
async def bidirectional():
    async def simple_make_generator(stream):
        buffer = asyncio.Queue(maxsize=4)
        loop = asyncio.get_running_loop()

        async def write_to_buffer(stream):
            async for chunk in stream:
                await buffer.put(chunk)
            await buffer.put(b"")
            await buffer.join()

        write_task = loop.create_task(write_to_buffer(stream))
        while True:
            chunk = await buffer.get()
            yield chunk
            buffer.task_done()
            if not chunk:
                break
        await write_task

    response = await make_response(simple_make_generator(request.body))
    response.timeout = None  # No timeout for this route
    return response


if __name__ == "__main__":
    from hypercorn.config import Config
    from hypercorn.asyncio import serve
    import asyncio
    asyncio.run(serve(app, Config()))

And this works perfectly.

[vvvin333]

I tried partially related stuff with re-streaming audio data from another source of bytes using asyncio.Queue. And had two sources for having unbreakable stream. So when one source is awaiting, another one always produced "silence" chunks. Then my queue is "always" filled in.

from asyncio import CancelledError, Queue
from aiostream import stream

...
async def stream_from_queue(queue: Queue) -> AsyncGenerator:
   combined_stream = stream.merge(get_audio_chunk(queue), generate_silence())
   async with combined_stream.stream() as streamer:
          try:
               async for chunk in streamer: 
                   yield chunk
          except CancelledError:
          ...

P.S. Though I sticked with the problem of starlette wrappers which added some redundant bytes which breaks my ICY (Icecast) protocol for meta data inserting. https://stackoverflow.com/q/79142077/15080117

[adriangb]

See #1830.

Will be fixed by #2732 eventually.