diff --git a/src/debugpy/_vendored/pydevd/_pydevd_bundle/pydevd_comm.py b/src/debugpy/_vendored/pydevd/_pydevd_bundle/pydevd_comm.py index 352bb532..b935320c 100644 --- a/src/debugpy/_vendored/pydevd/_pydevd_bundle/pydevd_comm.py +++ b/src/debugpy/_vendored/pydevd/_pydevd_bundle/pydevd_comm.py @@ -68,6 +68,7 @@ from _pydev_bundle.pydev_imports import _queue from _pydev_bundle._pydev_saved_modules import time, ThreadingEvent +from _pydev_bundle._pydev_saved_modules import threading from _pydev_bundle._pydev_saved_modules import socket as socket_module from _pydevd_bundle.pydevd_constants import ( DebugInfoHolder, @@ -83,6 +84,7 @@ silence_warnings_decorator, filter_all_warnings, IS_PY311_OR_GREATER, + PYDEVD_UNBLOCK_THREADS_ON_VARIABLES_TIMEOUT, ) from _pydev_bundle.pydev_override import overrides import weakref @@ -796,8 +798,22 @@ def internal_get_variable_json(py_db, request): except KeyError: pass else: - for child_var in variable.get_children_variables(fmt=fmt, scope=scope): - variables.append(child_var.get_var_data(fmt=fmt)) + # Resolving the children of a variable may block on another (suspended) thread -- e.g. a + # property getter that dispatches work to a background event loop thread. Since all + # threads are suspended at a breakpoint, that would deadlock the debugger. If it takes + # too long, resume the other threads until it completes (and re-suspend afterwards). + timeout_message = ( + "pydevd: Resolving variable children is taking too long (it may be waiting on another " + "thread). Resuming other threads until it finishes so the debugger doesn't hang." + ) + with pydevd_vars.unblock_threads_on_timeout( + py_db, + threading.current_thread(), + PYDEVD_UNBLOCK_THREADS_ON_VARIABLES_TIMEOUT, + on_timeout_message=timeout_message, + ): + for child_var in variable.get_children_variables(fmt=fmt, scope=scope): + variables.append(child_var.get_var_data(fmt=fmt)) except: try: exc, exc_type, tb = sys.exc_info() diff --git a/src/debugpy/_vendored/pydevd/_pydevd_bundle/pydevd_constants.py b/src/debugpy/_vendored/pydevd/_pydevd_bundle/pydevd_constants.py index b73bd489..9245eb47 100644 --- a/src/debugpy/_vendored/pydevd/_pydevd_bundle/pydevd_constants.py +++ b/src/debugpy/_vendored/pydevd/_pydevd_bundle/pydevd_constants.py @@ -364,6 +364,21 @@ def as_int_in_env(env_key, default): # elapses. PYDEVD_UNBLOCK_THREADS_TIMEOUT = as_float_in_env("PYDEVD_UNBLOCK_THREADS_TIMEOUT", -1.0) +# This timeout is used only when the mode that all threads are stopped/resumed at once is used +# (i.e.: multi_threads_single_notification). +# +# When resolving the children of a variable (i.e.: expanding it in the variables view), computing +# one of its attributes may end up blocked waiting on another (currently suspended) thread -- for +# instance a property getter that dispatches work to a background event loop running in another +# thread. As all threads are suspended at a breakpoint, that would deadlock the debugger. +# +# If the resolution doesn't finish until this timeout elapses, we resume all other threads until it +# finishes (and then suspend them again) so that such computations can complete instead of hanging +# forever. +# +# A negative value disables this behavior. +PYDEVD_UNBLOCK_THREADS_ON_VARIABLES_TIMEOUT = as_float_in_env("PYDEVD_UNBLOCK_THREADS_ON_VARIABLES_TIMEOUT", 3.0) + # Timeout to interrupt a thread (so, if some evaluation doesn't finish until this # timeout, the thread doing the evaluation is interrupted). # A value <= 0 means this is disabled. diff --git a/src/debugpy/_vendored/pydevd/_pydevd_bundle/pydevd_resolver.py b/src/debugpy/_vendored/pydevd/_pydevd_bundle/pydevd_resolver.py index 488793c4..f02f1d69 100644 --- a/src/debugpy/_vendored/pydevd/_pydevd_bundle/pydevd_resolver.py +++ b/src/debugpy/_vendored/pydevd/_pydevd_bundle/pydevd_resolver.py @@ -190,9 +190,13 @@ def _get_py_dictionary(self, var, names=None, used___dict__=False): name_as_str = "%r" % (name_as_str,) if not used___dict__: - if not hasattr(var, name): + try: + attr = getattr(var, name) + except AttributeError: + # Attribute went away (or the descriptor raised AttributeError): skip it. + # Note: getattr is called a single time on purpose -- calling hasattr first + # would evaluate the (possibly expensive or side-effecting) descriptor twice. continue - attr = getattr(var, name) else: attr = var.__dict__[name] diff --git a/src/debugpy/_vendored/pydevd/_pydevd_bundle/pydevd_vars.py b/src/debugpy/_vendored/pydevd/_pydevd_bundle/pydevd_vars.py index cc0fc280..94aaf1aa 100644 --- a/src/debugpy/_vendored/pydevd/_pydevd_bundle/pydevd_vars.py +++ b/src/debugpy/_vendored/pydevd/_pydevd_bundle/pydevd_vars.py @@ -8,6 +8,7 @@ from _pydevd_bundle.pydevd_xml import ExceptionOnEvaluate, get_type, var_to_xml from _pydev_bundle import pydev_log import functools +from contextlib import contextmanager from _pydevd_bundle.pydevd_thread_lifecycle import resume_threads, mark_thread_suspended, suspend_all_threads, suspend_threads_lock from _pydevd_bundle.pydevd_comm_constants import CMD_SET_BREAK @@ -309,41 +310,75 @@ def _run_with_interrupt_thread(original_func, py_db, curr_thread, frame, express return original_func(py_db, frame, expression, is_exec) -def _run_with_unblock_threads(original_func, py_db, curr_thread, frame, expression, is_exec): +@contextmanager +def unblock_threads_on_timeout(py_db, curr_thread, timeout, on_timeout_message=None): + """ + Context manager that runs the wrapped block and, if it doesn't finish until ``timeout`` seconds + elapse, resumes all other threads -- so that a computation depending on another (currently + suspended) thread can make progress -- and then suspends them again when the block finishes. + + This only has an effect when threads are suspended/resumed as a group + (i.e.: ``py_db.multi_threads_single_notification``) and ``timeout >= 0``; otherwise the block + simply runs unchanged. + + :param on_timeout_message: + Optional warning message sent to the client if the timeout elapses and the other threads are + resumed (so the user understands why other threads ran while execution was suspended). + """ on_timeout_unblock_threads = None timeout_tracker = py_db.timeout_tracker # : :type timeout_tracker: TimeoutTracker - if py_db.multi_threads_single_notification: - unblock_threads_timeout = pydevd_constants.PYDEVD_UNBLOCK_THREADS_TIMEOUT - else: - unblock_threads_timeout = -1 # Don't use this if threads are managed individually. - - if unblock_threads_timeout >= 0: - pydev_log.info("Doing evaluate with unblock threads timeout: %s.", unblock_threads_timeout) + if py_db.multi_threads_single_notification and timeout >= 0: + pydev_log.info("Running with unblock threads timeout: %s.", timeout) tid = get_current_thread_id(curr_thread) - def on_timeout_unblock_threads(): - on_timeout_unblock_threads.called = True - pydev_log.info("Resuming threads after evaluate timeout.") - resume_threads("*", except_thread=curr_thread) - py_db.threads_suspended_single_notification.on_thread_resume(tid, curr_thread) + # State shared between the timeout callback (which runs on the TimeoutTracker daemon thread) + # and the ``finally`` block below (which runs on ``curr_thread``). Both the resume and the + # re-suspend transitions are performed under ``suspend_threads_lock`` so they are serialized: + # the ``finally`` cannot re-suspend before the callback finishes resuming, and the callback + # cannot resume after the ``finally`` has decided the block is finished. + unblock_state = {"resumed": False, "finished": False} - on_timeout_unblock_threads.called = False + def on_timeout_unblock_threads(): + with suspend_threads_lock: + if unblock_state["finished"]: + # The block already finished; don't resume threads that are meant to stay suspended. + return + unblock_state["resumed"] = True + pydev_log.info("Resuming threads after timeout.") + if on_timeout_message is not None: + try: + py_db.writer.add_command(py_db.cmd_factory.make_warning_message(on_timeout_message)) + except Exception: + pydev_log.exception() + resume_threads("*", except_thread=curr_thread) + py_db.threads_suspended_single_notification.on_thread_resume(tid, curr_thread) try: if on_timeout_unblock_threads is None: - return _run_with_interrupt_thread(original_func, py_db, curr_thread, frame, expression, is_exec) + yield else: - with timeout_tracker.call_on_timeout(unblock_threads_timeout, on_timeout_unblock_threads): - return _run_with_interrupt_thread(original_func, py_db, curr_thread, frame, expression, is_exec) - + with timeout_tracker.call_on_timeout(timeout, on_timeout_unblock_threads): + yield finally: - if on_timeout_unblock_threads is not None and on_timeout_unblock_threads.called: + if on_timeout_unblock_threads is not None: with suspend_threads_lock: - mark_thread_suspended(curr_thread, CMD_SET_BREAK) - py_db.threads_suspended_single_notification.increment_suspend_time() - suspend_all_threads(py_db, except_thread=curr_thread) - py_db.threads_suspended_single_notification.on_thread_suspend(tid, curr_thread, CMD_SET_BREAK) + unblock_state["finished"] = True + if unblock_state["resumed"]: + mark_thread_suspended(curr_thread, CMD_SET_BREAK) + py_db.threads_suspended_single_notification.increment_suspend_time() + suspend_all_threads(py_db, except_thread=curr_thread) + py_db.threads_suspended_single_notification.on_thread_suspend(tid, curr_thread, CMD_SET_BREAK) + + +def _run_with_unblock_threads(original_func, py_db, curr_thread, frame, expression, is_exec): + if py_db.multi_threads_single_notification: + unblock_threads_timeout = pydevd_constants.PYDEVD_UNBLOCK_THREADS_TIMEOUT + else: + unblock_threads_timeout = -1 # Don't use this if threads are managed individually. + + with unblock_threads_on_timeout(py_db, curr_thread, unblock_threads_timeout): + return _run_with_interrupt_thread(original_func, py_db, curr_thread, frame, expression, is_exec) def _evaluate_with_timeouts(original_func): diff --git a/src/debugpy/_vendored/pydevd/tests_python/resources/_debugger_case_deadlock_thread_variables.py b/src/debugpy/_vendored/pydevd/tests_python/resources/_debugger_case_deadlock_thread_variables.py new file mode 100644 index 00000000..31a1697e --- /dev/null +++ b/src/debugpy/_vendored/pydevd/tests_python/resources/_debugger_case_deadlock_thread_variables.py @@ -0,0 +1,80 @@ +""" +The idea here is that a secondary thread does the processing of instructions, and an object has a +property whose getter dispatches work to that secondary thread and blocks waiting for the result. + +So, when all threads are stopped at a breakpoint, *expanding* that object in the variables view (which +evaluates its properties) would be locked until the secondary thread is allowed to run. + +This mirrors real-world objects such as lancedb's ``LanceDBConnection``, whose property getters block +on a background asyncio event loop running in a daemon thread. +""" + +import threading + +try: + from queue import Queue +except ImportError: + from Queue import Queue + + +class EchoThread(threading.Thread): + def __init__(self, queue): + threading.Thread.__init__(self) + self.daemon = True + self._queue = queue + self.started = threading.Event() + + def run(self): + self.started.set() + while True: + obj = self._queue.get() + if obj == "finish": + break + + obj.result = obj.value + 1 + obj.event.set() # Break here 2 + + +class NotificationObject(object): + def __init__(self, value): + self.value = value + self.result = None + self.event = threading.Event() + + +class Connection(object): + """ + Mimics a native-backed connection whose property getter dispatches to a background thread and + blocks on the result (like lancedb's LanceDBConnection). + """ + + def __init__(self, queue): + self._queue = queue + self.storage_options = None + + @property + def read_consistency_interval(self): + obj = NotificationObject(41) + self._queue.put(obj) + assert obj.event.wait() # Blocks until the (suspended) EchoThread processes the request. + return obj.result + + def __repr__(self): + return "Connection(read_consistency_interval=)" + + +def main(): + queue = Queue() + echo_thread = EchoThread(queue) + processor = Connection(queue) + echo_thread.start() + echo_thread.started.wait() + + print("stop here") # Break here 1 + + queue.put("finish") + + +if __name__ == "__main__": + main() + print("TEST SUCEEDED!") diff --git a/src/debugpy/_vendored/pydevd/tests_python/test_debugger_json.py b/src/debugpy/_vendored/pydevd/tests_python/test_debugger_json.py index a4a8c026..06e9ecaf 100644 --- a/src/debugpy/_vendored/pydevd/tests_python/test_debugger_json.py +++ b/src/debugpy/_vendored/pydevd/tests_python/test_debugger_json.py @@ -6178,6 +6178,35 @@ def get_environ(self): writer.finished_ok = True +def test_debugger_case_deadlock_thread_variables(case_setup_dap): + # Expanding a variable whose property getter blocks on another (suspended) thread must not + # deadlock the debugger: the other threads should be resumed until the resolution finishes. + + def get_environ(self): + env = os.environ.copy() + env["PYDEVD_UNBLOCK_THREADS_ON_VARIABLES_TIMEOUT"] = "0.5" + return env + + with case_setup_dap.test_file("_debugger_case_deadlock_thread_variables.py", get_environ=get_environ) as writer: + json_facade = JsonFacade(writer) + json_facade.write_launch() + json_facade.write_set_breakpoints(writer.get_line_index_with_content("Break here 1")) + + json_facade.write_make_initial_run() + json_hit = json_facade.wait_for_thread_stopped() + + # Expanding "processor" evaluates its "read_consistency_interval" property, which blocks on + # the (suspended) EchoThread. If threads aren't resumed, this will deadlock. + processor_var = json_facade.get_local_var(json_hit.frame_id, "processor") + name_to_var = json_facade.get_name_to_var(processor_var.variablesReference) + assert "read_consistency_interval" in name_to_var + assert name_to_var["read_consistency_interval"].value == "42" + + json_facade.write_continue() + + writer.finished_ok = True + + def test_debugger_case_breakpoint_on_unblock_thread_eval(case_setup_dap): from _pydevd_bundle._debug_adapter.pydevd_schema import EvaluateResponse