Futures cancelled after response.write()

[offline]

Long story short

After executing response.write, future that use aiomysql raise CancelledError exception. If I run the same function one line before response.write - everything works. The problem is I need to respond to client before starting to write to database, so I must do it after response.write

Steps to reproduce

body = json.dumps({"status": "success"}).encode("utf8")
resp = web.StreamResponse(headers={
    "content-type": "application/json",
    "content-length": str(len(body)),
})
await resp.prepare(request)
resp.write(body)
await resp.drain()
await resp.write_eof()
await some_async_func_that_use_aiomysql()
return resp

Your environment

ubuntu, python3.5

[fafhrd91]

Did you get cancelled exception from aiohttp or aiomysql?

[offline]

Traceback (most recent call last):
  File "/private/var/www/moments-aiohttp/handlers/data.py", line 54, in process
    (cuid, name, payload, now)
  File "/private/var/www/moments-aiohttp/.env/lib/python3.5/site-packages/aiomysql/cursors.py", line 239, in execute
    yield from self._query(query)
  File "/private/var/www/moments-aiohttp/.env/lib/python3.5/site-packages/aiomysql/cursors.py", line 460, in _query
    yield from conn.query(q)
  File "/private/var/www/moments-aiohttp/.env/lib/python3.5/site-packages/aiomysql/connection.py", line 398, in query
    yield from self._read_query_result(unbuffered=unbuffered)
  File "/private/var/www/moments-aiohttp/.env/lib/python3.5/site-packages/aiomysql/connection.py", line 582, in _read_query_result
    yield from result.read()
  File "/private/var/www/moments-aiohttp/.env/lib/python3.5/site-packages/aiomysql/connection.py", line 835, in read
    first_packet = yield from self.connection._read_packet()
  File "/private/var/www/moments-aiohttp/.env/lib/python3.5/site-packages/aiomysql/connection.py", line 520, in _read_packet
    packet_header = yield from self._read_bytes(4)
  File "/private/var/www/moments-aiohttp/.env/lib/python3.5/site-packages/aiomysql/connection.py", line 558, in _read_bytes
    data = yield from self._reader.readexactly(num_bytes)
  File "/Library/Frameworks/Python.framework/Versions/3.5/lib/python3.5/asyncio/streams.py", line 670, in readexactly
    block = yield from self.read(n)
  File "/Library/Frameworks/Python.framework/Versions/3.5/lib/python3.5/asyncio/streams.py", line 627, in read
    yield from self._wait_for_data('read')
  File "/Library/Frameworks/Python.framework/Versions/3.5/lib/python3.5/asyncio/streams.py", line 457, in _wait_for_data
    yield from self._waiter
  File "/Library/Frameworks/Python.framework/Versions/3.5/lib/python3.5/asyncio/futures.py", line 361, in __iter__
    yield self  # This tells Task to wait for completion.
  File "/Library/Frameworks/Python.framework/Versions/3.5/lib/python3.5/asyncio/tasks.py", line 296, in _wakeup
    future.result()
  File "/Library/Frameworks/Python.framework/Versions/3.5/lib/python3.5/asyncio/futures.py", line 266, in result
    raise CancelledError
concurrent.futures._base.CancelledError

[fafhrd91]

i think you should execute your query in separate task. that should solve the problem.

[offline]

What do you mean "in separate task"? Like in separate thread?

[jettify]

Take a look on https://docs.python.org/3/library/asyncio-task.html#asyncio.shield probably is what you want

…

On Tue, Nov 29, 2016, 20:59 Anderson ***@***.***> wrote: What do you mean "in separate task"? Like in separate thread? — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub <#1430 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AANoZ9GAQOJKm5Fh4fZrhPR5NQig0zKOks5rDHYWgaJpZM4K9GVd> .

[fafhrd91]

asyncio.ensure_future(some_async_func_that_use_aiomysql())

[offline]

@jettify while it helps to execute the async code, it has two problems

1. All exceptions are silent, if something goes wrong - you don't know about it
2. After the line with shield is executed - it doesn't continue to next line.

Here is a full example:

import json
import logging
import asyncio
import aiomysql
import concurrent.futures
from aiohttp import web
from aiomysql.cursors import DictCursor

logging.basicConfig(level=logging.DEBUG)


async def do_something(pool):
    try:
        async with pool.acquire() as conn:
            async with conn.cursor() as cur:
                await cur.execute("SELECT 4")
    except Exception as e:
        print(traceback.format_exc())

    return 1


async def index(request):
    body = json.dumps({"status": "success"}).encode("utf8")
    resp = web.StreamResponse(headers={
        "content-type": "application/json",
        "content-length": str(len(body)),
    })
    await resp.prepare(request)
    resp.write(body)
    await resp.drain()
    await resp.write_eof()

    await asyncio.shield(do_something(request.app["mysql"]))
    print("Next line")
    return resp




async def get_mysql_pool(loop):
    pool = await aiomysql.create_pool(
        host='localhost',
        user='moments',
        password='moments',
        db='moments',
        cursorclass=DictCursor,
        loop=loop
    )
    return pool


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    mysql = loop.run_until_complete(get_mysql_pool(loop))
    app = web.Application(loop=loop, debug=True)
    app["executor"] = concurrent.futures.ThreadPoolExecutor(max_workers=4)
    app["mysql"] = mysql
    app.router.add_get('/', index)
    web.run_app(app)

You will see that the line with print("Next line") will not be printed.
One more thing - sometimes it does work. It looks like if the sql run fast enough - it procced to next line, if you put there await asyncio.sleep(1) to slower the function execution - it will always fail to get to the next line.

[asvetlov]

That's perfectly fine if a client resets a connection after receiving EOF.
Moreover technically client might do disconnection at any point during request processing.

Current behavior is very good and clean: server's code receives a notification about client's disconnection early. It allows to stop/cancel long running tasks if needed but also allows to protect cancellation if desired.

Your snippet uses shield improperly maybe: logically it should be after sending the last chunk of data but before channel closing.

Spawning new task in this particular case is also a good solution if you'll wrap it into try/except block and log all caught exceptions manually.
Another question is: do you want to count some_async_func_that_use_aiomysql() call duration as part of total request's processing time or not? What do you want to see in access log?

[offline]

Moving the shield execution above await resp.write_eof() doesn't help either. Looks like after resp.write() has been executed - all futures are canceled. Shielded futures handle cancelation, but unable to run lines after it.

Regarding access log - in some situations people will want to know only the time server spoke with the client, in other - how long whole operation run. Maybe some configuration options may be added to aiohttp to address this issue?

@asvetlov when you say spawning a new task, what do you mean by that? Can you give an example?

[asvetlov]

Task spawning == loop.create_task() == ensure_future().

No, it's not how aiohttp works. Inner futures are cancelled only on connection closing, most likely by socket shutting down by client side.

Yes, shielded coroutine doesn't prevent raising CancelledError in outer one: take a look on third snippet from official doc

Please, please, read the fucking docs carefully.

[asvetlov]

It's not an aiohttp bug

[lock]

This thread has been automatically locked since there has not been
any recent activity after it was closed. Please open a new issue for
related bugs.

If you feel like there's important points made in this discussion,
please include those exceprts into that new issue.