From: Daniele Varrazzo <daniele.varrazzo@gmail.com>
Date: Sat, 26 Mar 2022 18:02:32 +0000 (+0100)
Subject: fix: guard connection._pipeline using the connection lock
X-Git-Tag: 3.1~146^2~8
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=7900ea8c4063e85396d0f53366ed13f6c86c2401;p=thirdparty%2Fpsycopg.git

fix: guard connection._pipeline using the connection lock
---

diff --git a/psycopg/psycopg/connection.py b/psycopg/psycopg/connection.py
index 6c09c1722..1a0b9393a 100644
--- a/psycopg/psycopg/connection.py
+++ b/psycopg/psycopg/connection.py
@@ -870,12 +870,20 @@ class Connection(BaseConnection[Row]):
     @contextmanager
     def pipeline(self) -> Iterator[None]:
         """Context manager to switch the connection into pipeline mode."""
-        if self._pipeline is not None:
-            # calling pipeline recursively is no-op.
+        with self.lock:
+            if self._pipeline is None:
+                # We must enter pipeline mode: create a new one
+                pipeline = self._pipeline = Pipeline(self.pgconn)
+            else:
+                # we are already in pipeline mode: bail out as soon as we
+                # leave the lock block.
+                pipeline = None
+
+        if not pipeline:
+            # No-op re-entered inner pipeline block.
             yield
             return
 
-        pipeline = self._pipeline = Pipeline(self.pgconn)
         try:
             with pipeline:
                 try:
diff --git a/psycopg/psycopg/connection_async.py b/psycopg/psycopg/connection_async.py
index 52709d670..2c0321bb6 100644
--- a/psycopg/psycopg/connection_async.py
+++ b/psycopg/psycopg/connection_async.py
@@ -298,12 +298,20 @@ class AsyncConnection(BaseConnection[Row]):
     @asynccontextmanager
     async def pipeline(self) -> AsyncIterator[None]:
         """Context manager to switch the connection into pipeline mode."""
-        if self._pipeline is not None:
-            # calling pipeline recursively is no-op.
+        async with self.lock:
+            if self._pipeline is None:
+                # We must enter pipeline mode: create a new one
+                pipeline = self._pipeline = AsyncPipeline(self.pgconn)
+            else:
+                # we are already in pipeline mode: bail out as soon as we
+                # leave the lock block.
+                pipeline = None
+
+        if not pipeline:
+            # No-op re-entered inner pipeline block.
             yield
             return
 
-        pipeline = self._pipeline = AsyncPipeline(self.pgconn)
         try:
             async with pipeline:
                 try: