asyncio意外asyncio.exceptions.CancelledError

Question

可重现的例子。使用随机超时模拟服务器

import asyncio
import datetime
import random
from aiohttp import web

async def uptime_handler(request):
    chance = random.randint(0, 4)
    if chance == 0:
        await asyncio.sleep(50)
        return
    response = web.StreamResponse()
    response.headers['Content-Type'] = 'text/html'
    await response.prepare(request)

    for i in range(100000):
        formatted_date = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        message = f'{formatted_date}<br>'
        await response.write(message.encode('utf-8'))
        await asyncio.sleep(0.01)

if __name__ == '__main__':
    app = web.Application()
    app.add_routes([ web.get('/', uptime_handler),])
    web.run_app(app)

和一个并行“下载文件”的客户端

import asyncio
import random
import traceback

import aiohttp

URL = 'http://127.0.0.1:8080/'

__http__: aiohttp.ClientSession = None
__semaphore__: asyncio.Semaphore = None

async def download_file(client, url: str):
    try:
        r = await client.get(url, timeout=random.randint(5, 15))
        async for data in r.content.iter_chunked(8192):
            await asyncio.sleep(0.01)
        stub = 2 + 2
        foo = "bar"  # garbage code
    except asyncio.exceptions.TimeoutError:
        try:
            print(f'catch TIMEOUT in download_file')
            await asyncio.sleep(5)
        except asyncio.exceptions.CancelledError:
            print('ALARM ' + traceback.format_exc())

async def _download(client, idx) -> int:
    try:
        for i in range(10):
            print(f'{idx} start')
            try:
                await download_file(client, URL)
            except aiohttp.client.ClientPayloadError:
                break
    except BaseException as e:
        print('**********' + traceback.format_exc())
    finally:
        print(f'{idx} finish')
        __semaphore__.release()
    return idx

async def main():
    global __semaphore__
    __semaphore__ = asyncio.BoundedSemaphore(10)
    async with aiohttp.ClientSession()as client:
        i = 0
        while True:
            i += 1
            await __semaphore__.acquire()
            asyncio.ensure_future(_download(client, i % 10))

if __name__ == '__main__':
    asyncio.run(main())

日志

1 start
...
0 start
catch TIMEOUT in download_file
catch TIMEOUT in download_file
catch TIMEOUT in download_file
8 start
...
0 start
catch TIMEOUT in download_file
catch TIMEOUT in download_file
ALARM Traceback (most recent call last):
  File "F:\dev\test\asynciotest\test_sleep3.9.py", line 15, in download_file
    async for data in r.content.iter_chunked(8192):
  File "C:\Python\lib\site-packages\aiohttp\streams.py", line 39, in __anext__
    rv = await self.read_func()
  File "C:\Python\lib\site-packages\aiohttp\streams.py", line 380, in read
    await self._wait("read")
  File "C:\Python\lib\site-packages\aiohttp\streams.py", line 305, in _wait
    with self._timer:
  File "C:\Python\lib\site-packages\aiohttp\helpers.py", line 641, in __enter__
    raise asyncio.TimeoutError from None
asyncio.exceptions.TimeoutError

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "F:\dev\test\asynciotest\test_sleep3.9.py", line 22, in download_file
    await asyncio.sleep(5)
  File "C:\Python\lib\asyncio\tasks.py", line 654, in sleep
    return await future
asyncio.exceptions.CancelledError

3 start

有时等待TimeioutError处理程序中的asyncio.sleep()会导致CancellationError。为什么？谁取消了asyncio.sleep()协程？

在windows10x64 (1909)上复制/centos8+蟒蛇3.7.9,3.8.3,3.9.5,3.9.6

Answer 1

服务器中有一个无法满足的条件。

random.randint(1, 4)

永远不能返回0，因此条件

if chance == 0:

永远不会是真的。

看来剩下的问题是：服务器能在5- 15秒内写出100K的datetime字符串吗？

我猜有时它无法完成100K的写入，所以客户端取消了请求，从而导致asyncio.exceptions.CancelledError