Adds CELERYD_FORCE_EXECV option. Closes #496

celery/app/defaults.py

@@ -156,6 +156,7 @@ NAMESPACES = {
         "CONCURRENCY": Option(0, type="int"),
         "ETA_SCHEDULER": Option(None, type="string"),
         "ETA_SCHEDULER_PRECISION": Option(1.0, type="float"),
+        "FORCE_EXECV": Option(False, type="bool"),
         "HIJACK_ROOT_LOGGER": Option(True, type="bool"),
         "CONSUMER": Option("celery.worker.consumer.Consumer"),
         "LOG_FORMAT": Option(DEFAULT_PROCESS_LOG_FMT),

celery/concurrency/processes/forking.py

@@ -57,9 +57,9 @@ if sys.platform != "win32":
         returncode = None
 
         def __init__(self, process_obj):
-            self.should_fork = process_obj.should_fork
+            self.force_execv = process_obj.force_execv
 
-            if not self.should_fork:
+            if self.force_execv:
                 sys.stdout.flush()
                 sys.stderr.flush()
                 r, w = os.pipe()

celery/concurrency/processes/pool.py

@@ -532,7 +532,8 @@ class Pool(object):
     SoftTimeLimitExceeded = SoftTimeLimitExceeded
 
     def __init__(self, processes=None, initializer=None, initargs=(),
-            maxtasksperchild=None, timeout=None, soft_timeout=None):
+            maxtasksperchild=None, timeout=None, soft_timeout=None,
+            force_execv=False):
         self._setup_queues()
         self._taskqueue = Queue.Queue()
         self._cache = {}
@@ -542,6 +543,7 @@ class Pool(object):
         self._maxtasksperchild = maxtasksperchild
         self._initializer = initializer
         self._initargs = initargs
+        self._force_execv = force_execv
 
         if soft_timeout and SIG_SOFT_TIMEOUT is None:
             warnings.warn(UserWarning("Soft timeouts are not supported: "
@@ -599,7 +601,7 @@ class Pool(object):
     def _create_worker_process(self):
         sentinel = Event()
         w = self.Process(
-            should_fork=False,
+            force_execv=self._force_execv,
             target=worker,
             args=(self._inqueue, self._outqueue,
                     self._initializer, self._initargs,

celery/concurrency/processes/process.py

@@ -7,5 +7,5 @@ class Process(_Process):
     _Popen = Popen
 
     def __init__(self, *args, **kwargs):
-        self.should_fork = kwargs.pop("should_fork", True)
+        self.force_execv = kwargs.pop("force_execv", False)
         super(Process, self).__init__(*args, **kwargs)

celery/worker/__init__.py

@@ -68,7 +68,8 @@ class Pool(abstract.StartStopComponent):
                                 maxtasksperchild=w.max_tasks_per_child,
                                 timeout=w.task_time_limit,
                                 soft_timeout=w.task_soft_time_limit,
-                                putlocks=w.pool_putlocks)
+                                putlocks=w.pool_putlocks,
+                                force_execv=w.force_execv)
         return pool
 
 
@@ -152,6 +153,7 @@ class WorkController(configurated):
     task_soft_time_limit = from_config()
     max_tasks_per_child = from_config()
     pool_putlocks = from_config()
+    force_execv = from_config()
     prefetch_multiplier = from_config()
     state_db = from_config()
     disable_rate_limits = from_config()

docs/configuration.rst

@@ -966,6 +966,24 @@ A sequence of modules to import when the celery daemon starts.
 This is used to specify the task modules to import, but also
 to import signal handlers and additional remote control commands, etc.
 
+.. setting:: CELERYD_FORCE_EXECV
+
+CELERYD_FORCE_EXECV
+~~~~~~~~~~~~~~~~~~~
+
+On Unix the processes pool will fork, so that child processes
+start with the same memory as the parent process.
+
+This can cause problems as there is a known deadlock condition
+with pthread locking primitives when `fork()` is combined with threads.
+
+You should enable this setting if you are experiencing hangs (deadlocks),
+especially in combination with time limits or having a max tasks per child limit.
+
+This option will be enabled by default in a later version.
+
+This is not a problem on Windows, as it does not have `fork()`.
+
 .. setting:: CELERYD_MAX_TASKS_PER_CHILD
 
 CELERYD_MAX_TASKS_PER_CHILD