Switch between thread & global for app.backend

celery/app/base.py

@@ -8,6 +8,7 @@
 from datetime import datetime
 from operator import attrgetter
 
+from billiard.process import current_process
 from click.exceptions import Exit
 from kombu import pools
 from kombu.clocks import LamportClock
@@ -229,6 +230,8 @@ def __init__(self, main=None, loader=None, backend=None,
                  **kwargs):
 
         self._local = threading.local()
+        self._global = current_process()
+        self._green = False
 
         self.clock = LamportClock()
         self.main = main
@@ -1249,14 +1252,32 @@ def amqp(self):
         """AMQP related functionality: :class:`~@amqp`."""
         return instantiate(self.amqp_cls, app=self)
 
+    def inform_green(self, is_green):
+        if is_green:
+            self._green = is_green
+
+    def _cached_or_create(self, name, factory, thread_safe):
+        """Thread or process store.
+
+        :param: thread_safe: func(created) => True/False
+        """
+        cached = getattr(self._local, name, None) or getattr(self._global, name, None)
+        if cached is not None:
+            return cached
+
+        created = factory()
+        use_global = self._green or thread_safe(created)
+        setattr(self._global if use_global else self._local, name, created)
+        return created
+
     @property
     def backend(self):
         """Current backend instance."""
-        try:
-            return self._local.backend
-        except AttributeError:
-            self._local.backend = new_backend = self._get_backend()
-            return new_backend
+        return self._cached_or_create(
+            'backend',
+            factory=self._get_backend,
+            thread_safe=lambda x: x.thread_safe,
+        )
 
     @property
     def conf(self):

celery/backends/base.py

@@ -101,6 +101,9 @@ class Backend:
     #: Set to true if the backend is persistent by default.
     persistent = True
 
+    #: Set to true if backend lib is thread-safe
+    thread_safe = False
+
     retry_policy = {
         'max_retries': 20,
         'interval_start': 0,

celery/backends/redis.py

@@ -191,6 +191,8 @@ class RedisBackend(BaseKeyValueStoreBackend, AsyncBackendMixin):
     GET, MGET, DEL, INCRBY, EXPIRE, SET, SETEX
     """
 
+    thread_safe = True
+
     ResultConsumer = ResultConsumer
 
     #: :pypi:`redis` client module.

celery/worker/worker.py

@@ -127,6 +127,8 @@ def setup_instance(self, queues=None, ready_callback=None, pidfile=None,
 
         # Initialize bootsteps
         self.pool_cls = _concurrency.get_implementation(self.pool_cls)
+        self.app.inform_green(self.pool_cls.is_green)
+
         self.steps = []
         self.on_init_blueprint()
         self.blueprint = self.Blueprint(

setup.cfg

@@ -8,15 +8,24 @@ all_files = 1
 # whenever it makes the code more readable.
 max-line-length = 117
 extend-ignore =
-    E203,  # incompatible with black https://github.com/psf/black/issues/315#issuecomment-395457972
-    D102,  # Missing docstring in public method
-    D104,  # Missing docstring in public package
-    D105,  # Missing docstring in magic method
-    D107,  # Missing docstring in __init__
-    D401,  # First line should be in imperative mood; try rephrasing
-    D412,  # No blank lines allowed between a section header and its content
-    E741,  # ambiguous variable name '...'
-    E742,  # ambiguous class definition '...'
+    # incompatible with black https://github.com/psf/black/issues/315#issuecomment-395457972
+    E203,
+    # Missing docstring in public method
+    D102,
+    # Missing docstring in public package
+    D104,
+    # Missing docstring in magic method
+    D105,
+    # Missing docstring in __init__
+    D107,
+    # First line should be in imperative mood; try rephrasing
+    D401,
+    # No blank lines allowed between a section header and its content
+    D412,
+    # ambiguous variable name '...'
+    E741,
+    # ambiguous class definition '...'
+    E742,
 per-file-ignores =
    t/*,setup.py,examples/*,docs/*,extra/*:
        # docstrings