Repository: redis/redis-py
File: redis/asyncio/client.py (Python)

Code before:

    def pubsub(self, **kwargs) -> "PubSub":
        """
        Return a Publish/Subscribe object. With this object, you can
        subscribe to channels and listen for messages that get published to
        them.
        """
        return PubSub(
            self.connection_pool, event_dispatcher=self._event_dispatcher, **kwargs
        )

    def monitor(self) -> "Monitor":
        return Monitor(self.connection_pool)

    def client(self) -> "Redis":
        return self.__class__(
            connection_pool=self.connection_pool, single_connection_client=True
        )

    async def __aenter__(self: _RedisT) -> _RedisT:
        """
        Async context manager entry. Increments a usage counter so that the
        connection pool is only closed (via aclose()) when no context is using
        the client.
        """
        async with self._usage_lock:
            self._usage_counter += 1
        try:
            # Initialize the client (i.e. establish connection, etc.)
            return await self.initialize()
        except Exception:
            # If initialization fails, decrement the counter to keep it in sync
            async with self._usage_lock:
                self._usage_counter -= 1
            raise

    async def _decrement_usage(self) -> int:
        """
        Helper coroutine to decrement the usage counter while holding the lock.
        Returns the new value of the usage counter.
        """
        async with self._usage_lock:
            self._usage_counter -= 1
            return self._usage_counter

    async def __aexit__(self, exc_type, exc_value, traceback):
        """
        Async context manager exit. Decrements a usage counter. If this is the
        last exit (counter becomes zero), the client closes its connection pool.
        """
        current_usage = await asyncio.shield(self._decrement_usage())

Review comment (suggestion):
I would suggest adding another function _increment_usage for this operation. It will be easier to read and follow the code if here you call _increment_usage and below in the except clause you call _decrement_usage. The same applies to cluster.py changes as well.

Code after:
                     key.
            time: 5, thread-2 acquired `my-lock` now that it's available.
                     thread-2 sets the token to "xyz"
            time: 6, thread-1 finishes its work and calls release(). if the
                     token is *not* stored in thread local storage, then
                     thread-1 would see the token value as "xyz" and would be
                     able to successfully release the thread-2's lock.

        ``raise_on_release_error`` indicates whether to raise an exception when
        the lock is no longer owned when exiting the context manager. By default,
        this is True, meaning an exception will be raised. If False, the warning
        will be logged and the exception will be suppressed.

        In some use cases it's necessary to disable thread local storage. For
        example, if you have code where one thread acquires a lock and passes
        that lock instance to a worker thread to release later. If thread
        local storage isn't disabled in this case, the worker thread won't see
        the token set by the thread that acquired the lock. Our assumption
        is that these cases aren't common and as such default to using
        thread local storage."""
        if lock_class is None:
            lock_class = Lock
        return lock_class(
            self,
            name,
            timeout=timeout,
            sleep=sleep,
            blocking=blocking,
            blocking_timeout=blocking_timeout,
            thread_local=thread_local,
            raise_on_release_error=raise_on_release_error,
        )

    def pubsub(self, **kwargs) -> "PubSub":
        """
        Return a Publish/Subscribe object. With this object, you can
        subscribe to channels and listen for messages that get published to
        them.
        """
        return PubSub(
            self.connection_pool, event_dispatcher=self._event_dispatcher, **kwargs
        )

    def monitor(self) -> "Monitor":
        return Monitor(self.connection_pool)

    def client(self) -> "Redis":
        return self.__class__(
            connection_pool=self.connection_pool, single_connection_client=True
        )