Awaitable event for write flow control and drain method

[jordaneremieff]

Refers to #70

Calling transport.drain() when writing is paused raises an exception because uvloop doesn't implement a drain method. I found this solution implemented in the following places:

https://gitlab.com/pgjones/quart/commit/cf82d539c67125eb3c3b96e2decb65966f6d174b
sanic-org/sanic@89df1b4

I was able to re-create the error described in the issue with an http chunking consumer:

future: <Task finished coro=<App.__call__() done, defined at /Users/erm/Documents/GitHub/asgi/chunking.py:10> exception=AttributeError("'uvloop.loop.TCPTransport' object has no attribute 'drain'",)> Traceback (most recent call last): File "/Users/erm/Documents/GitHub/asgi/chunking.py", line 40, in __call__ "more_body": bool(i != x), File "/Users/erm/Documents/GitHub/asgi/uvicorn/uvicorn/protocols/http.py", line 156, in send await self.transport.drain() AttributeError: 'uvloop.loop.TCPTransport' object has no attribute 'drain'

After implementing the drain method this no longer occurs. Please advise any changes that may be necessary as my current understanding of the flow control mechanisms is limited.

[jordaneremieff]

To provide a bit more context, this issue has some relevant discussion: sanic-org/sanic#1176

[danigosa]

@ERM could you post the chunked consumer code/tool?

[jordaneremieff]

@danigosa I was using a chunked consumer code initially, but have since been able to reproduce by triggering the drain method with a larger request that does not chunk, here are two examples:

import asyncio


class App:

    def __init__(self, scope):
        self.scope = scope

    async def __call__(self, receive, send):
        await send({
            'type': 'http.response.start',
            'status': 200,
            'headers': [
                (b"content-type", b"text/plain"),
            ]
        })

        body = b"Hello World." * 65536

        await send({
            'type': 'http.response.body',
            'body': body,
            'more_body': False
        })
        await asyncio.sleep(0.1)

and a chunking consumer:

import asyncio


class App:

    def __init__(self, scope):
        self.scope = scope

    async def __call__(self, receive, send):
        await send({
            'type': 'http.response.start',
            'status': 200,
            'headers': [
                (b"cache-control", b"no-cache"),
                (b"content-type", b"text/event-stream"),
                (b"transfer-encoding", b"chunked"),
            ]
        })

        body = b"Hello world." * 65536
        max_ = 5
        for i in range(1, max_ + 1):
            if i == max_:
                more_body = False
            else:
                more_body = True
            await send({
                'type': 'http.response.body',
                'body': body,
                'more_body': more_body
            })
            await asyncio.sleep(0.1)

[danigosa]

@ERM awesomeness, really appreciate the sharing 👍 asyncio based servers (and even worse uvloop ones) they are still having problems compared to the oldie pure WSGI. I will investigate.

[jordaneremieff]

@danigosa what other problems are you running into? If you open any issues I will help where possible.

[danigosa]

Mainly problems with flow control and uvloop/asyncio discrepancies, nowadays the most stable framework by far is aiohttp.

[jordaneremieff]

@danigosa Other than the flow control issue described here, are there other issues you have encountered?

[danigosa]

@ERM I experienced SSL nightmares with uvloop in earlier versions but in production we do SSL termination so I did not have to investigate it a lot, and I think in the new version (0.10) they have focused on improving this

[danigosa]

@ERM BTW we are using a patched version of uvicorn in production with your PR and we cannot find issues (using your consumers example), whilst the master version does for the specific tests

[tomchristie]

Closing in favor of #81.

[tomchristie]

Actually, reviewing this, I'm not so sure I do want to close in favor of #81.

[jordaneremieff]

After experimenting with the protocol behavior quite a bit more, I've realised this solution is actually incomplete - it needs to also handle the drain event when the connection is lost as well. Closing this now since it seems a different solution will likely be implemented in the next release.

[danigosa]

@ERM
is this what you refer to:

'Unexpected ASGI message type 'http.disconnect'.
apistar/server/asgi.py in resolve at line 115
            return body
        while True:
            message = await receive()
            if not message['type'] == 'http.request':
                error = "'Unexpected ASGI message type '%s'."
                raise Exception(error % message['type'])
            body += message.get('body', b'')
            if not message.get('more_body', False):
                break
        return http.Body(body)

It's happening from time to time in our servers, going back to master