SIGUSR1 now logs stacktrace of all active threads. Inspired by https://gist.github.com/737056

celery/apps/worker.py

@@ -13,7 +13,7 @@ from celery import platforms
 from celery import signals
 from celery.app import app_or_default
 from celery.exceptions import ImproperlyConfigured, SystemTerminate
-from celery.utils import get_full_cls_name, LOG_LEVELS, isatty
+from celery.utils import get_full_cls_name, LOG_LEVELS, isatty, cry
 from celery.utils import term
 from celery.worker import WorkController
 
@@ -231,6 +231,7 @@ class Worker(object):
                 install_worker_restart_handler(worker)
         install_worker_term_handler(worker)
         install_worker_int_handler(worker)
+        install_cry_handler(worker.logger)
         signals.worker_init.send(sender=worker)
 
     def osx_proxy_detection_workaround(self):
@@ -303,6 +304,15 @@ def install_worker_restart_handler(worker):
     platforms.install_signal_handler("SIGHUP", restart_worker_sig_handler)
 
 
+def install_cry_handler(logger):
+
+    def cry_handler(signum, frame):
+        """Signal handler logging the stacktrace of all active threads."""
+        logger.error("\n" + cry())
+
+    platforms.install_signal_handler("SIGUSR1", cry_handler)
+
+
 def install_HUP_not_supported_handler(worker):
 
     def warn_on_HUP_handler(signum, frame):

celery/tests/utils.py

@@ -12,10 +12,8 @@ try:
     import __builtin__ as builtins
 except ImportError:    # py3k
     import builtins
-try:
-    from StringIO import StringIO
-except ImportError:
-    from io import StringIO
+
+from celery.utils.compat import StringIO
 
 from nose import SkipTest
 

celery/utils/__init__.py

@@ -5,12 +5,16 @@ import sys
 import operator
 import importlib
 import logging
+import threading
+import traceback
 
 from inspect import getargspec
 from itertools import islice
+from pprint import pprint
 
 from kombu.utils import gen_unique_id, rpartition
 
+from celery.utils.compat import StringIO
 from celery.utils.functional import partial
 
 
@@ -409,3 +413,33 @@ class cached_property(object):
 
     def deleter(self, fdel):
         return self.__class__(self.__get, self.__set, fdel)
+
+
+def cry():
+    """Return stacktrace of all active threads.
+
+    From https://gist.github.com/737056
+
+    """
+    tmap = {}
+    main_thread = None
+    # get a map of threads by their ID so we can print their names
+    # during the traceback dump
+    for t in threading.enumerate():
+        if t.ident:
+            tmap[t.ident] = t
+        else:
+            main_thread = t
+
+    out = StringIO()
+    for tid, frame in sys._current_frames().iteritems():
+        thread = tmap.get(tid, main_thread)
+        out.write("%s\n" % (thread.getName(), ))
+        out.write("=================================================\n")
+        traceback.print_stack(frame, file=out)
+        out.write("=================================================\n")
+        out.write("LOCAL VARIABLES\n")
+        out.write("=================================================\n")
+        pprint(frame.f_locals, stream=out)
+        out.write("\n\n")
+    return out.getvalue()

celery/utils/compat.py

@@ -11,6 +11,14 @@ try:
 except ImportError:
     from collections import UserDict
 
+try:
+    from cStringIO import StringIO
+except ImportError:
+    try:
+        from StringIO import StringIO
+    except ImportError:
+        from io import StringIO
+
 ############## urlparse.parse_qsl ###########################################
 
 try:

celery/worker/__init__.py

@@ -24,7 +24,9 @@ TERMINATE = 0x3
 WORKER_SIGRESET = frozenset(["SIGTERM",
                              "SIGHUP",
                              "SIGTTIN",
-                             "SIGTTOU"])
+                             "SIGTTOU",
+                             "SIGUSR1",
+                             "SIGUSR2"])
 
 #: List of signals to ignore when a child process starts.
 WORKER_SIGIGNORE = frozenset(["SIGINT"])