Add task blocked watchdog.

newsfragments/591.feature.rst

@@ -0,0 +1,3 @@
+Add a watchdog for blocked tasks to Trio. This will automatically print a
+warning (and a full traceback of all threads) if a function is blocked without
+yielding for 5 seconds (by default).

trio/_core/_run.py

@@ -34,6 +34,7 @@
     CancelShieldedCheckpoint,
     WaitTaskRescheduled,
 )
+from ._watchdog import TrioWatchdog
 from .. import _core
 
 # At the bottom of this file there's also some "clever" code that generates
@@ -1156,7 +1157,9 @@ def run(
     *args,
     clock=None,
     instruments=(),
-    restrict_keyboard_interrupt_to_checkpoints=False
+    restrict_keyboard_interrupt_to_checkpoints=False,
+    use_watchdog=True,
+    watchdog_timeout=5
 ):
     """Run a trio-flavored async function, and return the result.
 
@@ -1213,6 +1216,14 @@ def run(
           main thread (this is a Python limitation), or if you use
           :func:`catch_signals` to catch SIGINT.
 
+      use_watchdog (bool): Enables the Trio task watchdog. This will spawn
+          a separate thread that will check if any tasks are blocked,
+          and if so will notify you and print the stack traces of all
+          threads to show exactly where the program is blocked.
+
+      watchdog_timeout (int): The number of seconds the watchdog will wait
+          before notifying that the main thread is blocked.
+
     Returns:
       Whatever ``async_fn`` returns.
 
@@ -1252,6 +1263,13 @@ def run(
     GLOBAL_RUN_CONTEXT.runner = runner
     locals()[LOCALS_KEY_KI_PROTECTION_ENABLED] = True
 
+    if use_watchdog:
+        watchdog = TrioWatchdog(watchdog_timeout)
+    else:
+        watchdog = None
+
+    GLOBAL_RUN_CONTEXT.watchdog = watchdog
+
     # KI handling goes outside the core try/except/finally to avoid a window
     # where KeyboardInterrupt would be allowed and converted into an
     # TrioInternalError:
@@ -1263,14 +1281,18 @@ def run(
                 with closing(runner):
                     # The main reason this is split off into its own function
                     # is just to get rid of this extra indentation.
-                    result = run_impl(runner, async_fn, args)
+                    result = run_impl(
+                        runner, async_fn, args, watchdog=watchdog
+                    )
             except TrioInternalError:
                 raise
             except BaseException as exc:
                 raise TrioInternalError(
                     "internal error in trio - please file a bug!"
                 ) from exc
             finally:
+                if use_watchdog:
+                    watchdog.stop()
                 GLOBAL_RUN_CONTEXT.__dict__.clear()
             return result.unwrap()
     finally:
@@ -1286,7 +1308,7 @@ def run(
 _MAX_TIMEOUT = 24 * 60 * 60
 
 
-def run_impl(runner, async_fn, args):
+def run_impl(runner, async_fn, args, watchdog):
     __tracebackhide__ = True
 
     runner.instrument("before_run")
@@ -1298,6 +1320,8 @@ def run_impl(runner, async_fn, args):
         "<init>",
         system_task=True,
     )
+    if watchdog is not None:
+        watchdog.start()
 
     # You know how people talk about "event loops"? This 'while' loop right
     # here is our event loop:
@@ -1370,6 +1394,8 @@ def run_impl(runner, async_fn, args):
             task = batch.pop()
             GLOBAL_RUN_CONTEXT.task = task
             runner.instrument("before_task_step", task)
+            if watchdog is not None:
+                watchdog.notify_alive_before()
 
             next_send = task._next_send
             task._next_send = None
@@ -1388,6 +1414,9 @@ def run_impl(runner, async_fn, args):
             except BaseException as task_exc:
                 final_result = Error(task_exc)
 
+            if watchdog is not None:
+                watchdog.notify_alive_after()
+
             if final_result is not None:
                 # We can't call this directly inside the except: blocks above,
                 # because then the exceptions end up attaching themselves to

trio/_core/_watchdog.py

@@ -0,0 +1,84 @@
+import sys
+
+import threading
+import traceback
+
+
+class TrioWatchdog(object):
+    def __init__(self, timeout=5):
+        self._stopped = False
+        self._thread = None
+        self._notify_event = threading.Event()
+        self._timeout = timeout
+
+        self._before_counter = 0
+        self._after_counter = 0
+
+    def notify_alive_before(self):
+        """
+        Notifies the watchdog that the Trio thread is alive before running
+        a task.
+        """
+        self._before_counter += 1
+        self._notify_event.set()
+
+    def notify_alive_after(self):
+        """
+        Notifies the watchdog that the Trio thread is alive after running a
+        task.
+        """
+        self._after_counter += 1
+
+    def _main_loop(self):
+        while True:
+            if self._stopped:
+                return
+
+            self._notify_event.clear()
+            orig_starts = self._before_counter
+            orig_stops = self._after_counter
+            if orig_starts == orig_stops:
+                # main thread asleep; nothing to do until it wakes up
+                self._notify_event.wait()
+                if self._stopped:
+                    return
+            else:
+                self._notify_event.wait(timeout=self._timeout)
+                if self._stopped:
+                    return
+
+                if orig_starts == self._before_counter \
+                        and orig_stops == self._after_counter:
+                    print(
+                        "Trio Watchdog has not received any notifications in "
+                        "5 seconds, main thread is blocked!",
+                        file=sys.stderr
+                    )
+                    # faulthandler is not very useful to us, honestly
+                    # faulthandler.dump_traceback(all_threads=True)
+                    print(
+                        "Printing the traceback of all threads:",
+                        file=sys.stderr
+                    )
+                    self._print_all_threads()
+
+    def _print_all_threads(self):
+        # separated for indent reasons, damned 80 char limit
+        for thread in threading.enumerate():
+            print(
+                "Thread {} (most recent call last):".format(thread.name),
+                file=sys.stderr
+            )
+            # scary internal function!
+            traceback.print_stack(sys._current_frames()[thread.ident])
+
+    def start(self):
+        self._thread = threading.Thread(
+            target=self._main_loop, name="<trio watchdog>", daemon=True
+        )
+        self._thread.start()
+
+    def stop(self):
+        self._stopped = True
+        self._notify_event.set()
+        self._thread.join()

trio/_core/tests/test_watchdog.py

@@ -0,0 +1,30 @@
+import time
+
+import contextlib
+from io import StringIO
+
+from ..tests.tutil import slow
+from ... import _core
+from ..._timeouts import sleep
+
+MAGIC_TEXT = "Trio Watchdog has not received any notifications in 5 seconds, \
+main thread is blocked!"
+
+
+@slow
+def test_watchdog():
+    async def _inner_test():
+        target = StringIO()
+        with contextlib.redirect_stderr(target):
+            time.sleep(2)
+
+        assert target.getvalue().startswith(MAGIC_TEXT)
+
+        target = StringIO()
+        with contextlib.redirect_stderr(target):
+            await sleep(2)
+
+        # if pytest puts garbage in stderr this won't fail
+        assert not target.getvalue().startswith(MAGIC_TEXT)
+
+    _core.run(_inner_test, use_watchdog=True, watchdog_timeout=1)