why pubsub cannot read message occasioanlly and how to troubleshoot?

[zillionare]

we have write a redis-log adaptor:

class RedisLogReceiver:
    def __init__(
        self,
        dsn: str,
        channel_name: str,
        filename: str,
        backup_count: int = 7,
        max_bytes: Union[str, int] = 10 * 1024 * 1024,
        fmt: str = None,
    ):
        self._dsn = dsn
        self._channel_name = channel_name
        self._backup_count = backup_count
        self._max_bytes = self.parse_max_bytes(max_bytes)
        self._dir = os.path.dirname(filename)
        if not os.path.exists(self._dir):
            try:
                os.makedirs(self._dir)
            except Exception as e:
                print(e)
                print("创建日志目录失败，已将日志目录更改为：/tmp/omega.log")
                filename = "/tmp/omega.log"

        self._filename = os.path.split(filename)[-1]
        self._fmt = (
            fmt
            or "%(asctime)s %(levelname)-1.1s %(process)d %(name)s:%(funcName)s:%(lineno)s | %(message)s"
        )

        # the file handler to save log messages
        self._fh = open(filename, mode="a", encoding="utf-8", buffering=1)

        # bytes written. to decide when to rotate files
        self._written_bytes = os.path.getsize(filename)

        # the redis connection
        self._redis = None

        # the channel returned by redis.subscribe
        self._channel = None

        # the loop for listen and dump log
        self._reader_task = None

    def _write(self, msg: str):
        self._written_bytes += len(msg)
        if (1 + self._written_bytes) % 4096 == 0:
            self._fh.flush()

        if self._written_bytes > self._max_bytes:
            self.rotate()

        if self._fh is None:
            print(msg)
        else:
            self._fh.write(msg)
            self._fh.write("\n")

    async def stop(self):
        self._fh.flush()
        self._fh.close()
        self._fh = None

        await self._channel.unsubscribe(self._channel_name)
        await self._reader_task
        await self._channel.close()
        await self._redis.close()

    async def start(self):
        self._redis = aioredis.from_url(self._dsn)
        self._channel = self._redis.pubsub(ignore_subscribe_messages=True)
        await self._channel.subscribe(self._channel_name)
        self._reader_task = asyncio.create_task(self.reader())

    async def reader(self):
        async for msg in self._channel.listen():
            msg = msg.get("data").decode("utf-8")
            self._write(msg)

above code works on several machines but not all. All machines has same configuration:

redis = 5.0.6 same build
aioredis = 2.0.1
python=3.9

the above code will receive 'subscribe message' all the time though. We tried to publish message from redis-client but no good. We have tried example code in documentation (write in a standalone, executeable python file) but no good too.

Could anybody tell me why, or if the information is not enough, how should I do troubleshoot and collect more info?

[Andrew-Chen-Wang]

Have you checked whether the connection was disconnected, the event loop is still running, connections for each subscription were simply broken after 3 seconds, whether you hit the timeout, and whether the health check was the reason for failure?

[zillionare]

1. the connection should not be disconnected, but we don't know how to verify(if it's disconnected, there're should be exception, right?)
2. the event loop is still running, since the program is still running
3. connections for each subscription were simply broken after 3 seconds: why 3 seconds?
4. whether you hit the timeout: there should be no timeout regarded. the same code runs on other machine smoothly
5. we'll try health check later.

by the way, we got another exception today, not sure if it's related (pubsub involved though):

.tox/py39/lib/python3.9/site-packages/pyemit/emit.py:358: in stop
    await pubsub.close()
.tox/py39/lib/python3.9/site-packages/aioredis/client.py:3972: in reset
    await self.connection.disconnect()
.tox/py39/lib/python3.9/site-packages/aioredis/connection.py:806: in disconnect
    await self._writer.wait_closed()  # type: ignore[union-attr]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <StreamWriter transport=<TCPTransport closed=True reading=False 0x5637a0f30780> reader=<StreamReader 58 bytes transport=<TCPTransport closed=True reading=False 0x5637a0f30780>>>

    async def wait_closed(self):
>       await self._protocol._get_close_waiter(self)
E       RuntimeError: await wasn't used with future

why aioredis not always return coroutine here?

[zillionare]

also, we have another related issue here: https://bugs.python.org/issue46223

not sure it's related

[Andrew-Chen-Wang]

Yes, that may be a bug in Python. We properly close and wait_closed, so I'm not sure what's wrong. Maybe we prematurely canceled on write?