ensure connection is released

aiopg/pool.py

@@ -189,8 +189,7 @@ def _fill_free_pool(self, override_min):
             conn = self._free[-1]
             if conn.closed:
                 self._free.pop()
-            elif self._recycle > -1 \
-                    and self._loop.time() - conn.last_usage > self._recycle:
+            elif -1 < self._recycle < self._loop.time() - conn.last_usage:
                 conn.close()
                 self._free.pop()
             else:

aiopg/utils.py

@@ -1,5 +1,6 @@
 import asyncio
 import sys
+import psycopg2
 
 PY_35 = sys.version_info >= (3, 5)
 PY_352 = sys.version_info >= (3, 5, 2)
@@ -142,24 +143,18 @@ def __aexit__(self, exc_type, exc, tb):
 
 
 class _PoolAcquireContextManager(_ContextManager):
-    __slots__ = ('_coro', '_conn', '_pool')
+    __slots__ = ('_coro', '_obj', '_pool')
 
     def __init__(self, coro, pool):
-        self._coro = coro
-        self._conn = None
+        super().__init__(coro)
         self._pool = pool
 
     if PY_35:
-        @asyncio.coroutine
-        def __aenter__(self):
-            self._conn = yield from self._coro
-            return self._conn
-
         @asyncio.coroutine
         def __aexit__(self, exc_type, exc, tb):
-            yield from self._pool.release(self._conn)
+            yield from self._pool.release(self._obj)
             self._pool = None
-            self._conn = None
+            self._obj = None
 
 
 class _PoolConnectionContextManager:
@@ -238,11 +233,19 @@ def __enter__(self):
     def __exit__(self, *args):
         try:
             self._cur.close()
-            self._pool.release(self._conn)
+        except psycopg2.ProgrammingError:
+            # seen instances where the cursor fails to close:
+            #   https://github.com/aio-libs/aiopg/issues/364
+            # We close it here so we don't return a bad connection to the pool
+            self._conn.close()
+            raise
         finally:
-            self._pool = None
-            self._conn = None
-            self._cur = None
+            try:
+                self._pool.release(self._conn)
+            finally:
+                self._pool = None
+                self._conn = None
+                self._cur = None
 
 
 if not PY_35:

tests/conftest.py

@@ -12,6 +12,7 @@
 import uuid
 import warnings
 
+
 from docker import APIClient
 
 import aiopg
@@ -127,41 +128,62 @@ def pytest_generate_tests(metafunc):
 def pg_server(unused_port, docker, session_id, pg_tag, request):
     if not request.config.option.no_pull:
         docker.pull('postgres:{}'.format(pg_tag))
-    container = docker.create_container(
+
+    container_args = dict(
         image='postgres:{}'.format(pg_tag),
         name='aiopg-test-server-{}-{}'.format(pg_tag, session_id),
         ports=[5432],
         detach=True,
     )
-    docker.start(container=container['Id'])
-    inspection = docker.inspect_container(container['Id'])
-    host = inspection['NetworkSettings']['IPAddress']
-    pg_params = dict(database='postgres',
-                     user='postgres',
-                     password='mysecretpassword',
-                     host=host,
-                     port=5432)
-    delay = 0.001
-    for i in range(100):
-        try:
-            conn = psycopg2.connect(**pg_params)
-            cur = conn.cursor()
-            cur.execute("CREATE EXTENSION hstore;")
-            cur.close()
-            conn.close()
-            break
-        except psycopg2.Error:
-            time.sleep(delay)
-            delay *= 2
-    else:
-        pytest.fail("Cannot start postgres server")
-    container['host'] = host
-    container['port'] = 5432
-    container['pg_params'] = pg_params
-    yield container
-
-    docker.kill(container=container['Id'])
-    docker.remove_container(container['Id'])
+
+    is_darwin = sys.platform == "darwin"
+
+    # bound IPs do not work on OSX
+    host = "127.0.0.1"
+    host_port = 5432
+
+    if is_darwin:
+        host_port = 55432
+        container_args['host_config'] = docker.create_host_config(port_bindings={5432: (host, host_port)})
+
+    container = docker.create_container(**container_args)
+
+    try:
+        docker.start(container=container['Id'])
+
+        # This does not work on OSX
+        if not is_darwin:
+            inspection = docker.inspect_container(container['Id'])
+            host = inspection['NetworkSettings']['IPAddress']
+
+        server_params = dict(database='postgres',
+                             user='postgres',
+                             password='mysecretpassword',
+                             host=host,
+                             port=host_port)
+        delay = 0.001
+        for i in range(100):
+            try:
+                conn = psycopg2.connect(**server_params)
+                cur = conn.cursor()
+                cur.execute("CREATE EXTENSION hstore;")
+                cur.close()
+                conn.close()
+                break
+            except psycopg2.Error:
+                time.sleep(delay)
+                delay *= 2
+        else:
+            pytest.fail("Cannot start postgres server")
+
+        container['host'] = host
+        container['port'] = host_port
+        container['pg_params'] = server_params
+
+        yield container
+    finally:
+        docker.kill(container=container['Id'])
+        docker.remove_container(container['Id'])
 
 
 @pytest.fixture

tests/pep492/test_async_await.py

@@ -53,6 +53,24 @@ async def test_cursor_create_with_context_manager(make_connection):
     assert cursor.closed
 
 
+@asyncio.coroutine
+async def test_pool_context_manager_timeout(pg_params, loop: asyncio.BaseEventLoop):
+    async with aiopg.create_pool(loop=loop, **pg_params, minsize=1, maxsize=1) as pool:
+        cursor_ctx = await pool.cursor()
+        with cursor_ctx as cursor:
+            hung_task = cursor.execute('SELECT pg_sleep(1000);')
+            # start task
+            fut = loop.create_task(hung_task)
+
+            # sleep for a bit so it gets going
+            await asyncio.sleep(1)
+
+            print()
+
+    assert cursor.closed
+    assert pool.closed
+
+
 @asyncio.coroutine
 async def test_cursor_with_context_manager(make_connection):
     conn = await make_connection()