Raise a clear error if there's a race in pool intialization.

Raise an InterfaceError if:

* Two or more tasks try to initialize the same Pool object concurrently;

* A pool method, such as fetchval(), is called while the Pool is being
  initialized.

See also issue #320.

Author: 1st1

asyncpg/pool.py

@@ -305,8 +305,8 @@ class Pool:
     __slots__ = ('_queue', '_loop', '_minsize', '_maxsize',
                  '_init', '_connect_args', '_connect_kwargs',
                  '_working_addr', '_working_config', '_working_params',
-                 '_holders', '_initialized', '_closing', '_closed',
-                 '_connection_class', '_generation')
+                 '_holders', '_initialized', '_initializing', '_closing',
+                 '_closed', '_connection_class', '_generation')
 
     def __init__(self, *connect_args,
                  min_size,
@@ -359,6 +359,7 @@ def __init__(self, *connect_args,
 
         self._holders = []
         self._initialized = False
+        self._initializing = False
         self._queue = asyncio.LifoQueue(maxsize=self._maxsize, loop=self._loop)
 
         self._working_addr = None
@@ -387,9 +388,20 @@ def __init__(self, *connect_args,
     async def _async__init__(self):
         if self._initialized:
             return
+        if self._initializing:
+            raise exceptions.InterfaceError(
+                'pool is being initialized in another task')
         if self._closed:
             raise exceptions.InterfaceError('pool is closed')
+        self._initializing = True
+        try:
+            await self._initialize()
+            return self
+        finally:
+            self._initializing = False
+            self._initialized = True
 
+    async def _initialize(self):
         if self._minsize:
             # Since we use a LIFO queue, the first items in the queue will be
             # the last ones in `self._holders`.  We want to pre-connect the
@@ -412,9 +424,6 @@ async def _async__init__(self):
 
                 await asyncio.gather(*connect_tasks, loop=self._loop)
 
-        self._initialized = True
-        return self
-
     def set_connect_args(self, dsn=None, **connect_kwargs):
         r"""Set the new connection arguments for this pool.
 
@@ -703,6 +712,11 @@ async def expire_connections(self):
 
     def _check_init(self):
         if not self._initialized:
+            if self._initializing:
+                raise exceptions.InterfaceError(
+                    'pool is being initialized, but not yet ready: '
+                    'likely there is a race between creating a pool and '
+                    'using it')
             raise exceptions.InterfaceError('pool is not initialized')
         if self._closed:
             raise exceptions.InterfaceError('pool is closed')

tests/test_pool.py

@@ -852,6 +852,35 @@ async def test_pool_set_connection_args(self):
         self.assertEqual(con.get_settings().application_name,
                          'set_conn_args_test_2')
 
+    async def test_pool_init_race(self):
+        pool = self.create_pool(database='postgres', min_size=1, max_size=1)
+
+        t1 = asyncio.ensure_future(pool, loop=self.loop)
+        t2 = asyncio.ensure_future(pool, loop=self.loop)
+
+        await t1
+        with self.assertRaisesRegex(
+                asyncpg.InterfaceError,
+                r'pool is being initialized in another task'):
+            await t2
+
+        await pool.close()
+
+    async def test_pool_init_and_use_race(self):
+        pool = self.create_pool(database='postgres', min_size=1, max_size=1)
+
+        pool_task = asyncio.ensure_future(pool, loop=self.loop)
+        await asyncio.sleep(0, loop=self.loop)
+
+        with self.assertRaisesRegex(
+                asyncpg.InterfaceError,
+                r'being initialized, but not yet ready'):
+
+            await pool.fetchval('SELECT 1')
+
+        await pool_task
+        await pool.close()
+
 
 @unittest.skipIf(os.environ.get('PGHOST'), 'using remote cluster for testing')
 class TestHotStandby(tb.ClusterTestCase):