Add max_inactive_connection_lifetime parameter to Pool.

asyncpg/_testbase.py

@@ -157,6 +157,7 @@ def create_pool(dsn=None, *,
  min_size=10,
  max_size=10,
  max_queries=50000,
+ max_inactive_connection_lifetime=60.0,
  setup=None,
  init=None,
  loop=None,
@@ -166,6 +167,7 @@ def create_pool(dsn=None, *,
  dsn,
  min_size=min_size, max_size=max_size,
  max_queries=max_queries, loop=loop, setup=setup, init=init,
+ max_inactive_connection_lifetime=max_inactive_connection_lifetime,
  **connect_kwargs)
 
 

asyncpg/pool.py

@@ -86,19 +86,24 @@ class PoolConnectionHolder:
 
  __slots__ = ('_con', '_pool', '_loop',
  '_connect_args', '_connect_kwargs',
- '_max_queries', '_setup', '_init')
+ '_max_queries', '_setup', '_init',
+ '_max_inactive_time', '_in_use',
+ '_inactive_callback')
 
  def __init__(self, pool, *, connect_args, connect_kwargs,
- max_queries, setup, init):
+ max_queries, setup, init, max_inactive_time):
 
  self._pool = pool
  self._con = None
 
  self._connect_args = connect_args
  self._connect_kwargs = connect_kwargs
  self._max_queries = max_queries
+ self._max_inactive_time = max_inactive_time
  self._setup = setup
  self._init = init
+ self._inactive_callback = None
+ self._in_use = False
 
  async def connect(self):
  assert self._con is None
@@ -134,6 +139,8 @@ async def acquire(self) -> PoolConnectionProxy:
  if self._con is None:
  await self.connect()
 
+ self._maybe_cancel_inactive_callback()
+
  proxy = PoolConnectionProxy(self, self._con)
 
  if self._setup is not None:
@@ -154,9 +161,13 @@ async def acquire(self) -> PoolConnectionProxy:
  self._con = None
  raise ex
 
+ self._in_use = True
  return proxy
 
  async def release(self):
+ assert self._in_use
+ self._in_use = False
+
  if self._con.is_closed():
  self._con = None
 
@@ -181,7 +192,13 @@ async def release(self):
  self._con = None
  raise ex
 
+ assert self._inactive_callback is None
+ if self._max_inactive_time and self._con is not None:
+ self._inactive_callback = self._pool._loop.call_later(
+ self._max_inactive_time, self._deactivate_connection)
+
  async def close(self):
+ self._maybe_cancel_inactive_callback()
  if self._con is None:
  return
  if self._con.is_closed():
@@ -194,6 +211,7 @@ async def close(self):
  self._con = None
 
  def terminate(self):
+ self._maybe_cancel_inactive_callback()
  if self._con is None:
  return
  if self._con.is_closed():
@@ -205,6 +223,18 @@ def terminate(self):
  finally:
  self._con = None
 
+ def _maybe_cancel_inactive_callback(self):
+ if self._inactive_callback is not None:
+ self._inactive_callback.cancel()
+ self._inactive_callback = None
+
+ def _deactivate_connection(self):
+ assert not self._in_use
+ if self._con is None or self._con.is_closed():
+ return
+ self._con.terminate()
+ self._con = None
+
 
 class Pool:
  """A connection pool.
@@ -225,6 +255,7 @@ def __init__(self, *connect_args,
  min_size,
  max_size,
  max_queries,
+ max_inactive_connection_lifetime,
  setup,
  init,
  loop,
@@ -247,6 +278,11 @@ def __init__(self, *connect_args,
  if max_queries <= 0:
  raise ValueError('max_queries is expected to be greater than zero')
 
+ if max_inactive_connection_lifetime < 0:
+ raise ValueError(
+ 'max_inactive_connection_lifetime is expected to be greater '
+ 'or equal to zero')
+
  self._minsize = min_size
  self._maxsize = max_size
 
@@ -265,6 +301,7 @@ def __init__(self, *connect_args,
  connect_args=connect_args,
  connect_kwargs=connect_kwargs,
  max_queries=max_queries,
+ max_inactive_time=max_inactive_connection_lifetime,
  setup=setup,
  init=init)
 
@@ -511,6 +548,7 @@ def create_pool(dsn=None, *,
  min_size=10,
  max_size=10,
  max_queries=50000,
+ max_inactive_connection_lifetime=300.0,
  setup=None,
  init=None,
  loop=None,
@@ -548,6 +586,9 @@ def create_pool(dsn=None, *,
  :param int max_size: Max number of connections in the pool.
  :param int max_queries: Number of queries after a connection is closed
  and replaced with a new connection.
+ :param float max_inactive_connection_lifetime:
+ Number of seconds after which inactive connections in the
+ pool will be closed. Pass ``0`` to disable this mechanism.
  :param coroutine setup: A coroutine to prepare a connection right before
  it is returned from :meth:`~pool.Pool.acquire`.
  An example use case would be to automatically
@@ -567,7 +608,9 @@ def create_pool(dsn=None, *,
  An :exc:`~asyncpg.exceptions.InterfaceError` will be raised on any
  attempted operation on a released connection.
  """
