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.