Anonymous avatar Anonymous committed 97a9826

Add support for the itimer module with blocking detection

This will allow for subsecond blocking detection thanks to the itimer
module recently released by Slide, Inc:

Comments (0)

Files changed (4)


     from eventlet import tpool
     tpool.QUIET = not state
-def hub_blocking_detection(state = False):
+def hub_blocking_detection(state = False, resolution = 1):
     """Toggles whether Eventlet makes an effort to detect blocking
     behavior in other code.  It does this by setting a SIGALARM with a short 
     from eventlet import hubs
     hubs.get_hub().debug_blocking = state
+    hubs.get_hub().debug_blocking_resolution = resolution
     if(not state):


 import warnings
 import signal
+    import itimer
+    HAS_ITIMER = True
+except ImportError:
+    pass
 from import greenlets as greenlet, clear_sys_exc_info
 from eventlet.hubs import timer
 from eventlet import patcher
         self.timers_canceled = 0
         self.debug_exceptions = True
         self.debug_blocking = False
+        self.debug_blocking_resolution = 1
     def block_detect_pre(self):
         # shortest alarm we can possibly raise is one second
         tmp = signal.signal(signal.SIGALRM, alarm_handler)
         if tmp != alarm_handler:
             self._old_signal_handler = tmp
-        signal.alarm(1)
+        if HAS_ITIMER:
+            itimer.alarm(self.debug_blocking_resolution)
+        else:
+            signal.alarm(self.debug_blocking_resolution)
     def block_detect_post(self):
         if (hasattr(self, "_old_signal_handler") and


     import sys
     return skip_if(sys.platform.startswith('win'))(func)
+def skip_if_no_itimer(func):
+    """ Decorator that skips a test if the `itimer` module isn't found """
+    has_itimer = False
+    try:
+        import itimer
+        has_itimer = True
+    except ImportError:
+        pass
+    return skip_unless(has_itimer)(func)
 class TestIsTakingTooLong(Exception):
     """ Custom exception class to be raised when a test's runtime exceeds a limit. """


-from tests import LimitedTestCase, main, skip_with_pyevent
+from tests import LimitedTestCase, main, skip_with_pyevent, skip_if_no_itimer
 import time
 import eventlet
 from eventlet import hubs
         self.assertRaises(RuntimeError, gt.wait)
+    @skip_if_no_itimer
+    def test_block_detect_with_itimer(self):
+        def look_im_blocking():
+            import time
+            time.sleep(0.5)
+        from eventlet import debug
+        debug.hub_blocking_detection(True, resolution=0.1)
+        gt = eventlet.spawn(look_im_blocking)
+        self.assertRaises(RuntimeError, gt.wait)
+        debug.hub_blocking_detection(False)
 class TestSuspend(LimitedTestCase):
     def test_suspend_doesnt_crash(self):
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.