- return Pool(dsn,
- min_size=min_size, max_size=max_size,
- max_queries=max_queries, loop=loop, setup=setup, init=init,
- **connect_kwargs)
+ return Pool(
+ dsn,
+ min_size=min_size, max_size=max_size,
+ max_queries=max_queries, loop=loop, setup=setup, init=init,
+ max_inactive_connection_lifetime=max_inactive_connection_lifetime,
+ **connect_kwargs)

tests/test_pool.py

@@ -11,6 +11,7 @@
 import os
 import platform
 import random
+import time
 import unittest
 
 from asyncpg import _testbase as tb
@@ -457,6 +458,99 @@ async def worker(pool):
  finally:
  await pool.execute('DROP TABLE exmany')
 
+ async def test_pool_max_inactive_time_01(self):
+ async with self.create_pool(
+ database='postgres', min_size=1, max_size=1,
+ max_inactive_connection_lifetime=0.1) as pool:
+
+ # Test that it's OK if a query takes longer time to execute
+ # than `max_inactive_connection_lifetime`.
+
+ con = pool._holders[0]._con
+
+ for _ in range(3):
+ await pool.execute('SELECT pg_sleep(0.5)')
+ self.assertIs(pool._holders[0]._con, con)
+
+ self.assertEqual(
+ await pool.execute('SELECT 1::int'),
+ 'SELECT 1')
+ self.assertIs(pool._holders[0]._con, con)
+
+ async def test_pool_max_inactive_time_02(self):
+ async with self.create_pool(
+ database='postgres', min_size=1, max_size=1,
+ max_inactive_connection_lifetime=0.5) as pool:
+
+ # Test that we have a new connection after pool not
+ # being used longer than `max_inactive_connection_lifetime`.
+
+ con = pool._holders[0]._con
+
+ self.assertEqual(
+ await pool.execute('SELECT 1::int'),
+ 'SELECT 1')
+ self.assertIs(pool._holders[0]._con, con)
+
+ await asyncio.sleep(1, loop=self.loop)
+ self.assertIs(pool._holders[0]._con, None)
+
+ self.assertEqual(
+ await pool.execute('SELECT 1::int'),
+ 'SELECT 1')
+ self.assertIsNot(pool._holders[0]._con, con)
+
+ async def test_pool_max_inactive_time_03(self):
+ async with self.create_pool(
+ database='postgres', min_size=1, max_size=1,
+ max_inactive_connection_lifetime=1) as pool:
+
+ # Test that we start counting inactive time *after*
+ # the connection is being released back to the pool.
+
+ con = pool._holders[0]._con
+
+ await pool.execute('SELECT pg_sleep(0.5)')
+ await asyncio.sleep(0.6, loop=self.loop)
+
+ self.assertIs(pool._holders[0]._con, con)
+
+ self.assertEqual(
+ await pool.execute('SELECT 1::int'),
+ 'SELECT 1')
+ self.assertIs(pool._holders[0]._con, con)
+
+ async def test_pool_max_inactive_time_04(self):
+ # Chaos test for max_inactive_connection_lifetime.
+ DURATION = 2.0
+ START = time.monotonic()
+ N = 0
+
+ async def worker(pool):
+ nonlocal N
+ await asyncio.sleep(random.random() / 10 + 0.1, loop=self.loop)
+ async with pool.acquire() as con:
+ if random.random() > 0.5:
+ await con.execute('SELECT pg_sleep({:.2f})'.format(
+ random.random() / 10))
+ self.assertEqual(
+ await con.fetchval('SELECT 42::int'),
+ 42)
+
+ if time.monotonic() - START < DURATION:
+ await worker(pool)
+
+ N += 1
+
+ async with self.create_pool(
+ database='postgres', min_size=10, max_size=30,
+ max_inactive_connection_lifetime=0.1) as pool:
+
+ workers = [worker(pool) for _ in range(50)]
+ await asyncio.gather(*workers, loop=self.loop)
+
+ self.assertGreater(N, 50)
+
 
 @unittest.skipIf(os.environ.get('PGHOST'), 'using remote cluster for testing')
 class TestHostStandby(tb.ConnectedTestCase):