Repository: redis/redis-py
Pull request #2568 (2023-01-28 03:55:04)

Fix issue 2540: Synchronise concurrent command calls to single-client to single-client mode
### Pull Request check-list

- [x] Does `$ tox` pass with this change (including linting)?
- [x] Do the CI tests pass with this change (enable it first in your forked repo and wait for the github action build to finish)?
- [x] Is the new or changed code fully tested?
- [ ] Is a documentation update included (if this change modifies existing APIs, or introduces new ones)?
- [ ] Is there an example added to the examples folder (if applicable)?
- [ ] Was the change added to CHANGES file?


### Description of change

Fixing issue [#2540](https://github.com/redis/redis-py/issues/2540) (example script in issue tested and bug no longer occurs)

The underlying cause is that the connection in single-client mode isn't protected by a lock, resulting in multiple coroutines reading/writing to a connection stream when multiple commands are called in parallel (forexample with asyncio.gather)

Additionally, redundant connections can be made if the first time the client is used, it is used in parallel (forexample with asyncio.gather).

A simple lock mechanism has been added to the `execute_command` function to ensure that single-client mode is always executing commands in sequence. This lock is also used each time the `initialize` method is called, ensuring that multiple connections aren't spun up if the first time the client is called happens to be with a parallel mechanism.

Worth mentioning that the use cases for using parallel calling mechanisms in single-client mode are probably limited, though this still seems useful to have in terms of consistency.




codecov-commenter: # [Codecov](https://codecov.io/gh/redis/redis-py/pull/2568?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) Report
Base: **92.26**% // Head: **78.30**% // Decreases project coverage by **`-13.97%`** :warning:
> Coverage data is based on head [(`6660ed5`)](https://codecov.io/gh/redis/redis-py/pull/2568?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) compared to base [(`e39c7ba`)](https://codecov.io/gh/redis/redis-py/commit/e39c7babdd80ecdd930bed9d201e8fb9187309ac?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis).
> Patch coverage: 97.29% of modified lines in pull request are covered.

:mega: This organization is not using Codecov’s [GitHub App Integration](https://github.com/apps/codecov). We recommend you install it so Codecov can continue to function properly for your repositories. [Learn more](https://about.codecov.io/blog/codecov-is-updating-its-github-integration/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis)

<details><summary>Additional details and impacted files</summary>


```diff
@@             Coverage Diff             @@
##           master    #2568       +/-   ##
===========================================
- Coverage   92.26%   78.30%   -13.97%     
===========================================
  Files         115      115               
  Lines       29610    29645       +35     
===========================================
- Hits        27320    23213     -4107     
- Misses       2290     6432     +4142     
```


| [Impacted Files](https://codecov.io/gh/redis/redis-py/pull/2568?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) | Coverage Δ | |
|---|---|---|
| [tests/test\_asyncio/test\_connection.py](https://codecov.io/gh/redis/redis-py/pull/2568?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9hc3luY2lvL3Rlc3RfY29ubmVjdGlvbi5weQ==) | `96.52% <96.42%> (-1.18%)` | :arrow_down: |
| [redis/asyncio/client.py](https://codecov.io/gh/redis/redis-py/pull/2568?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvYXN5bmNpby9jbGllbnQucHk=) | `93.10% <100.00%> (+0.59%)` | :arrow_up: |
| [tests/test\_asyncio/test\_cluster.py](https://codecov.io/gh/redis/redis-py/pull/2568?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9hc3luY2lvL3Rlc3RfY2x1c3Rlci5weQ==) | `18.51% <0.00%> (-79.10%)` | :arrow_down: |
| [tests/test\_cluster.py](https://codecov.io/gh/redis/redis-py/pull/2568?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9jbHVzdGVyLnB5) | `17.99% <0.00%> (-78.96%)` | :arrow_down: |
| [redis/asyncio/cluster.py](https://codecov.io/gh/redis/redis-py/pull/2568?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvYXN5bmNpby9jbHVzdGVyLnB5) | `16.96% <0.00%> (-74.72%)` | :arrow_down: |
| [redis/cluster.py](https://codecov.io/gh/redis/redis-py/pull/2568?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvY2x1c3Rlci5weQ==) | `18.75% <0.00%> (-71.64%)` | :arrow_down: |
| [redis/asyncio/parser.py](https://codecov.io/gh/redis/redis-py/pull/2568?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvYXN5bmNpby9wYXJzZXIucHk=) | `17.30% <0.00%> (-63.47%)` | :arrow_down: |
| [redis/commands/cluster.py](https://codecov.io/gh/redis/redis-py/pull/2568?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvY29tbWFuZHMvY2x1c3Rlci5weQ==) | `36.95% <0.00%> (-57.07%)` | :arrow_down: |
| [redis/crc.py](https://codecov.io/gh/redis/redis-py/pull/2568?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvY3JjLnB5) | `45.45% <0.00%> (-54.55%)` | :arrow_down: |
| [tests/mocks.py](https://codecov.io/gh/redis/redis-py/pull/2568?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvbW9ja3MucHk=) | `68.96% <0.00%> (-27.59%)` | :arrow_down: |
| ... and [33 more](https://codecov.io/gh/redis/redis-py/pull/2568?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) | |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis)

</details>

[:umbrella: View full report at Codecov](https://codecov.io/gh/redis/redis-py/pull/2568?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis).   
:loudspeaker: Do you have feedback about the report comment? [Let us know in this issue](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis).


Patch:
diff --git a/CHANGES b/CHANGES
index d89079ba..9d82341a 100644
--- a/CHANGES
+++ b/CHANGES
@@ -1,3 +1,4 @@
+    * Support `.unlink()` in ClusterPipeline
     * Simplify synchronous SocketBuffer state management
     * Fix string cleanse in Redis Graph
     * Make PythonParser resumable in case of error (#2510)
diff --git a/docs/advanced_features.rst b/docs/advanced_features.rst
index 4ad922fe..5fd20c2b 100644
--- a/docs/advanced_features.rst
+++ b/docs/advanced_features.rst
@@ -162,7 +162,7 @@ instance will wait for all the nodes to respond before returning the
 result to the caller. Command responses are returned as a list sorted in
 the same order in which they were sent. Pipelines can be used to
 dramatically increase the throughput of Redis Cluster by significantly
-reducing the the number of network round trips between the client and
+reducing the number of network round trips between the client and
 the server.
 
 .. code:: pycon
diff --git a/docs/examples/pipeline_examples.ipynb b/docs/examples/pipeline_examples.ipynb
index 490d2213..4e20375b 100644
--- a/docs/examples/pipeline_examples.ipynb
+++ b/docs/examples/pipeline_examples.ipynb
@@ -123,7 +123,7 @@
    "cell_type": "markdown",
    "metadata": {},
    "source": [
-    "The responses of the three commands are stored in a list. In the above example, the two first boolean indicates that the the `set` commands were successfull and the last element of the list is the result of the `get(\"a\")` comand."
+    "The responses of the three commands are stored in a list. In the above example, the two first boolean indicates that the `set` commands were successfull and the last element of the list is the result of the `get(\"a\")` comand."
    ]
   },
   {
diff --git a/redis/asyncio/client.py b/redis/asyncio/client.py
index e56fd022..3fc7fad8 100644
--- a/redis/asyncio/client.py
+++ b/redis/asyncio/client.py
@@ -253,6 +253,11 @@ class Redis(
 
         self.response_callbacks = CaseInsensitiveDict(self.__class__.RESPONSE_CALLBACKS)
 
+        # If using a single connection client, we need to lock creation-of and use-of
+        # the client in order to avoid race conditions such as using asyncio.gather
+        # on a set of redis commands
+        self._single_conn_lock = asyncio.Lock()
+
     def __repr__(self):
         return f"{self.__class__.__name__}<{self.connection_pool!r}>"
 
@@ -260,8 +265,10 @@ class Redis(
         return self.initialize().__await__()
 
     async def initialize(self: _RedisT) -> _RedisT:
-        if self.single_connection_client and self.connection is None:
-            self.connection = await self.connection_pool.get_connection("_")
+        if self.single_connection_client:
+            async with self._single_conn_lock:
+                if self.connection is None:
+                    self.connection = await self.connection_pool.get_connection("_")
         return self
 
     def set_response_callback(self, command: str, callback: ResponseCallbackT):
@@ -501,6 +508,8 @@ class Redis(
         command_name = args[0]
         conn = self.connection or await pool.get_connection(command_name, **options)
 
+        if self.single_connection_client:
+            await self._single_conn_lock.acquire()
         try:
             return await conn.retry.call_with_retry(
                 lambda: self._send_command_parse_response(
@@ -509,6 +518,8 @@ class Redis(
                 lambda error: self._disconnect_raise(conn, error),
             )
         finally:
+            if self.single_connection_client:
+                self._single_conn_lock.release()
             if not self.connection:
                 await pool.release(conn)
 
diff --git a/redis/cluster.py b/redis/cluster.py
index 5f730a85..d6dc02d4 100644
--- a/redis/cluster.py
+++ b/redis/cluster.py
@@ -2136,6 +2136,17 @@ class ClusterPipeline(RedisCluster):
 
         return self.execute_command("DEL", names[0])
 
+    def unlink(self, *names):
+        """
+        "Unlink a key specified by ``names``"
+        """
+        if len(names) != 1:
+            raise RedisClusterException(
+                "unlinking multiple keys is not implemented in pipeline command"
+            )
+
+        return self.execute_command("UNLINK", names[0])
+
 
 def block_pipeline_command(name: str) -> Callable[..., Any]:
     """


Test patch:
diff --git a/tests/test_asyncio/test_connection.py b/tests/test_asyncio/test_connection.py
index bf59dbe6..8e4fdac3 100644
--- a/tests/test_asyncio/test_connection.py
+++ b/tests/test_asyncio/test_connection.py
@@ -6,6 +6,7 @@ from unittest.mock import patch
 import pytest
 
 import redis
+from redis.asyncio import Redis
 from redis.asyncio.connection import (
     BaseParser,
     Connection,
@@ -41,6 +42,50 @@ async def test_invalid_response(create_redis):
     await r.connection.disconnect()
 
 
+@pytest.mark.onlynoncluster
+async def test_single_connection():
+    """Test that concurrent requests on a single client are synchronised."""
+    r = Redis(single_connection_client=True)
+
+    init_call_count = 0
+    command_call_count = 0
+    in_use = False
+
+    class Retry_:
+        async def call_with_retry(self, _, __):
+            # If we remove the single-client lock, this error gets raised as two
+            # coroutines will be vying for the `in_use` flag due to the two
+            # asymmetric sleep calls
+            nonlocal command_call_count
+            nonlocal in_use
+            if in_use is True:
+                raise ValueError("Commands should be executed one at a time.")
+            in_use = True
+            await asyncio.sleep(0.01)
+            command_call_count += 1
+            await asyncio.sleep(0.03)
+            in_use = False
+            return "foo"
+
+    mock_conn = mock.MagicMock()
+    mock_conn.retry = Retry_()
+
+    async def get_conn(_):
+        # Validate only one client is created in single-client mode when
+        # concurrent requests are made
+        nonlocal init_call_count
+        await asyncio.sleep(0.01)
+        init_call_count += 1
+        return mock_conn
+
+    with mock.patch.object(r.connection_pool, "get_connection", get_conn):
+        with mock.patch.object(r.connection_pool, "release"):
+            await asyncio.gather(r.set("a", "b"), r.set("c", "d"))
+
+    assert init_call_count == 1
+    assert command_call_count == 2
+
+
 @skip_if_server_version_lt("4.0.0")
 @pytest.mark.redismod
 @pytest.mark.onlynoncluster
diff --git a/tests/test_cluster.py b/tests/test_cluster.py
index da6a8e4b..1bf57a35 100644
--- a/tests/test_cluster.py
+++ b/tests/test_cluster.py
@@ -2703,6 +2703,25 @@ class TestClusterPipeline:
             with pytest.raises(RedisClusterException):
                 pipe.delete("a", "b")
 
+    def test_unlink_single(self, r):
+        """
+        Test a single unlink operation
+        """
+        r["a"] = 1
+        with r.pipeline(transaction=False) as pipe:
+            pipe.unlink("a")
+            assert pipe.execute() == [1]
+
+    def test_multi_unlink_unsupported(self, r):
+        """
+        Test that multi unlink operation is unsupported
+        """
+        with r.pipeline(transaction=False) as pipe:
+            r["a"] = 1
+            r["b"] = 2
+            with pytest.raises(RedisClusterException):
+                pipe.unlink("a", "b")
+
     def test_brpoplpush_disabled(self, r):
         """
         Test that brpoplpush is disabled for ClusterPipeline