Fix hang after rapid continue with multiple threads

src/debugpy/_vendored/pydevd/pydevd.py

@@ -376,6 +376,7 @@ def do_wait_suspend(...):
         "_suspended_thread_id_to_thread",
         "_pause_requested",
         "_py_db",
+        "_on_resumed_callbacks",
     ]
 
     NOTIFY_OF_PAUSE_TIMEOUT = 0.5
@@ -389,6 +390,42 @@ def __init__(self, py_db):
         self._lock = thread.allocate_lock()
         self._suspended_thread_id_to_thread = {}
         self._pause_requested = False
+        self._on_resumed_callbacks = []
+
+    def add_on_resumed_callback(self, callback):
+        """
+        Calls `callback` when the last suspend notification is followed by a resume notification.
+
+        If no suspend notification is waiting for a resume (e.g.: a continue was requested after the
+        last stop was already notified as resumed), no resume notification will come, so, it's called
+        right away.
+        """
+        with self._lock:
+            if self._last_resume_notification_time < self._last_suspend_notification_time:
+                self._on_resumed_callbacks.append(callback)
+                return
+        pydev_log.info("Resume already notified: calling on resumed callback right away.")
+        callback()
+
+    def _notify_resumed(self, thread_id):
+        # Must be called with self._lock held.
+        self._last_resume_notification_time = self._last_suspend_notification_time
+        self.send_resume_notification(thread_id)
+        callbacks = self._on_resumed_callbacks
+        self._on_resumed_callbacks = []
+        for callback in callbacks:
+            callback()
+
+    def _notify_suspended(self, thread_id, thread, stop_reason, suspend_time):
+        # Must be called with self._lock held.
+        if self._on_resumed_callbacks:
+            # A resume of the previous stop was requested, but another thread reached a new
+            # stop before any suspended thread left its wait loop (so, the resume was never
+            # notified). Notify it now, otherwise the pending callbacks are never called.
+            pydev_log.info("Sending pending resume notification before new suspend notification.")
+            self._notify_resumed(thread_id)
+        self._last_suspend_notification_time = suspend_time
+        self.send_suspend_notification(thread_id, thread, stop_reason)
 
     def send_suspend_notification(self, thread_id, thread, stop_reason):
         raise AssertionError("abstract: subclasses must override.")
@@ -416,11 +453,10 @@ def _notify_after_timeout(self, global_suspend_time):
         with self._lock:
             if self._suspended_thread_id_to_thread:
                 if global_suspend_time > self._last_suspend_notification_time:
-                    self._last_suspend_notification_time = global_suspend_time
                     # Notify about any thread which is currently suspended.
                     pydev_log.info("Sending suspend notification after timeout.")
                     thread_id, thread = next(iter(self._suspended_thread_id_to_thread.items()))
-                    self.send_suspend_notification(thread_id, thread, CMD_THREAD_SUSPEND)
+                    self._notify_suspended(thread_id, thread, CMD_THREAD_SUSPEND, global_suspend_time)
 
     def on_thread_suspend(self, thread_id, thread, stop_reason):
         with self._lock:
@@ -436,8 +472,7 @@ def on_thread_suspend(self, thread_id, thread, stop_reason):
             if stop_reason != CMD_THREAD_SUSPEND or pause_requested:
                 if self._suspend_time_request > self._last_suspend_notification_time:
                     pydev_log.info("Sending suspend notification.")
