Fix the semaphore accounting in PooledPg

* close() released the slots of the cached connections a second time.
* pooled_connection() leaked a slot when the connection could not be
  established.

Author: Cito

dbutils/pooled_pg.py

@@ -224,9 +224,17 @@ def pooled_connection(self):
         if self._connections and not self._connections.acquire(self._blocking):
             raise TooManyConnectionsError
         try:
-            return self._cache.get_nowait()
-        except Empty:
-            return self.steady_connection()
+            try:
+                return self._cache.get_nowait()
+            except Empty:
+                return self.steady_connection()
+        except Exception:
+            # the connection could not be established, so give back the
+            # reservation that has been made above (note that this must
+            # not happen when the reservation itself has already failed)
+            if self._connections:
+                self._connections.release()
+            raise
 
     def connection(self):
         """Get a steady, cached PostgreSQL connection from the pool."""
@@ -248,13 +256,16 @@ def cache(self, con):
 
     def close(self):
         """Close all connections in the pool."""
+        # Note that the connections in the cache have already given back
+        # their share of the generally allowed connections when they were
+        # returned to the pool, so the semaphore must not be released here.
+        # Connections that are still in use are not closed here, and they
+        # will release their share as usual by calling the cache() method.
         while 1:
             try:
                 con = self._cache.get_nowait()
                 with suppress(Exception):
                     con.close()
-                if self._connections:
-                    self._connections.release()
             except Empty:
                 break
 

docs/changelog.html

@@ -41,6 +41,13 @@ <h2>3.2.0</h2>
 connection had already been returned to the pool.</p></li>
 <li><p>Fixed <span class="docutils literal">pooled_db.SharedDBConnection</span> violating the contract between
 <span class="docutils literal">__eq__</span> and <span class="docutils literal">__hash__</span>; shared connections now compare by identity.</p></li>
+<li><p>Fixed <span class="docutils literal">PooledPg.close()</span> releasing the semaphore for the connections in
+the pool cache, which had already released it when they were returned to
+the pool, so that the pool handed out more connections at the same time
+than allowed by <span class="docutils literal">maxconnections</span> after it had been closed.</p></li>
+<li><p>Fixed <span class="docutils literal">PooledPg</span> not releasing the semaphore when a connection could not
+be established, so that every failed attempt permanently reduced the number
+of connections the pool was willing to hand out.</p></li>
 </ul>
 </section>
 <section id="section-2">

docs/changelog.rst

@@ -33,6 +33,13 @@ Bugfixes:
   connection had already been returned to the pool.
 * Fixed ``pooled_db.SharedDBConnection`` violating the contract between
   ``__eq__`` and ``__hash__``; shared connections now compare by identity.
+* Fixed ``PooledPg.close()`` releasing the semaphore for the connections in
+  the pool cache, which had already released it when they were returned to
+  the pool, so that the pool handed out more connections at the same time
+  than allowed by ``maxconnections`` after it had been closed.
+* Fixed ``PooledPg`` not releasing the semaphore when a connection could not
+  be established, so that every failed attempt permanently reduced the number
+  of connections the pool was willing to hand out.
 
 3.1.2
 =====

tests/test_pooled_pg.py

@@ -13,6 +13,7 @@
 from queue import Empty, Queue
 from threading import Thread
 
+import pg
 import pytest
 
 from dbutils.pooled_pg import (
@@ -122,6 +123,62 @@ def test_close_connection():
     del db
 
 
+def test_close_all():
+    """Check that closing the pool keeps the connections accounted for."""
+    pool = PooledPg(0, 1, 1, False, None, None, False, 'PooledPgTestDB')
+    assert pool._cache.qsize() == 0
+    assert pool._connections._value == 1
+    db = pool.connection()
+    db_con = db._con
+    db.close()
+    # the cached connection has already given back its share of the
+    # generally allowed connections when it was returned to the pool
+    assert pool._cache.qsize() == 1
+    assert pool._connections._value == 1
+    pool.close()
+    # closing the pool discards the cached connection,
+    # but it must not give back its share a second time
+    assert pool._cache.qsize() == 0
+    assert pool._connections._value == 1
+    assert db_con._closed
+    # so the pool still hands out only one connection at a time
+    db = pool.connection()
+    assert db._con is not db_con
+    assert pool._connections._value == 0
+    with pytest.raises(TooManyConnectionsError):
+        pool.connection()
+    # closing the pool again (this also happens when it is deleted)
+    # must not change the accounting while a connection is still in use
+    pool.close()
+    assert pool._connections._value == 0
+    # the connection in use gives back its share when it is returned
+    db.close()
+    assert pool._cache.qsize() == 1
+    assert pool._connections._value == 1
+
+
+def test_connection_error_does_not_use_up_a_connection():
+    """Check that a failing connection attempt is properly accounted for."""
+    pool = PooledPg(0, 0, 1, False, None, None, False, dbname='ok')
+    assert pool._cache.qsize() == 0
+    assert pool._connections._value == 1
+    # the mock database raises an error when the database is named 'error'
+    pool._kwargs['dbname'] = 'error'
+    with pytest.raises(pg.InternalError):
+        pool.connection()
+    pool._kwargs['dbname'] = 'ok'
+    # the failed attempt must not have used up the allowed connection
+    assert pool._connections._value == 1
+    db = pool.connection()
+    assert db.dbname == 'ok'
+    assert pool._connections._value == 0
+    # but the connection that could be established still counts
+    with pytest.raises(TooManyConnectionsError):
+        pool.connection()
+    db.close()
+    assert pool._connections._value == 1
+
+
 @pytest.mark.parametrize(("mincached", "maxcached", "cached", "rounds"), [
     # every round takes the given number of connections out of the pool
     # and expects the given number of cached connections after closing them