From bca22054df25c9eea4ead88667fc0a74dd092922 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Janne=20R=C3=B6nkk=C3=B6?= <jroo@iki.fi>
Date: Mon, 28 Jan 2019 08:39:56 +0200
Subject: [PATCH] Fix concurrency issue in timeout timer value processing

According to multiprocessing documentation for Value
(https://docs.python.org/3/library/multiprocessing.html#multiprocessing.Value)
reads and writes are protected with lock when the lock argument is True
(the default) or the lock argument is an instance of Lock or RLock. The
documentation states that operations like += are not atomic as that
involves reading and writing.

On the worker side the critical section includes also storing finished task
result because the timeout could happen after the task function has
finished but before the result has been stored and timer.value has been
updated to tell the guard process that the task has been finished.

On the guard side the critical section includes all checks done to see
if the worker has timed out or died and the actual reincarnation function
because the worker could update timer value to -1 (idle) or -2 (recycle)
after the guard has seen timer value 0 (timeout) and is going to terminate
the worker.
---
 django_q/cluster.py | 36 +++++++++++++++++++-----------------
 1 file changed, 19 insertions(+), 17 deletions(-)

diff --git a/django_q/cluster.py b/django_q/cluster.py
index 4a67505..23a9761 100644
--- a/django_q/cluster.py
+++ b/django_q/cluster.py
@@ -209,13 +209,14 @@ class Sentinel(object):
         while not self.stop_event.is_set() or not counter:
             # Check Workers
             for p in self.pool:
-                # Are you alive?
-                if not p.is_alive() or p.timer.value == 0:
-                    self.reincarnate(p)
-                    continue
-                # Decrement timer if work is being done
-                if p.timer.value > 0:
-                    p.timer.value -= cycle
+                with p.timer.get_lock():
+                    # Are you alive?
+                    if not p.is_alive() or p.timer.value == 0:
+                        self.reincarnate(p)
+                        continue
+                    # Decrement timer if work is being done
+                    if p.timer.value > 0:
+                        p.timer.value -= cycle
             # Check Monitor
             if not self.monitor.is_alive():
                 self.reincarnate(self.monitor)
@@ -382,16 +383,17 @@ def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT):
                 result = ('{} : {}'.format(e, traceback.format_exc()), False)
                 if error_reporter:
                     error_reporter.report()
-        # Process result
-        task['result'] = result[0]
-        task['success'] = result[1]
-        task['stopped'] = timezone.now()
-        result_queue.put(task)
-        timer.value = -1  # Idle
-        # Recycle
-        if task_count == Conf.RECYCLE:
-            timer.value = -2  # Recycled
-            break
+        with timer.get_lock():
+            # Process result
+            task['result'] = result[0]
+            task['success'] = result[1]
+            task['stopped'] = timezone.now()
+            result_queue.put(task)
+            timer.value = -1  # Idle
+            # Recycle
+            if task_count == Conf.RECYCLE:
+                timer.value = -2  # Recycled
+                break
     logger.info(_('{} stopped doing work').format(name))