-                    self._last_suspend_notification_time = self._suspend_time_request
-                    self.send_suspend_notification(thread_id, thread, stop_reason)
+                    self._notify_suspended(thread_id, thread, stop_reason, self._suspend_time_request)
                 else:
                     pydev_log.info(
                         "Suspend not sent (it was already sent). Last suspend % <= Last resume %s",
@@ -455,8 +490,7 @@ def on_thread_resume(self, thread_id, thread):
             self._suspended_thread_id_to_thread.pop(thread_id)
             if self._last_resume_notification_time < self._last_suspend_notification_time:
                 pydev_log.info("Sending resume notification.")
-                self._last_resume_notification_time = self._last_suspend_notification_time
-                self.send_resume_notification(thread_id)
+                self._notify_resumed(thread_id)
             else:
                 pydev_log.info(
                     "Resume not sent (it was already sent). Last resume %s >= Last suspend %s",
@@ -474,32 +508,19 @@ def notify_thread_suspended(self, thread_id, thread, stop_reason):
 
 
 class ThreadsSuspendedSingleNotification(AbstractSingleNotificationBehavior):
-    __slots__ = AbstractSingleNotificationBehavior.__slots__ + ["multi_threads_single_notification", "_callbacks", "_callbacks_lock"]
+    __slots__ = AbstractSingleNotificationBehavior.__slots__ + ["multi_threads_single_notification"]
 
     def __init__(self, py_db):
         AbstractSingleNotificationBehavior.__init__(self, py_db)
         # If True, pydevd will send a single notification when all threads are suspended/resumed.
         self.multi_threads_single_notification = False
-        self._callbacks_lock = threading.Lock()
-        self._callbacks = []
-
-    def add_on_resumed_callback(self, callback):
-        with self._callbacks_lock:
-            self._callbacks.append(callback)
 
     @overrides(AbstractSingleNotificationBehavior.send_resume_notification)
     def send_resume_notification(self, thread_id):
         py_db = self._py_db()
         if py_db is not None:
             py_db.writer.add_command(py_db.cmd_factory.make_thread_resume_single_notification(thread_id))
 
-            with self._callbacks_lock:
-                callbacks = self._callbacks
-                self._callbacks = []
-
-            for callback in callbacks:
-                callback()
-
     @overrides(AbstractSingleNotificationBehavior.send_suspend_notification)
     def send_suspend_notification(self, thread_id, thread, stop_reason):
         py_db = self._py_db()

src/debugpy/_vendored/pydevd/tests_python/test_single_notification.py

@@ -292,3 +292,68 @@ def test_single_notification_4(single_notification_behavior, notification_queue,
     join_thread(t2)
     wait_for_notification(notification_queue, "resume")
     assert notification_queue.qsize() == 0
+
+
+def test_single_notification_5(single_notification_behavior, notification_queue, _dummy_pydb):
+    """
+    5. Continue while running
+
+    - breakpoint hits -> send notification
+    - user presses continue -> resume notification -> continue is answered
+    - user presses continue again while everything is running
+      - no resume notification will come, so, it must be answered right away
+    """
+    thread_info = _ThreadInfo()
+
+    single_notification_behavior.increment_suspend_time()
+    thread_info.state = STATE_SUSPEND
+    t = run_as_pydevd_daemon_thread(_dummy_pydb, single_notification_behavior.do_wait_suspend, thread_info, CMD_SET_BREAK)
+    wait_for_notification(notification_queue, "suspend")
+
+    called = []
+    single_notification_behavior.add_on_resumed_callback(lambda: called.append(1))
+    assert called == []
+
+    thread_info.state = STATE_RUN
+    wait_for_notification(notification_queue, "resume")
+    join_thread(t)
+    assert called == [1]
+
+    single_notification_behavior.add_on_resumed_callback(lambda: called.append(2))
+    assert called == [1, 2]
+    assert notification_queue.qsize() == 0
+
+
+def test_single_notification_6(single_notification_behavior, notification_queue, _dummy_pydb):
+    """
+    6. Breakpoint in another thread before resuming
+
+    - breakpoint hits in first -> send notification
+    - user presses continue (answered on the resume notification)
+    - 2nd hits a breakpoint before the first leaves its wait loop, which suspends the
+      first again (so, the first never sends the resume notification)
+    - send the resume notification (and answer the continue) before the new suspend notification
+    """
+    thread_info1 = _ThreadInfo()
+    thread_info2 = _ThreadInfo()
+
+    single_notification_behavior.increment_suspend_time()
+    thread_info1.state = STATE_SUSPEND
+    t1 = run_as_pydevd_daemon_thread(_dummy_pydb, single_notification_behavior.do_wait_suspend, thread_info1, CMD_SET_BREAK)
+    wait_for_notification(notification_queue, "suspend")
+
+    single_notification_behavior.add_on_resumed_callback(lambda: notification_queue.put("continue answered"))
+
+    single_notification_behavior.increment_suspend_time()
+    thread_info2.state = STATE_SUSPEND
+    t2 = run_as_pydevd_daemon_thread(_dummy_pydb, single_notification_behavior.do_wait_suspend, thread_info2, CMD_SET_BREAK)
+    wait_for_notification(notification_queue, "resume")
+    wait_for_notification(notification_queue, "continue answered")
+    wait_for_notification(notification_queue, "suspend")
+
+    thread_info1.state = STATE_RUN
+    thread_info2.state = STATE_RUN
+    wait_for_notification(notification_queue, "resume")
+    join_thread(t1)
+    join_thread(t2)
+    assert notification_queue.qsize() == 0

tests/debugpy/test_threads.py

@@ -322,3 +322,34 @@ def poll_worker(worker, gate, parked):
         assert threads[stop.thread_id] == "poller"
 
         session.request_continue()
+
+
+def test_continue_while_running(pyfile, target, run):
+    """
+    A "continue" request that arrives when no thread is suspended must still be
+    answered.
+
+    Pressing Continue twice in quick succession sends the second request after the
+    first one has already resumed the debuggee. The "continue" response is sent with
+    the next resume notification, and with nothing suspended there is none coming, so
+    the adapter waited for it and stopped handling every later request.
+    """
+
+    @pyfile
+    def code_to_debug():
+        import debuggee
+        from debuggee import backchannel
+
+        debuggee.setup()
+        print("stop here")  # @bp
+        backchannel.receive()
+
+    with debug.Session() as session:
+        backchannel = session.open_backchannel()
+        with run(session, target(code_to_debug)):
+            session.set_breakpoints(code_to_debug, all)
+
+        session.wait_for_stop()
+        session.request_continue()
+        session.request_continue()
+        backchannel.send("done")