Switch between thread & global for app.backend

celery/app/base.py

@@ -1,5 +1,6 @@
 """Actual App instance implementation."""
 import inspect
+import multiprocessing
 import os
 import sys
 import threading
@@ -229,6 +230,8 @@ def __init__(self, main=None, loader=None, backend=None,
                  **kwargs):
 
         self._local = threading.local()
+        self._global = multiprocessing.current_process()
+        self._green = False
 
         self.clock = LamportClock()
         self.main = main
@@ -1249,14 +1252,34 @@ 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(