Fix async connection pool lock performance issue

Move ensure_connection() outside the lock in ConnectionPool.get_connection() to prevent lock contention during socket connection and TLS handshake operations. This matches the pattern already used in BlockingConnectionPool and significantly improves performance under high load.

Co-authored-by: petyaslavova <194077574+petyaslavova@users.noreply.github.com>

Author: Copilot

redis/asyncio/connection.py

@@ -1215,16 +1215,17 @@ def can_get_connection(self) -> bool:
         version="5.3.0",
     )
     async def get_connection(self, command_name=None, *keys, **options):
+        """Get a connected connection from the pool"""
         async with self._lock:
-            """Get a connected connection from the pool"""
             connection = self.get_available_connection()
-            try:
-                await self.ensure_connection(connection)
-            except BaseException:
-                await self.release(connection)
-                raise
 
-        return connection
+        # We now perform the connection check outside of the lock.
+        try:
+            await self.ensure_connection(connection)
+            return connection
+        except BaseException:
+            await self.release(connection)
+            raise
 
     def get_available_connection(self):
         """Get a connection from the pool, without making sure it is connected"""

tests/test_asyncio/test_connection_pool.py

@@ -222,6 +222,93 @@ async def test_pool_disconnect(self, master_host):
             await pool.disconnect(inuse_connections=False)
             assert conn.is_connected
 
+    async def test_lock_not_held_during_connection_establishment(self):
+        """
+        Test that the connection pool lock is not held during the
+        ensure_connection call, which involves socket connection and handshake.
+        This is important for performance under high load.
+        """
+
+        class SlowConnectConnection(DummyConnection):
+            """Connection that simulates slow connection establishment"""
+
+            def __init__(self, **kwargs):
+                super().__init__(**kwargs)
+                self.connect_called = False
+                self.lock_held_during_connect = None
+
+            async def connect(self):
+                self.connect_called = True
+                # Check if the pool's lock is held during connection
+                pool = self.kwargs.get("_pool")
+                if pool:
+                    self.lock_held_during_connect = pool._lock.locked()
+                # Simulate slow connection
+                await asyncio.sleep(0.01)
+                self._connected = True
+
+        async with self.get_pool(
+            connection_class=SlowConnectConnection,
+            connection_kwargs={"_pool": None},
+        ) as pool:
+            # Pass the pool to the connection so it can check lock status
+            pool.connection_kwargs["_pool"] = pool
+
+            # Get a connection - this should call connect() outside the lock
+            connection = await pool.get_connection()
+
+            # Verify connect was called
+            assert connection.connect_called
+
+            # Verify the lock was NOT held during connect
+            assert (
+                connection.lock_held_during_connect is False
+            ), "Lock should not be held during connection establishment"
+
+            await pool.release(connection)
+
+    async def test_concurrent_connection_acquisition_performance(self):
+        """
+        Test that multiple concurrent connection acquisitions don't block
+        each other during connection establishment.
+        """
+
+        class SlowConnectConnection(DummyConnection):
+            """Connection that simulates slow connection establishment"""
+
+            async def connect(self):
+                # Simulate slow connection (e.g., network latency, TLS handshake)
+                await asyncio.sleep(0.05)
+                self._connected = True
+
+        async with self.get_pool(
+            connection_class=SlowConnectConnection, max_connections=10
+        ) as pool:
+            # Start acquiring multiple connections concurrently
+            start_time = asyncio.get_running_loop().time()
+
+            # Try to get 3 connections concurrently
+            connections = await asyncio.gather(
+                pool.get_connection(),
+                pool.get_connection(),
+                pool.get_connection(),
+            )
+
+            elapsed_time = asyncio.get_running_loop().time() - start_time
+
+            # With proper lock handling, these should complete mostly in parallel
+            # If the lock was held during connect(), it would take 3 * 0.05 = 0.15s
+            # With lock only during pop, it should take ~0.05s (connections in parallel)
+            # We allow some overhead, so check it's less than 0.12s
+            assert elapsed_time < 0.12, (
+                f"Concurrent connections took {elapsed_time}s, "
+                f"suggesting lock was held during connection establishment"
+            )
+
+            # Clean up
+            for conn in connections:
+                await pool.release(conn)
+
 
 class TestBlockingConnectionPool:
     @asynccontextmanager