From 68f84c8a8bcd6d3690e69a49edecb4ae26a5500d Mon Sep 17 00:00:00 2001 From: "Laura F. D" Date: Sun, 12 Aug 2018 16:13:27 +0100 Subject: [PATCH 1/3] Add task blocked watchdog. Closes #591. --- trio/_core/_run.py | 29 +++++++++++++-- trio/_core/_watchdog.py | 81 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 107 insertions(+), 3 deletions(-) create mode 100644 trio/_core/_watchdog.py diff --git a/trio/_core/_run.py b/trio/_core/_run.py index 4e7cbb2425..cc4e245379 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,8 @@ def run( *args, clock=None, instruments=(), - restrict_keyboard_interrupt_to_checkpoints=False + restrict_keyboard_interrupt_to_checkpoints=False, + use_watchdog=True ): """Run a trio-flavored async function, and return the result. @@ -1213,6 +1215,11 @@ 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. + Returns: Whatever ``async_fn`` returns. @@ -1252,6 +1259,11 @@ def run( GLOBAL_RUN_CONTEXT.runner = runner locals()[LOCALS_KEY_KI_PROTECTION_ENABLED] = True + if use_watchdog: + watchdog = TrioWatchdog() + else: + watchdog = None + # 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 +1275,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 +1285,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 +1302,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 +1314,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 +1388,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 +1408,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..aa79e538f5 --- /dev/null +++ b/trio/_core/_watchdog.py @@ -0,0 +1,81 @@ +import sys + +import threading +import traceback + + +class TrioWatchdog(object): + def __init__(self): + self._stopped = False + self._thread = None + self._notify_event = threading.Event() + + 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=5) + 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 From d343965b622f3a8238090aeb536a884ea71dbe52 Mon Sep 17 00:00:00 2001 From: "Laura F. D" Date: Mon, 13 Aug 2018 14:05:47 +0100 Subject: [PATCH 2/3] Add a test for the watchdog. --- newsfragments/591.feature.rst | 3 +++ trio/_core/_run.py | 2 ++ trio/_core/tests/test_watchdog.py | 30 ++++++++++++++++++++++++++++++ 3 files changed, 35 insertions(+) create mode 100644 newsfragments/591.feature.rst create mode 100644 trio/_core/tests/test_watchdog.py diff --git a/newsfragments/591.feature.rst b/newsfragments/591.feature.rst new file mode 100644 index 0000000000..007207317d --- /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. \ No newline at end of file diff --git a/trio/_core/_run.py b/trio/_core/_run.py index cc4e245379..b9539c754b 100644 --- a/trio/_core/_run.py +++ b/trio/_core/_run.py @@ -1264,6 +1264,8 @@ def run( 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: diff --git a/trio/_core/tests/test_watchdog.py b/trio/_core/tests/test_watchdog.py new file mode 100644 index 0000000000..5ac809098a --- /dev/null +++ b/trio/_core/tests/test_watchdog.py @@ -0,0 +1,30 @@ +import threading +import time +from io import StringIO + +import contextlib + +from ..tests.tutil import slow +from .. import _run + + +@slow +async def test_watchdog(): + watchdog = _run.GLOBAL_RUN_CONTEXT.watchdog + target = StringIO() + with contextlib.redirect_stderr(target): + time.sleep(7) + + assert target.getvalue().startswith( + "Trio Watchdog has not received any " + "notifications in 5 seconds, main " + "thread is blocked!" + ) + + # checkpoint to ensure the watchdog gets a notification, sees that its + # dead, and winds down its thread + watchdog.stop() + await _run.checkpoint() + time.sleep(6) # ensure if the watchdog is waiting for 5s, it wakes + await _run.checkpoint() + assert not watchdog._thread.is_alive() From 753753be65de75c8dc1369d3738ed01045fbaa8d Mon Sep 17 00:00:00 2001 From: "Laura F. D" Date: Wed, 22 Aug 2018 17:58:56 +0100 Subject: [PATCH 3/3] Make changes as appropriate --- newsfragments/591.feature.rst | 2 +- trio/_core/_run.py | 8 ++++-- trio/_core/_watchdog.py | 7 +++-- trio/_core/tests/test_watchdog.py | 44 +++++++++++++++---------------- 4 files changed, 34 insertions(+), 27 deletions(-) diff --git a/newsfragments/591.feature.rst b/newsfragments/591.feature.rst index 007207317d..dad72d6b2e 100644 --- a/newsfragments/591.feature.rst +++ b/newsfragments/591.feature.rst @@ -1,3 +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. \ No newline at end of file +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 b9539c754b..6f176c881b 100644 --- a/trio/_core/_run.py +++ b/trio/_core/_run.py @@ -1158,7 +1158,8 @@ def run( clock=None, instruments=(), restrict_keyboard_interrupt_to_checkpoints=False, - use_watchdog=True + use_watchdog=True, + watchdog_timeout=5 ): """Run a trio-flavored async function, and return the result. @@ -1220,6 +1221,9 @@ def run( 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. @@ -1260,7 +1264,7 @@ def run( locals()[LOCALS_KEY_KI_PROTECTION_ENABLED] = True if use_watchdog: - watchdog = TrioWatchdog() + watchdog = TrioWatchdog(watchdog_timeout) else: watchdog = None diff --git a/trio/_core/_watchdog.py b/trio/_core/_watchdog.py index aa79e538f5..df03691962 100644 --- a/trio/_core/_watchdog.py +++ b/trio/_core/_watchdog.py @@ -5,10 +5,11 @@ class TrioWatchdog(object): - def __init__(self): + 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 @@ -42,7 +43,7 @@ def _main_loop(self): if self._stopped: return else: - self._notify_event.wait(timeout=5) + self._notify_event.wait(timeout=self._timeout) if self._stopped: return @@ -79,3 +80,5 @@ def start(self): 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 index 5ac809098a..8e87540d12 100644 --- a/trio/_core/tests/test_watchdog.py +++ b/trio/_core/tests/test_watchdog.py @@ -1,30 +1,30 @@ -import threading import time -from io import StringIO import contextlib +from io import StringIO from ..tests.tutil import slow -from .. import _run +from ... import _core +from ..._timeouts import sleep + +MAGIC_TEXT = "Trio Watchdog has not received any notifications in 5 seconds, \ +main thread is blocked!" @slow -async def test_watchdog(): - watchdog = _run.GLOBAL_RUN_CONTEXT.watchdog - target = StringIO() - with contextlib.redirect_stderr(target): - time.sleep(7) - - assert target.getvalue().startswith( - "Trio Watchdog has not received any " - "notifications in 5 seconds, main " - "thread is blocked!" - ) - - # checkpoint to ensure the watchdog gets a notification, sees that its - # dead, and winds down its thread - watchdog.stop() - await _run.checkpoint() - time.sleep(6) # ensure if the watchdog is waiting for 5s, it wakes - await _run.checkpoint() - assert not watchdog._thread.is_alive() +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)