diff --git a/newsfragments/591.feature.rst b/newsfragments/591.feature.rst new file mode 100644 index 0000000000..dad72d6b2e --- /dev/null +++ b/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). \ No newline at end of file diff --git a/trio/_core/_run.py b/trio/_core/_run.py index 4e7cbb2425..6f176c881b 100644 --- a/trio/_core/_run.py +++ b/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,7 +1281,9 @@ 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: @@ -1271,6 +1291,8 @@ def run( "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): "", 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 diff --git a/trio/_core/_watchdog.py b/trio/_core/_watchdog.py new file mode 100644 index 0000000000..df03691962 --- /dev/null +++ b/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="", daemon=True + ) + self._thread.start() + + def stop(self): + self._stopped = True + self._notify_event.set() + self._thread.join() diff --git a/trio/_core/tests/test_watchdog.py b/trio/_core/tests/test_watchdog.py new file mode 100644 index 0000000000..8e87540d12 --- /dev/null +++ b/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)