Add option for thread local Redis locks

Added option to the Redis backend
:paramref:`.RedisBackend.thread_local_lock`, which when set to False will
disable the use of a threading local  by the ``redis`` module in its
distributed lock service, which is known to interfere with the lock's
behavior when used in an "async" use case, within dogpile this would be
when using the :paramref:`.CacheRegion.async_creation_runner` feature. The
default is conservatively being left at True, but it's likely this should
be set to False in all cases, so a warning is emitted if this flag is not
set to False in conjunction with the distributed lock. Added an optional
argument to :class:`.RedisBackend` that specifies whether or not a
thread-local Redis lock should be used.  This is the default, but it breaks
asynchronous runner compatibility.

Fixes: #171
Closes: #175
Pull-request: https://github.com/sqlalchemy/dogpile.cache/pull/175
Pull-request-sha: 755ae1b7bd826096f9d6df3ba7ab2e20f1ac4ecd

Change-Id: Ica4ed122b86c218e92cdf6174f79ff3870067934

1 files changed, 67 insertions, 0 deletions

diff --git a/tests/cache/test_redis_backend.py b/tests/cache/test_redis_backend.py
index 8baa3e8..2df350f 100644
--- a/tests/cache/test_redis_backend.py
+++ b/tests/cache/test_redis_backend.py
@@ -1,4 +1,7 @@
+from concurrent.futures import ThreadPoolExecutor
 import os
+from threading import Event
+import time
 from unittest import TestCase
 
 from mock import Mock
@@ -6,6 +9,8 @@ from mock import patch
 import pytest
 
 from dogpile.cache.region import _backend_loader
+from . import eq_
+from ._fixtures import _GenericBackendFixture
 from ._fixtures import _GenericBackendTest
 from ._fixtures import _GenericMutexTest
 
@@ -57,6 +62,68 @@ class RedisDistributedMutexTest(_TestRedisConn, _GenericMutexTest):
     }
 
 
+class RedisAsyncCreationTest(_TestRedisConn, _GenericBackendFixture, TestCase):
+    backend = "dogpile.cache.redis"
+    config_args = {
+        "arguments": {
+            "host": REDIS_HOST,
+            "port": REDIS_PORT,
+            "db": 0,
+            "distributed_lock": True,
+            # This is the important bit:
+            "thread_local_lock": False,
+        }
+    }
+
+    def test_distributed_async_locks(self):
+        pool = ThreadPoolExecutor(max_workers=1)
+        ev = Event()
+
+        # A simple example of how people may implement an async runner -
+        # plugged into a thread pool executor.
+        def asyncer(cache, key, creator, mutex):
+            def _call():
+                try:
+                    value = creator()
+                    cache.set(key, value)
+                finally:
+                    # If a thread-local lock is used here, this will fail
+                    # because generally the async calls run in a different
+                    # thread (that's the point of async creators).
+                    try:
+                        mutex.release()
+                    except Exception:
+                        pass
+                    else:
+                        ev.set()
+
+            return pool.submit(_call)
+
+        reg = self._region(
+            region_args={"async_creation_runner": asyncer},
+            config_args={"expiration_time": 0.1},
+        )
+
+        @reg.cache_on_arguments()
+        def blah(k):
+            return k * 2
+
+        # First call adds to the cache without calling the async creator.
+        eq_(blah("asd"), "asdasd")
+
+        # Wait long enough to cause the cached value to get stale.
+        time.sleep(0.3)
+
+        # This will trigger the async runner and return the stale value.
+        eq_(blah("asd"), "asdasd")
+
+        # Wait for the the async runner to finish or timeout. If the mutex
+        # release errored, then the event won't be set and we'll timeout.
+        # On <= Python 3.1, wait returned nothing. So check is_set after.
+        ev.wait(timeout=1.0)
+        eq_(ev.is_set(), True)
+
+
 @patch("redis.StrictRedis", autospec=True)
 class RedisConnectionTest(TestCase):
     backend = "dogpile.cache.redis"