Makes sure tasks always use the right app when using force_execv. Closes #1072

celery/app/base.py

@@ -8,6 +8,7 @@
 """
 from __future__ import absolute_import
 
+import os
 import threading
 import warnings
 
@@ -39,6 +40,7 @@ from .utils import (
     AppPickler, Settings, bugreport, _unpickle_app, _unpickle_app_v2,
 )
 
+_EXECV = os.environ.get('FORKED_BY_MULTIPROCESSING')
 BUILTIN_FIXUPS = frozenset([
     'celery.fixups.django:fixup',
 ])
@@ -148,7 +150,14 @@ class Celery(object):
 
     def task(self, *args, **opts):
         """Creates new task class from any callable."""
-
+        if _EXECV and not opts.get('_force_evaluate'):
+            # When using execv the task in the original module will point to a
+            # different app, so doing things like 'add.request' will point to
+            # a differnt task instance.  This makes sure it will always use
+            # the task instance from the current app.
+            # Really need a better solution for this :(
+            from . import shared_task as proxies_to_curapp
+            return proxies_to_curapp(*args, _force_evaluate=True, **opts)
         def inner_create_task_cls(shared=True, filter=None, **opts):
 
             def _create_task_cls(fun):

celery/app/builtins.py

@@ -51,7 +51,7 @@ def add_backend_cleanup_task(app):
 
     """
 
-    @app.task(name='celery.backend_cleanup')
+    @app.task(name='celery.backend_cleanup', _force_evaluate=True)
     def backend_cleanup():
         app.backend.cleanup()
     return backend_cleanup
@@ -69,7 +69,7 @@ def add_unlock_chord_task(app):
     from celery import result as _res
 
     @app.task(name='celery.chord_unlock', max_retries=None,
-              default_retry_delay=1, ignore_result=True)
+              default_retry_delay=1, ignore_result=True, _force_evaluate=True)
     def unlock_chord(group_id, callback, interval=None, propagate=False,
             max_retries=None, result=None, Result=_res.AsyncResult):
         if interval is None:
@@ -87,7 +87,7 @@ def add_unlock_chord_task(app):
 def add_map_task(app):
     from celery.canvas import subtask
 
-    @app.task(name='celery.map')
+    @app.task(name='celery.map', _force_evaluate=True)
     def xmap(task, it):
         task = subtask(task).type
         return [task(item) for item in it]
@@ -98,7 +98,7 @@ def add_map_task(app):
 def add_starmap_task(app):
     from celery.canvas import subtask
 
-    @app.task(name='celery.starmap')
+    @app.task(name='celery.starmap', _force_evaluate=True)
     def xstarmap(task, it):
         task = subtask(task).type
         return [task(*item) for item in it]
@@ -109,7 +109,7 @@ def add_starmap_task(app):
 def add_chunk_task(app):
     from celery.canvas import chunks as _chunks
 
-    @app.task(name='celery.chunks')
+    @app.task(name='celery.chunks', _force_evaluate=True)
     def chunks(task, it, n):
         return _chunks.apply_chunks(task, it, n)
     return chunks

celery/concurrency/processes.py

@@ -36,9 +36,6 @@ WORKER_SIGIGNORE = frozenset(['SIGINT'])
 
 def process_initializer(app, hostname):
     """Initializes the process so it can be used to process tasks."""
-    app.set_current()
-    set_default_app(app)
-    trace._tasks = app._tasks  # make sure this optimization is set.
     platforms.signals.reset(*WORKER_SIGRESET)
     platforms.signals.ignore(*WORKER_SIGIGNORE)
     platforms.set_mp_process_title('celeryd', hostname=hostname)
@@ -51,8 +48,14 @@ def process_initializer(app, hostname):
                   str(os.environ.get('CELERY_LOG_REDIRECT_LEVEL')))
     app.loader.init_worker()
     app.loader.init_worker_process()
-    app.finalize()
-
+    if os.environ.get('FORKED_BY_MULTIPROCESSING'):
+        # pool did execv after fork
+        trace.setup_worker_optimizations(app)
+    else:
+        app.set_current()
+        set_default_app(app)
+        app.finalize()
+        trace._tasks = app._tasks  # enables fast_trace_task optimization.
     from celery.task.trace import build_tracer
     for name, task in items(app.tasks):
         task.__trace__ = build_tracer(name, task, app.loader, hostname)

celery/task/trace.py

@@ -311,6 +311,7 @@ def setup_worker_optimizations(app):
     # we set this to always return our app.  This is a hack,
     # and means that only a single app can be used for workers
     # running in the same process.
+    app.set_current()
     set_default_app(app)
 
     # evaluate all task classes by finalizing the app.