Bidirectional streaming

[benblack769]

Is concurrent bidirectional streaming supported by httpcore? This is a key feature of HTTP/2.0 that I was hoping httpcore/httpx would support, but it does not seem like it is.

I wrote an httpx client to handle a bidirectional streaming http server, one that sends body request packets back at the same time it receives body response packets.

    content_list = []
    async with httpx.AsyncClient(http1=False, http2=True) as client:
        async with client.stream(method="POST", url=f"http://127.0.0.1:8000/bidirectional", content=generate_chunks(), timeout=10.) as response:
            async for chunk in response.aiter_raw():
                   ....

However, I found that this logic in httpcore https://github.com/encode/httpcore/blob/master/httpcore/_async/http2.py#L110 where it awaits for the body to send before returning the Response ensures that responses cannot be accepted by the client at the same time as requests are sent.

One incredibly naive solution I found is to simply replace:

await self._send_request_body(request=request, stream_id=stream_id)

With

asyncio.get_event_loop().create_task(self._send_request_body(request=request, stream_id=stream_id))

This way, the body is sent concurrently with the response being received.

Is this something that you hope to support? If so, then I can take a look in the code and try to find a better solution.

[tomchristie]

You're looking for the network_stream extension...

https://www.encode.io/httpcore/extensions/#network_stream

That'll give you a read/write interface directly onto either the TCP connection (HTTP/1.1) or onto an individual HTTP/2 stream.

I'd recommend that you start out by giving it a go from httpcore directly, and then switch up to using it with httpx once you're confident you've got everything working...

url = httpcore.URL(b"http", b"127.0.0.1", 8080, b"http://www.example.com")
with httpcore.stream("POST", url) as response:
    network_stream = response.extensions["network_stream"]
    ...

We documented it for "CONNECT" method requests and requests using the Upgrade header. It should also be suitable for bi-directional streaming.

I wrote an httpx client to handle a bidirectional streaming http server, one that sends body request packets back at the same time it receives body response packets.

It'd be useful to understand your use-case more clearly here?

Also please do please let us know how you get on with this, it'd be great to get more feedback on our network stream interface.

[benblack769]

So I ran into a problem with the "network_stream" extension fairly quickly. It only seems to be avaliable for the sync API. But bidirectional streaming means concurrent uploads and downloads, so it must be avaliable via an async API.

Here is the code I am trying:

import httpcore
from httpcore._async.connection_pool import AsyncConnectionPool
url = httpcore.URL(b"http://127.0.0.1:8000")

pool = AsyncConnectionPool(http1=False, http2=True)

async def main():
    headers = {"Host": "\0"}
    request = httpcore.Request(url=url, method='POST', headers=headers)
    response = await pool.handle_async_request(request)
    network_stream = response.extensions["network_stream"] # error here!!!
    print(network_stream)

import asyncio
loop = asyncio.get_event_loop()
loop.run_until_complete(main())

[benblack769]

Or perhaps it is possible to use this stream in some concurrent manner?

[benblack769]

Sorry about the curt replies, I was just having a very hard time getting anything to work with any library, and this was only one of many client/server combinations I was trying.

I finally got something working using the nghttp client:

Test Server:

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


@app.post("/bidirectional")
async def bidirectional():
    @stream_with_context
    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()))

Client:

# create large file to process
python -c 'print("data"*1000*1000*8)' > massive_request.txt 
# call the service and get the response concurrently.
nghttp -d massive_request.txt --no-content-length -O http://127.0.0.1:8000/bidirectional > processed_output.txt

The key requirement which is very difficult to get to work (but does work in this example) is that the server cannot store the entire request or response (note the Queue(max_size=4)).

[tomchristie]

For the server-side part you might(?) find it easier to create an ASGI app (rather than working at a request/response level). It's a bit low level and fiddly, but it'd probably be simpler for structuring concurrent read/write operations against.

Also I'd recommend using anyio for it's TaskGroup functionality, rather than spawning tasks with asyncio's loop interfaces.

The one other thing that I realized after my reply was that it turns out "network_stream" isn't currently supported with HTTP/2. (Only with HTTP/1.1) Looks like there's a chunk of work that still needs doing there.

[benblack769]

Well on the server side, the Quart framework does allow for this concurrent request-response handling for HTTP2 POST requests, as desired, so that will be my go-to technology, as ASGI does seem a bit low level for my needs. Just it would be slightly preferable to use starlette, due to some existing internal knowledge of the system.

On the client side, its a bit different. I took a look around, and could not find any http2 client that supports bidirectional streaming in python. However, other langauges have some support, for example NodeJS's builtin HTTP2 module https://nodejs.org/api/http2.html#client-side-example (Requires a bit of callback hell to use, but I managed to get bidirectional streaming to work in the end) or in c/c++ there is the nghttp2 library (haven't tried it directly, but NodeJS depends on it), or on the command line, where there is nghttp or curl enabled with nghttp2, which both work. So it should be possible.

As for httpcore, a refactoring that would prove very useful would be if for the AsyncHTTP2Connection class specifically, if the handle_async_request could be split into three async functions:

# class AsyncHTTP2Connection:
async def init_async_request(self, request: Request) -> int:
    ...
    return stream_id
async def send_async_request(self, request: Request, stream_id:int) -> None:
     # sends headers and body to server
     ...
async def receive_async_request(self, request: Request, stream_id:int) -> Response:
     # receives response from server
     ...
     return Response(...)

For backwards compatability, the current handle_async_request function could be implemented as so:

async def handle_async_request(self, request: Request) -> Response:
    stream_id = await self.init_async_request(request)
    await self.send_async_request(request, stream_id)
    return await self.receive_async_request(request, stream_id)

and the user could potentially do something like this:

stream_id = await self.init_async_request(request)
send_task = loop.create_task(self.send_async_request(request, stream_id))
response = await self.receive_async_request(request, stream_id)
async for chunk in response.content:
     # process chunk
await send_task

If this is possible, then that might just meet my current needs.

p.s. Thanks for mentioning AnyIO, it does seem like a good technology, especially to use as a basis for a library, but I haven't had a chance to really look at it yet.

[benblack769]

@tomchristie I ended up simply abandoning HTTP, and used websockets instead. Still have some trouble managing backpressure, but it seems to be doing better than HTTP.