Ensure the cachedproperty creation/setting is thread-safe

When the cachedproperty descriptor is attached to an object
that needs to be only created/set by one thread at a time we
should ensure that this is done safely by using a lock to
prevent multiple threads from creating and assigning the
associated attribute.

Fixes bug 1366156

Change-Id: I0545683f83402097f54c34a6b737904e6edd85b3

2 files changed, 48 insertions, 6 deletions

diff --git a/taskflow/tests/unit/test_utils.py b/taskflow/tests/unit/test_utils.py
index 4518f96..5d4b761 100644
--- a/taskflow/tests/unit/test_utils.py
+++ b/taskflow/tests/unit/test_utils.py
@@ -17,7 +17,10 @@
 import collections
 import functools
 import inspect
+import random
 import sys
+import threading
+import time
 
 import six
 import testtools
@@ -437,6 +440,36 @@ class CachedPropertyTest(test.TestCase):
 
         self.assertEqual(None, inspect.getdoc(A.b))
 
+    def test_threaded_access_property(self):
+        called = collections.deque()
+
+        class A(object):
+            @misc.cachedproperty
+            def b(self):
+                called.append(1)
+                # NOTE(harlowja): wait for a little and give some time for
+                # another thread to potentially also get in this method to
+                # also create the same property...
+                time.sleep(random.random() * 0.5)
+                return 'b'
+
+        a = A()
+        threads = []
+        try:
+            for _i in range(0, 20):
+                t = threading.Thread(target=lambda: a.b)
+                t.daemon = True
+                threads.append(t)
+            for t in threads:
+                t.start()
+        finally:
+            while threads:
+                t = threads.pop()
+                t.join()
+
+        self.assertEqual(1, len(called))
+        self.assertEqual('b', a.b)
+
 
 class AttrDictTest(test.TestCase):
     def test_ok_create(self):
diff --git a/taskflow/utils/misc.py b/taskflow/utils/misc.py
index 8e5e192..a993f29 100644
--- a/taskflow/utils/misc.py
+++ b/taskflow/utils/misc.py
@@ -27,6 +27,7 @@ import os
 import re
 import string
 import sys
+import threading
 import time
 import traceback
 
@@ -163,7 +164,7 @@ def decode_json(raw_data, root_types=(dict,)):
 
 
 class cachedproperty(object):
-    """A descriptor property that is only evaluated once..
+    """A *thread-safe* descriptor property that is only evaluated once.
 
     This caching descriptor can be placed on instance methods to translate
     those methods into properties that will be cached in the instance (avoiding
@@ -176,6 +177,7 @@ class cachedproperty(object):
     after the first call to 'get_thing' occurs.
     """
     def __init__(self, fget):
+        self._lock = threading.RLock()
         # If a name is provided (as an argument) then this will be the string
         # to place the cached attribute under if not then it will be the
         # function itself to be wrapped into a property.
@@ -205,12 +207,19 @@ class cachedproperty(object):
     def __get__(self, instance, owner):
         if instance is None:
             return self
-        try:
+        # Quick check to see if this already has been made (before acquiring
+        # the lock). This is safe to do since we don't allow deletion after
+        # being created.
+        if hasattr(instance, self._attr_name):
             return getattr(instance, self._attr_name)
-        except AttributeError:
-            value = self._fget(instance)
-            setattr(instance, self._attr_name, value)
-            return value
+        else:
+            with self._lock:
+                try:
+                    return getattr(instance, self._attr_name)
+                except AttributeError:
+                    value = self._fget(instance)
+                    setattr(instance, self._attr_name, value)
+                    return value
 
 
 def wallclock():