4.6.0 - ConnectionError max number of clients reached

[nsteinmetz]

Version: 4.6.0

Platform: Python 3.10 / Docker or Linux

Server : Redis 7.0.11

Description:

We have an application using redis in an async context for months now. With the ugrade of redis-py from 4.5.x to 4.6.0, we saw our app on staging environment starts failing with ConnectionError max number of clients reached

I first run a review of the code to add the missing await redis_conn.close() statements that were missing but the issue still occurs.

It seems that we reach the 10.000 max clients settings in one hour approx whereas we never used to so far. Unfortunately on this environment, I don't track this metric.

On staging last week, I used a workaround to set a client timeout to pass the one hour issue (config set timeout 60)

Doing it again a few minutes ago, I could drop connection from 8k to 160 and it's stable then around 160.

On production environment (Python 3.9, redis 4.5.5), we have 25 connected clients on avergage over the last 6 months with a similar activity to our staging environment. And there is no timeout configuration set, so it's disabled (default configuration)

On staging, reverting to 4.5.5 version of redis-py and with timeout disabled (aka value to 0), after a few minutes it's stable around 20 connections whereas it was to be 1000+ in the same amount of time with 4.6.0.

My code used to be basically :

from redis.asyncio import Redis
from app import settings

async def get_redis_connection():
    redis = Redis.from_url(settings.REDIS_URL)
    return redis

async def list_queued_job_ids():
    redis_conn = await get_redis_connection()
    ... do something with redis data and related code ...
    await redis_conn.close()
    return something

So I don't know if it's a regression in 4.6.0 with the AbstractClass maybe ? or something to add/change in my code but I didn't find any documentation that could help me on the changes to be made.

Let me know if you need further details.

[nsteinmetz]

hi @woutdenolf @dvora-h,

As you worked on and validated the PR #2734 about extracting abstract async connection class would this issue make sense to you ?

[woutdenolf]

Could you create a full example that reproduces the issue?

You can hit maxclients like this but that's to be expected

import asyncio
from redis.asyncio import Redis


async def get_redis_connection():
    redis = Redis.from_url("redis://localhost:6379/0")
    return redis


async def list_queued_job_ids():
    redis_conn = await get_redis_connection()
    result = await redis_conn.incr("counter")
    await asyncio.sleep(1)
    await redis_conn.close()
    return result


async def main(fail=False):
    aws = (list_queued_job_ids() for _ in range(10000 + int(fail)))
    print(max(await asyncio.gather(*aws)))


asyncio.run(main(fail=False))
asyncio.run(main(fail=True))

[nsteinmetz]

Hi,

I'll try before the start of my summer break at the end of week or I'll do once I'm back at the latest. I'll also try to test to see if it's related to the backend/API or one of our workers

And again, on production environment with 4.5.5 release, I'm stable with ~20 connections. So our code is definitely not supposed to work this way to hit the 10.000 maxclients limits as your sample above ;-)

[woutdenolf]

Issue #2814 reports connections not being released to their pool. Maybe related?

[nsteinmetz]

Issue #2814 reports connections not being released to their pool. Maybe related?

I'm not in a cluster context (single redis instance) but who knows... and I think I have some Pipelines in the code btw

[nsteinmetz]

Ok so digging in the code right now and doiing some tests.

So the workflow:

• The IoT device send somes information to our MQTT backend
• The MQTTListener task will get the message and create and arq job for the MQTT worker
• the MQTT worer will see the task and either process it for some functions or will create a telemetry arq task for telemetry worker
• Telemetry worker will see the task to run if any and process it
• end of process

As telemetry worker is the most active, I can see the issue with it.

Telemetry worker is structured as below:

async def device_callback_telemetry(session, device_id, payload):
    logging.debug("Received telemetry from device %s" % device_id)

    redis_conn = await get_redis_connection()

    device_infos = await get_device_infos(session, redis_conn, device_id)
    if device_infos.get("state") is not None and device_infos["state"] == DeviceState.ready:
        telemetry = Telemetry()
        telemetry.ParseFromString(payload)
        await process_telemetry(
            session,
            redis_conn,
            device_infos["id"],
            telemetry,
            device_infos["component"],
        )
    else:
        logging.warning("Received telemetry from %s but this device has not been initialised" % device_id)

    await redis_conn.close()

Then, the redis_conn is passed to a multiple of child functions for yout information till end of process.

It seems that a new client is created every time a new telemetry is sent whereas it was not the case with 4.5.5 release. With my first try with 4.6.0 I also added the await redis_conn.close() which were missing in the code.

So I really wonder why this change happens and why clients are not closed as expected.

[nsteinmetz]

You can see the number of connected clients increase with sample below:

from redis.asyncio import Redis
import asyncio

