Fix debugger hang when expanding objects with blocking property getters (#2053)

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()

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.

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]
 

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):

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=<computed lazily>)"
+
+
+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!")

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