cancellederror still exist.

[larryclean]

Long story short

i've upgraded aiohttp to 2.3 and I processed all my requests with aiojobs.but my project still throws a lot of “cancellederror”,affected aioredis storage.

Expected behaviour

I expect cancellederror to be completely resolved

Actual behaviour

Cancellederror often happens when my application is not accessed for a long time or very complicated

Steps to reproduce

---

Your environment

aiohttp2.3
aioredis
aiomysql
python-socket.io
centos
python3.6.2

[hellysmile]

You need third-party lib to fix asyncio.CancelledError on clients disconnects.

Here is a couple of them:

https://pypi.python.org/pypi/async_armor
https://github.com/aio-libs/aiojobs

[asvetlov]

The question without code sample is too broad.

[larryclean]

Almost all methods operate at aioredis, so there is not code sample

[kxepal]

I could confirm, that aiohttp + aioredis causes CancelErrors from aioredis calls side when connection get interrupted by user or else. But I guess, that issue doesn't relates to aiohttp and we nothng here can do.

[asvetlov]

I see aiojobs mentioning in initial message.
If all aioredis calls are wrapped by aiojobs.spawn() -- client interruptions should not affect them.
Otherwise aiojobs is buggy, which is very unlikely.
I need a code sample for reproducing the problem.

[larryclean]

aiojobs decorator

def use_aiojob(app):
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kw):
            scheduler = app['siojobs']
            job = await scheduler.spawn(func(*args, **kw))
            return await job.wait()
        return wrapper
    return decorator

socket.io disconnect event

@sio.on('disconnect')
    @use_aiojob(app)
    async def disconnect(sid):
        """
        链接断开连接
        weixin.liu@serviceindeed.com 20170417
        :param sid:
        :return:
        """
        await factory.leave_room(sid)
        rooms = sio.rooms(sid)
        for room in rooms:
            sio.leave_room(sid, room)
        await sio.eio.disconnect(sid)

factory leave_roome

    async def leave_room(self, sid):
        """
        用户离开房间
        :param sid:
        :return:
        """
        try:
            data, user, sid_user_hall_room_list = await self.user_logic.leave_room(sid, self.check_sid_valid, self.port)
            if sid_user_hall_room_list:
                return None
            if data:
                hall = data.get('hall')
                room = data.get('room')
                user_id = data.get('user_id')
                if int(room) > 0:
                    online = await self.game_logic.leave_game(hall, room, user_id)
                    if online:
                        await self.sio.emit("update_game", online, room=str(hall) + "_" + str(room))
                rt_data_log(data, "leave_room", "退出房间")
                await self.sio.emit("notice", {
                    "type": "message",
                    "message": user.get("user_name", "") + "退出房间"
                }, room=str(hall) + "_" + str(room))
            await self.user_logic.delete_sid(sid, self.port)
        except RTException as e:
            print(e.error)

