Address concurrency issue on get_or_set

Pulled in fix (with some changes) from sebleier#179

Author: victorvess

redis_cache/backends/base.py

@@ -1,4 +1,6 @@
 from functools import wraps
+import time
+import uuid
 
 from django.core.cache.backends.base import (
     BaseCache, DEFAULT_TIMEOUT, InvalidCacheBackendError,
@@ -442,34 +444,42 @@ def get_or_set(
         lock_key = "__lock__" + key
         fresh_key = "__fresh__" + key
 
-        is_fresh = self._get(client, fresh_key)
         value = self._get(client, key)
-
+        is_fresh = self._get(client, fresh_key)
         if is_fresh:
             return value
 
-        timeout = self.get_timeout(timeout)
-        lock = self.lock(lock_key, timeout=lock_timeout)
-
-        acquired = lock.acquire(blocking=False)
+        fresh_timeout = self.get_timeout(timeout)
+        key_timeout = None if stale_cache_timeout is None else fresh_timeout + stale_cache_timeout
 
-        if acquired:
-            try:
-                value = func()
-            except Exception:
-                raise
+        token = uuid.uuid1().hex
+        lock = self.lock(lock_key, timeout=lock_timeout)
+        acquired = lock.acquire(blocking=False, token=token)
+
+        while True:
+            if acquired:
+                try:
+                    value = func()
+                except Exception:
+                    raise
+                else:
+                    pipeline = client.pipeline()
+                    pipeline.set(key, self.prep_value(value), key_timeout)
+                    pipeline.set(fresh_key, 1, fresh_timeout)
+                    pipeline.execute()
+                    return value
+                finally:
+                    lock.release()
+            elif value is None:
+                time.sleep(lock.sleep)
+                value = self._get(client, key)
+                if value is None:
+                    # If there is no value present yet, try to acquire the
+                    # lock again (maybe the other thread died for some reason
+                    # and we should try to compute the value instead).
+                    acquired = lock.acquire(blocking=False, token=token)
             else:
-                key_timeout = (
-                    None if stale_cache_timeout is None else timeout + stale_cache_timeout
-                )
-                pipeline = client.pipeline()
-                pipeline.set(key, self.prep_value(value), key_timeout)
-                pipeline.set(fresh_key, 1, timeout)
-                pipeline.execute()
-            finally:
-                lock.release()
-
-        return value
+                return value
 
     def _reinsert_keys(self, client):
         keys = list(client.scan_iter(match='*'))

tests/testapp/tests/base_tests.py

@@ -511,6 +511,10 @@ def expensive_function():
         self.assertEqual(expensive_function.num_calls, 2)
         self.assertEqual(value, 42)
 
+    def test_get_or_set_with_none_value(self):
+        value = self.cache.get_or_set('key', lambda: None, 1, None, 1)
+        assert value is None
+
     def test_get_or_set_serving_from_stale_value(self):
 
         def expensive_function(x):
@@ -537,6 +541,8 @@ def thread_worker(thread_id, return_value, timeout, lock_timeout, stale_cache_ti
         thread_2 = threading.Thread(target=thread_worker, args=(2, 'c', 1, None, 1))
         thread_3 = threading.Thread(target=thread_worker, args=(3, 'd', 1, None, 1))
         thread_4 = threading.Thread(target=thread_worker, args=(4, 'e', 1, None, 1))
+        thread_5 = threading.Thread(target=thread_worker, args=(5, None, 1, None, 1))
+        thread_6 = threading.Thread(target=thread_worker, args=(6, 'g', 1, None, 1))
 
         # First thread should complete and return its value
         thread_0.start()  # t = 0, valid from t = .5 - 1.5, stale from t = 1.5 - 2.5
@@ -556,19 +562,31 @@ def thread_worker(thread_id, return_value, timeout, lock_timeout, stale_cache_ti
         # before the first thread's stale cache has expired.
         time.sleep(.25)  # t = 2
         thread_4.start()
+        # Sixth thread will start after fourth thread's value is no longer current, explicitly sets
+        # none as the result
+        # valid from t = 4 - 5, stale from t = 5 - 6
+        time.sleep(1.5)  # t = 3.5
+        thread_5.start()
+        # Seventh thread will start after sixth value has cached - should get None
+        time.sleep(1)   # t = 4.5
+        thread_6.start()
 
         thread_0.join()
         thread_1.join()
         thread_2.join()
         thread_3.join()
         thread_4.join()
+        thread_5.join()
+        thread_6.join()
 
         self.assertEqual(results, {
             0: 'a',
-            1: None,
+            1: 'a',
             2: 'a',
             3: 'd',
-            4: 'a'
+            4: 'a',
+            5: None,
+            6: None,
         })
 
     def assertMaxConnection(self, cache, max_num):