get_redis_pool lock to avoid multiple pool being created

Author: samuelcolvin

HISTORY.rst

@@ -3,7 +3,12 @@
 History
 -------
 
-v0.10.2 (2017-XX-XX)
+v0.10.3 (2017-08-19)
+....................
+* fix bug with ``RedisMixin.get_redis_pool`` creating multiple queues
+* tweak drain logs
+
+v0.10.2 (2017-08-17)
 ....................
 * only save job on task in drain if re-enqueuing
 * add semaphore timeout to drains

arq/drain.py

@@ -17,6 +17,7 @@
 
 __all__ = ['Drain']
 
+# these loggers could do with more sensible names
 work_logger = logging.getLogger('arq.work')
 jobs_logger = logging.getLogger('arq.jobs')
 
@@ -97,6 +98,7 @@ async def iter(self, *raw_queues: bytes, pop_timeout=1):
             await self.redis.rpush(quit_queue, b'1')
             raw_queues = tuple(raw_queues) + (quit_queue,)
         while True:
+            work_logger.debug('task semaphore locked: %r', self.task_semaphore.locked())
             try:
                 with timeout(self.semaphore_timeout):
                     await self.task_semaphore.acquire()
@@ -115,7 +117,7 @@ async def iter(self, *raw_queues: bytes, pop_timeout=1):
             if self.burst_mode and raw_queue == quit_queue:
                 work_logger.debug('got job from the quit queue, stopping')
                 break
-            work_logger.debug('jobs in progress %d', self.jobs_in_progress)
+            work_logger.debug('yielding job, jobs in progress %d', self.jobs_in_progress)
             yield raw_queue, raw_data
 
     def add(self, coro, job, re_enqueue=False):

arq/utils.py

@@ -66,6 +66,7 @@ def __init__(self, *,
         self.loop = loop or getattr(self, 'loop', None) or asyncio.get_event_loop()
         self.redis_settings = redis_settings or getattr(self, 'redis_settings', None) or RedisSettings()
         self._redis_pool = existing_pool
+        self._create_pool_lock = asyncio.Lock(loop=self.loop)
 
     async def create_redis_pool(self, *, _retry=0) -> RedisPool:
         """
@@ -93,8 +94,9 @@ async def get_redis_pool(self) -> RedisPool:
         """
         Get the redis pool, if a pool is already initialised it's returned, else one is crated.
         """
-        if self._redis_pool is None:
-            self._redis_pool = await self.create_redis_pool()
+        async with self._create_pool_lock:
+            if self._redis_pool is None:
+                self._redis_pool = await self.create_redis_pool()
         return self._redis_pool
 
     async def get_redis_conn(self):

arq/version.py

@@ -2,4 +2,4 @@
 
 __all__ = ['VERSION']
 
-VERSION = StrictVersion('0.10.2')
+VERSION = StrictVersion('0.10.3')

tests/test_main.py

@@ -75,10 +75,13 @@ async def test_dispatch_work(tmpworkdir, loop, caplog, redis_conn):
             'recording health: <date time2> j_complete=0 j_failed=0 j_timedout=0 j_ongoing=0 q_high=1 q_dft=1 q_low=0\n'
             'starting main blpop loop\n'
             'populating quit queue to prompt exit: arq:quit-<random>\n'
-            'jobs in progress 1\n'
+            'task semaphore locked: False\n'
+            'yielding job, jobs in progress 1\n'
             'scheduling job <Job __id__ MockRedisDemoActor.high_add_numbers(3, 4, c=5) on high>, re-enqueue: False\n'
-            'jobs in progress 2\n'
+            'task semaphore locked: False\n'
+            'yielding job, jobs in progress 2\n'
             'scheduling job <Job __id__ MockRedisDemoActor.add_numbers(1, 2) on dft>, re-enqueue: False\n'
+            'task semaphore locked: False\n'
             'got job from the quit queue, stopping\n'
             'drain waiting 5.0s for 2 tasks to finish\n'
             'high queued  0.0XXs → __id__ MockRedisDemoActor.high_add_numbers(3, 4, c=5)\n'