re implementing 'reusable'

Author: samuelcolvin

arq/worker.py

@@ -60,6 +60,10 @@ class BaseWorker(RedisMixin):
     health_check_interval = 60
     health_check_key = b'arq:health-check'
 
+    #: Mostly used in tests; if true actors and the redis pool will not be closed at the end of run()
+    # allowing reuse of the worker, eg. ``worker.run()`` can be called multiple times.
+    reusable = False
+
     def __init__(self, *,
                  burst: bool=False,
                  shadows: list=None,
@@ -92,7 +96,7 @@ def __init__(self, *,
         signal.signal(signal.SIGTERM, self.handle_sig)
         signal.signal(SIG_PROXY, self.handle_proxy_signal)
         super().__init__(**kwargs)  # type: ignore # TODO
-        self._closing_lock = asyncio.Lock(loop=self.loop)
+        self._shutdown_lock = asyncio.Lock(loop=self.loop)
 
     async def shadow_factory(self) -> list:
         """
@@ -142,6 +146,7 @@ async def run(self):
         Main entry point for the the worker which initialises shadows, checks they look ok then runs ``work`` to
         perform jobs.
         """
+        self._stopped = False
         work_logger.info('Initialising work manager, burst mode: %s', self._burst_mode)
 
         shadows = await self.shadow_factory()
@@ -170,7 +175,7 @@ async def run(self):
         try:
             await self.work()
         finally:
-            await self.close()
+            await self.shutdown()
             if self._task_exception:
                 work_logger.error('Found task exception "%s"', self._task_exception)
                 raise self._task_exception
@@ -241,6 +246,7 @@ async def _check_health(self):
     def check_health(cls, **kwargs):
         """
         Run a health check on the worker return the appropriate exit code.
+
         :return: 0 if successful, 1 if not
         """
         self = cls(**kwargs)
@@ -333,17 +339,19 @@ def handle_execute_exc(cls, started_at, exc, j):
         exc_type = exc.__class__.__name__
         jobs_logger.exception('%-4s ran in%7.3fs ! %s: %s', j.queue, timestamp() - started_at, j, exc_type)
 
-    async def close(self):
-        with await self._closing_lock:
-            if self._closed:
-                return
+    async def shutdown(self):
+        with await self._shutdown_lock:
             if self._pending_tasks:
                 work_logger.info('shutting down worker, waiting for %d jobs to finish', len(self._pending_tasks))
                 await asyncio.wait(self._pending_tasks, loop=self.loop)
             t = (timestamp() - self.start) if self.start else 0
             work_logger.info('shutting down worker after %0.3fs ◆ %d jobs done ◆ %d failed ◆ %d timed out',
                              t, self.jobs_complete, self.jobs_failed, self.jobs_timed_out)
+            if not self.reusable:
+                await self.close()
 
+    async def close(self):
+        if not self._closed:
             if self._shadow_lookup:
                 await asyncio.gather(*[s.close(True) for s in self._shadow_lookup.values()], loop=self.loop)
             await super().close()

docs/index.rst

@@ -67,14 +67,14 @@ The old computer science proverb/joke goes:
 
     There are only two challenges in computer science: cache invalidation, naming things and the n + 1 problem.
 
-*arq* tries to use generally accepted terminology for as much as possible, however "actors" and "shadows" are not so
-standard and bear describing:
+*arq* tries to avoid confusion over what's named what by using generally accepted terminology as much as possible,
+however a few terms (like "actors" and "shadows") are not so standard and bear describing:
 
 An **Actor** is a class with some concurrent methods, you can define and use multiple actors. Actors hold a
-reference to a redis pool for enqueuing are generally singletons.
+reference to a redis pool for enqueuing jobs and are generally singletons.
 
 The **Worker** is the class which is responsible for running jobs for one or more actors. Workers should inherit
-from ``BaseWorker``, your application will generally only have one worker.
+from ``BaseWorker``, your application will generally have just one worker.
 
 Actors are therefore used in two distinctly different modes:
 

docs/usage.rst

@@ -25,41 +25,41 @@ For details on the *arq* CLI::
 Startup & Shutdown coroutines
 .............................
 
-The ``startup`` and ``shutdown`` are provided as a convenient way to run logic as actors start and finish,
+The ``startup`` and ``shutdown`` coroutines are provided as a convenient way to run logic as actors start and finish,
 however it's important to not that these methods **are not called by default when actors are initialised or closed**.
-They are however called when the actor started and closed on the worker, eg. in "shadow" mode, see above.
-In other words: if you need these coroutines to be called when using an actor in your code, that's your responsibility.
+They are however called when the actor was started and closed on the worker, eg. in "shadow" mode, see above.
+In other words: if you need these coroutines to be called when using an actor in your code; that's your responsibility.
 
-For example, in the above code there's no need for ``self.session`` when using the actor in "default" mode, eg. called
-with ``python demo.py``, so neither ``startup`` or ``shutdown`` are called.
+For example, in the above example there's no need for ``self.session`` when using the actor in "default" mode,
+eg. called with ``python demo.py``, so neither ``startup`` or ``shutdown`` are called.
 
 Health checks
 .............
 
 *arq* will automatically record some info about it's current state in redis every ``health_check_interval`` seconds,
 see :attr:`arq.worker.BaseWorker.health_check_interval`. That key/value will expire after ``health_check_interval + 1``
-so you can be sure if the variable exists you can be sure *arq* is alive and kicking (technically you can be sure it
+seconds so you can be sure if the variable exists *arq* is alive and kicking (technically you can be sure it
 was alive and kicking ``health_check_interval`` seconds ago).
 
-You can run a health check with the CLI using (assuming you're using the above example)::
+You can run a health check with the CLI (assuming you're using the above example)::
 
     arq --check demo.py
 
-The command will output the value of the health check if found,
+The command will output the value of the health check if found;
 then exit ``0`` if the key was found and ``1`` if it was not.
 
 A health check value takes the following form::
 
     Feb-20_11:02:40 j_complete=0 j_failed=0 j_timedout=0 j_ongoing=0 q_high=0 q_dft=0 q_low=0
 
-Where the values have the following meaning:
+Where the items have the following meaning:
 
 * ``j_complete`` the number of jobs completed
 * ``j_failed`` the number of jobs which have failed eg. raised an exception
 * ``j_timedout`` the number of jobs which have timed out, eg. exceeded :attr:`arq.worker.BaseWorker.timeout_seconds`
   and been cancelled
-* ``j_ongoing`` the number of jobs currently being performed.
-* ``q_*`` the number of pending jobs in each queue.
+* ``j_ongoing`` the number of jobs currently being performed
+* ``q_*`` the number of pending jobs in each queue
 
 Multiple Queues
 ...............

tests/test_worker.py

@@ -284,6 +284,23 @@ async def test_raise_worker_prepare(redis_conn, actor):
     await worker.close()
 
 
+async def test_reusable_worker(tmpworkdir, redis_conn, actor):
+    worker = Worker(burst=True, loop=actor.loop)
+    worker.reusable = True
+
+    await actor.add_numbers(1, 2)
+    assert not tmpworkdir.join('add_numbers').exists()
+    await worker.run()
+    assert tmpworkdir.join('add_numbers').read() == '3'
+    assert worker.jobs_failed == 0
+
+    await actor.add_numbers(3, 4)
+    await worker.run()
+    assert tmpworkdir.join('add_numbers').read() == '7'
+    assert worker.jobs_failed == 0
+    await worker.close()
+
+
 async def test_startup_shutdown(tmpworkdir, redis_conn, loop):
     worker = StartupWorker(burst=True, loop=loop)