async def get_redis_connection():
    redis = Redis.from_url("redis://localhost:6379/0")
    return redis


async def do_something():
    redis_conn = await get_redis_connection()
    result = await redis_conn.incr("counter")
    await asyncio.sleep(1)
    await redis_conn.close()
    return result


async def main():
    while True:
        res = await do_something()
        print(res)

asyncio.run(main())

and on the other side:

watch redis-cli info clients

[nsteinmetz]

Ok so the issue is with from_url:

Connected clients is stable with:

from redis.asyncio import Redis
import asyncio

async def get_redis_connection():
    redis = Redis(host="localhost", port=6379, db=0)
    return redis


async def do_something():
    redis_conn = await get_redis_connection()
    result = await redis_conn.incr("counter")
    await asyncio.sleep(1)
    await redis_conn.close()
    return result


async def main():
    while True:
        res = await do_something()
        print(res)

asyncio.run(main())

I also noticed that:

• when I Ctrl+C my script with from_url - clients may not be freed immediatly. If some are immediately freed, it's not all of them but only a part of them and the remaining ones some times later. If I also re-run the script, then seems it creates 20 new connections instead of 1 till around 100 and then it's back to 11 and increase progressively ?!
• when I Ctrl+C my script without from_url- the single connected client is immedialty freed

[woutdenolf]

Nice. I can reproduce it. I compared the behavior with the synchronous version and I noticed this:

If you make a sync equivalent of the code in #2831 (comment), the clients are disconnected when the connections are garbage collected:

#redis/connection.py

class AbstractConnection:

    def __del__(self):
        try:
            self.disconnect()
        except Exception:
            pass

There is no __del__ for the async version so disconnect never gets called. This means all connections you make stay connected.

[woutdenolf]

When I checkout v4.5.5 the connection does have a __del__.

It was removed in #2755

@kristjanvalur Can you provide some insight in why __del__ was removed? The use-case above needs it.

[kristjanvalur]

The reason was that the async version was doing all kinds of nasty stuff, including creating a new event loop, in order to perform a disconnect. __del__ is a synchronous method, whereas disconnect() is async.

The reasoning was also, that if there is a del call, there is no need to disconnect cleanly (i.e a socket disconnect will happen when the socket objects gets collected).
What I failed to take into account was that in case of connection pooling, this is not a disconnect, but a return to the pool.
connections need to return to the pool, not disconnect. We can do this more cleanly. I will create a pr which achieves this without having to invoke all the async stuff.

[kristjanvalur]

update: The difference in behaviour is that when Redis.from_url() is called, then Redis.auto_close_connection == False, whereas it is True otherwise.
I think this is a bug, since a private connection pool is indeed created, and should be closed.

When it is False, the pool is simply garbage collected, and also the connections in it. For some reason, however, without an explicit disconnect, the connection stays open. I need to investigate how that happens.

[nsteinmetz]

Hey @kristjanvalur

I also noticed that from_urlsignature is not the same for sync & async version

For async it's from_url(url, **kwargs) - as stated in the doc : https://redis-py.readthedocs.io/en/stable/connections.html#redis.Redis.from_url

For sync, it's

def from_url(cls,
             url: str,
             *,
             host: str | None = ...,
             port: int | None = ...,
             db: int | None = ...,
             password: str | None = ...,
             socket_timeout: float | None = ...,
             socket_connect_timeout: float | None = ...,
             socket_keepalive: bool | None = ...,
             socket_keepalive_options: Mapping[str, int | str] | None = ...,
             connection_pool: ConnectionPool | None = ...,
             unix_socket_path: str | None = ...,
             encoding: str = ...,
             encoding_errors: str = ...,
             charset: str | None = ...,
             errors: str | None = ...,
             decode_responses: Literal[True],
             retry_on_timeout: bool = ...,
             ssl: bool = ...,
             ssl_keyfile: str | None = ...,
             ssl_certfile: str | None = ...,
             ssl_cert_reqs: str | int | None = ...,
             ssl_ca_certs: str | None = ...,
             ssl_check_hostname: bool = ...,
             max_connections: int | None = ...,
             single_connection_client: bool = ...,
             health_check_interval: float = ...,
             client_name: str | None = ...,
             username: str | None = ...) -> Redis[str]
Return a Redis client object configured from the given URL

Back to async version, if I try:

async def get_redis_connection():
    redis = Redis.from_url("redis://localhost:6379/0", auto_close_connection_pool=False)
    return redis

I would get:

TypeError: AbstractConnection.__init__() got an unexpected keyword argument 'auto_close_connection_pool'

whereas I would expect it to be the same as:

async def get_redis_connection():
    redis = Redis(host="localhost", port=6379, db=0, auto_close_connection_pool=False)
    return redis