AttributeError: 'NoneType' object has no attribute 'feed' (possibly race condition on async disconnect)

[bentheiii]

Version: 4.3.4
Platform: python 3.8.13 on azure

Description: unfortunately I have not been able to replicate this error locally, I'm guessing it's a bizarre race condition (possibly an async resurgence of #732)

AttributeError: 'NoneType' object has no attribute 'feed'
  File "***", line **
    await pipeline.execute()
  File "redis/asyncio/client.py", line 1352, in execute
    return await conn.retry.call_with_retry(
  File "redis/asyncio/retry.py", line 59, in call_with_retry
    return await do()
  File "redis/asyncio/client.py", line 1266, in _execute_pipeline
    await self.parse_response(connection, args[0], **options)
  File "redis/asyncio/client.py", line 1291, in parse_response
    result = await super().parse_response(connection, command_name, **options)
  File "redis/asyncio/client.py", line 505, in parse_response
    response = await connection.read_response()
  File "redis/asyncio/connection.py", line 931, in read_response
    response = await self._parser.read_response(
  File "redis/asyncio/connection.py", line 526, in read_response
    await self.read_from_socket()
  File "redis/asyncio/connection.py", line 491, in read_from_socket
    self._reader.feed(buffer)

[emtrane]

I'm able to reproduce the same very consistently with a script like this.

import asyncio
import wrapt
from redis.asyncio.cluster import RedisCluster

async def main():
  host = "localhost"
  port = 6379
  MAX_HASH_SLOTS = 16384

  def _execute_command_wrapper():
      async def _wrapped(f, instance, args, kwargs):
          if len(args) == 1:
              if args[0] == "INFO":
                  return {"cluster_enabled": True}
              elif args[0] == "CLUSTER SLOTS":
                  return [[0, MAX_HASH_SLOTS - 1, [host, port]]]

          return await f(*args, **kwargs)

      return _wrapped

  wrapt.wrap_function_wrapper(
      "redis.asyncio.cluster", "ClusterNode.execute_command", _execute_command_wrapper()
  )

  rc = RedisCluster(
    host=host, 
    port=port
  )

  async def t():
    for i in range(1000):
        async with rc.pipeline() as p:
            p.sadd('foo', 'bar')
            p.expire('foo', 3)
            p.incr('bar')
            p.expire('bar', 3)
            results = await p.execute()
            print(results)

  coros = [t() for i in range(1000)]
  await asyncio.gather(*coros)

asyncio.run(main())

[utkarshgupta137]

Can you guys report back the results with the latest pre-release?

[emtrane]

I forgot to include that but I was running v4.4.0rc1.

[utkarshgupta137]

I'm unable to reproduce the error using the above code on my local mac with v4.4.0rc1.

[utkarshgupta137]

@emtrane Can you try creating a PR with the above code as a failing test?

[juanamari94]

@utkarshgupta137 I've reproduced the error on my local mac with v4.4.2, it often fails at at task the last final tasks (~98000 or so).

I believe that the internal state of the connection's parser is being changed by another task. My guess is that the self._reader variable is being modified by another task while it’s awaiting self.read_from_socket , or from self._stream.read(…)

• self._stream.read(…) in connection.py:

  redis-py/redis/asyncio/connection.py

  Line 389 in 18560b4

  buffer = await self._stream.read(self._read_size)

• self.read_from_socket in connection.py:

  redis-py/redis/asyncio/connection.py

  Line 406 in 18560b4

  await self.read_from_socket()

The self.on_disconnect is not async, so this further increases my suspicion that that’s the method that’s being called by a different task:

redis-py/redis/asyncio/connection.py

Lines 400 to 402 in 18560b4

	if not self._stream or not self._reader:
	self.on_disconnect()
	raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) from None

— this means that either there’s no stream or no reader at a point in time where there should have been one. Looks like the tasks are not collaborating towards a consistent state.

[utkarshgupta137]

@juanamari94 Macs can have weird issues related to the number of open files (connections). I recommend creating a pytest & then creating a PR to see if it still fails.
If it does, I think @kristjanvalur should be able to look into it.

Also, since the async Connection may be the source of the bug, try to recreate it without using the cluster client.

[kristjanvalur]

Interesting, I'll try to reproduce and figure out what's going on.

[kristjanvalur]

What is happening is that someone (a task) disconnects the connection using disconnect(). this starts by clearing the parser state before actually closing the socket.
Meanwhile, a "read" on the socket finishes on a different task. and it finds that the parser has been cleared.

Clearly thee intention here is that one Task can close a connection even if the other is busy reading, and what should
happen is that the parser, upon finding itself cleared, should raise a disconnect.

There are two ways to fix this:

• checking for the None and raising a Disconnection exception
• calling parser.on_disconnect() in the finally clause of the connection.disconnect() thing.

It should be possible to reproduce this problem using Mocks.

[juanamari94]

Hi @kristjanvalur , appreciate the insight! Would you like me to look into writing a unit test for this as @utkarshgupta137 mentioned? Or will you be taking this over?

[kristjanvalur]

I already have a unittest and a PR going on. There are two approaches to fixing this, we can discuss in the PR when ready.

[dvora-h]

Fixed in #2557