user_logic leave_room

    async def leave_room(self, sid, handle=None, port=0):
        """
        退出在线对弈
        :param sid:
        :return:
        """
        try:

            temp_user_sid = None
            redis = await self.get_redis()
            # 根据sid 查询当前用户
            key = REDIS_KEY['RT_SID_HALL_ROOM'].format(sid)
            result = await redis.hgetall(key)
            user_info = dict()
            if result:
                # 删除用户sid关联关系
                await redis.srem(REDIS_KEY['RT_USER_SID'].format(result.get('user_id')), sid)
                # 删除用户在房间下的这个sid
                await redis.srem(REDIS_KEY['RT_USER_HLL_ROOM_SID'].format(result.get('user_id'), result.get('hall'),
                                                                          result.get('room')), sid)
                # 设置sid最新一次所在的大厅和房间和用户ID
                await redis.delete(REDIS_KEY['RT_SID_HALL_ROOM'].format(sid))

                # 根据用户 查询 该用户是否还有sid 存在
                sid_list = await redis.smembers(REDIS_KEY['RT_USER_SID'].format(result.get('user_id')))
                # 根据用户 查询 该用户是否还有sid 存在
                sid_user_hall_room_list = await redis.smembers(
                    REDIS_KEY['RT_USER_HLL_ROOM_SID'].format(result.get('user_id'), result.get('hall'),
                                                             result.get('room')))
                # 获取用户信息
                user_info = await redis.hgetall(REDIS_KEY['RT_USER'].format(result.get('user_id')))

                # 校验sid是否有效 如果无效则删除
                temp_user_sid = sid_user_hall_room_list[:]
                if sid_user_hall_room_list:
                    for i in sid_user_hall_room_list:
                        if handle:
                            sids = await handle(i)
                            if not sids:
                                temp_user_sid.remove(i)
                                await redis.srem(
                                    REDIS_KEY['RT_USER_HLL_ROOM_SID'].format(result.get('user_id'),
                                                                             result.get('hall'),
                                                                             result.get('room')), i)
                # 如果用户在该房间下没有了sid
                if not temp_user_sid:
                    # 删除房间下在线用户
                    await redis.srem(REDIS_KEY['RT_USER_HALL'].format(result.get('hall'), result.get('room')),
                                     REDIS_KEY['RT_USER'].format(result.get('user_id')))

                # 如果用户已经没有sid
                if not sid_list:
                    # 用户设为离线当为匿名用户，则删除
                    user_info['online_status'] = 0
                    await redis.hset(REDIS_KEY['RT_USER'].format(result.get('user_id')), 'online_status', 0)
                    match = await redis.hgetall(
                        REDIS_KEY['RT_HASH_MATCH_HIS'].format(result.get('hall', 1), result.get('user_id')))
                    if match:
                        await redis.zrem(match.get('list_key'), match.get('rule_key'))
                        await redis.srem(
                            REDIS_KEY['RT_MATCH_SET_USER'].format(result.get('hall', 1), result.get('user_id')),
                            match.get('rule_key'))
                        await redis.delete(
                            REDIS_KEY['RT_HASH_MATCH_HIS'].format(result.get('hall', 1), result.get('user_id')))
            return result, user_info, temp_user_sid
        except Exception as e:
            error_log.error(traceback.format_exc())
            raise RTException(e)

gamelogic.leave_room

    async def leave_game(self, hall, room, user_id):
        """
        用户退出对局
        :param hall:
        :param room:
        :param user_id:
        :return:
        """
        try:
            redis = await self.get_redis()
            # 获取对局
            game_key = REDIS_KEY['RT_GAME'].format(hall, room)
            game = await redis.hgetall(game_key)
            if int(game.get("blackId")) != int(user_id) and int(game.get('whiteId')) != int(user_id):
                return None
            # 设置用户离线
            now = int(time.time())
            if int(game.get("blackId")) == int(user_id):
                game['blackIsOff'] = 1
                game['blackOffTime'] = now
            elif int(game.get('whiteId')) == int(user_id):
                game['whiteIsOff'] = 1
                game['whiteOffTime'] = now
            save_flag = False
            if int(game.get("blackId")) == int(user_id) or int(game.get('whiteId')) == int(user_id):
                save_flag = True
            else:
                return None
            game = await self._calc_game_time(game_key, game, save_flag)
            if int(game.get('offlineTime', -1)) > -1:
                if int(game.get("blackId")) == int(user_id) and int(game.get('curColor')) == 1:
                    game['isStop'] = 1
                elif int(game.get('whiteId')) == int(user_id) and int(game.get('curColor')) == -1:
                    game['isStop'] = 1
                await redis.hmset_dict(game_key, game)

            return self.__foramt_game_status(game)
        except Exception as e:
            raise RTException(e)

This is code snippet of my project，
cancellederror follow:

[2017-10-24 14:34:35][ERROR][Traceback (most recent call last):
  File "/home/wwwroot/ykrealtime/rtgame/logics/userlogic.py", line 203, in leave_room
    result = await redis.hgetall(key)
  File "/home/wwwroot/ykrealtime/venv36/lib/python3.6/site-packages/aioredis/util.py", line 77, in wait_make_dict
    res = yield from fut
concurrent.futures._base.CancelledError
][userlogic.py][257]

[larryclean]

Did you find any problems? @asvetlov

[asvetlov]

As I see from code and exceptions the problem is not in aiohttp itself but in python-socketio library.
It should catch cancellation and handle it as normal disconnection (call @sio.on('disconnect') with protecting the callback).
Nothing to do on aiohttp side.

[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].
[new issue]: https://github.com/aio-libs/aiohttp/issues/new