fix(asyncio): keep pipeline packing off the event loop

redis/asyncio/client.py

@@ -1928,8 +1928,9 @@ async def _execute_transaction(  # noqa: C901
         pre: CommandT = (("MULTI",), {})
         post: CommandT = (("EXEC",), {})
         cmds = (pre, *commands, post)
-        all_cmds = connection.pack_commands(
-            args for args, options in cmds if EMPTY_RESPONSE not in options
+        all_cmds = await self._pack_commands(
+            connection,
+            (args for args, options in cmds if EMPTY_RESPONSE not in options),
         )
         await connection.send_packed_command(all_cmds)
         errors = []
@@ -2004,7 +2005,7 @@ async def _execute_pipeline(
         self, connection: Connection, commands: CommandStackT, raise_on_error: bool
     ):
         # build up all commands into a single request to increase network perf
-        all_cmds = connection.pack_commands([args for args, _ in commands])
+        all_cmds = await self._pack_commands(connection, [args for args, _ in commands])
         await connection.send_packed_command(all_cmds)
 
         response = []
@@ -2020,6 +2021,12 @@ async def _execute_pipeline(
             self.raise_first_error(commands, response)
         return response
 
+    async def _pack_commands(
+        self, connection: Connection, commands: Iterable[Iterable[EncodableT]]
+    ) -> List[bytes]:
+        """Pack commands without blocking the event loop."""
+        return await asyncio.to_thread(connection.pack_commands, commands)
+
     def raise_first_error(self, commands: CommandStackT, response: Iterable[Any]):
         for i, r in enumerate(response):
             if isinstance(r, ResponseError):

tests/test_asyncio/test_pipeline.py

@@ -1,4 +1,5 @@
 import asyncio
+import threading
 from unittest import mock
 
 import pytest
@@ -17,6 +18,65 @@
 
 
 class TestPipeline:
+    @pytest.mark.parametrize("transaction", (False, True))
+    async def test_pipeline_packing_does_not_block_event_loop(self, transaction):
+        pipeline = Pipeline(
+            connection_pool=mock.MagicMock(),
+            response_callbacks={},
+            transaction=transaction,
+            shard_hint=None,
+        )
+        connection = mock.MagicMock()
+        packing_started = threading.Event()
+        release_packing = threading.Event()
+        packing_finished = threading.Event()
+        event_loop_ticks_during_packing = 0
+
+        def pack_commands(_commands):
+            packing_started.set()
+            release_packing.wait(timeout=1)
+            packing_finished.set()
+            return []
+
+        connection.pack_commands.side_effect = pack_commands
+        connection.send_packed_command = mock.AsyncMock()
+        pipeline.parse_response = mock.AsyncMock(
+            side_effect=[None, None, [True]] if transaction else [True]
+        )
+
+        async def observe_event_loop():
+            nonlocal event_loop_ticks_during_packing
+            while not packing_finished.is_set():
+                await asyncio.sleep(0)
+                if packing_started.is_set() and not packing_finished.is_set():
+                    event_loop_ticks_during_packing += 1
+
+        observer = asyncio.create_task(observe_event_loop())
+        await asyncio.sleep(0)
+        timer = threading.Timer(0.05, release_packing.set)
+        timer.start()
+
+        try:
+            if transaction:
+                result = await asyncio.wait_for(
+                    pipeline._execute_transaction(connection, [(("PING",), {})], True),
+                    timeout=1,
+                )
+            else:
+                result = await asyncio.wait_for(
+                    pipeline._execute_pipeline(connection, [(("PING",), {})], True),
+                    timeout=1,
+                )
+        finally:
+            release_packing.set()
+            observer.cancel()
+            await asyncio.gather(observer, return_exceptions=True)
+            timer.cancel()
+
+        assert packing_started.is_set()
+        assert result == [True]
+        assert event_loop_ticks_during_packing > 0
+
     async def test_pipeline_is_true(self, r):
         """Ensure pipeline instances are not false-y"""
         async with r.pipeline() as pipe: