fixes

python/ray/_raylet.pyx

@@ -466,6 +466,19 @@ cdef execute_task(
                             ray.util.pdb.set_trace(
                                 breakpoint_uuid=debugger_breakpoint)
                         outputs = function_executor(*args, **kwargs)
+                        if ray.worker.global_worker.debugger_breakpoint != "":
+                            # If this happens, the user typed "remote" and there
+                            # were no more remote calls left in this task. In that
+                            # case we just exit the debugger.
+                            ray.experimental.internal_kv._internal_kv_put(
+                                "RAY_PDB_{}".format(
+                                    ray.worker.global_worker.debugger_breakpoint),
+                                "{\"exit_debugger\": true}")
+                            ray.experimental.internal_kv._internal_kv_del(
+                                "RAY_PDB_CONTINUE_{}".format(
+                                    ray.worker.global_worker.debugger_breakpoint)
+                            )
+                            ray.worker.global_worker.debugger_breakpoint = ""
                     task_exception = False
                 except KeyboardInterrupt as e:
                     raise TaskCancelledError(

python/ray/scripts/scripts.py

@@ -6,6 +6,7 @@
 import os
 import subprocess
 import sys
+from telnetlib import Telnet
 import time
 import urllib
 import urllib.parse
@@ -150,6 +151,30 @@ def dashboard(cluster_config_file, cluster_name, port, remote_port):
                 from None
 
 
+def continue_debug_session():
+    active_sessions = ray.experimental.internal_kv._internal_kv_list(
+            "RAY_PDB_")
+
+    for active_session in active_sessions:
+        if active_session.startswith(b"RAY_PDB_CONTINUE"):
+            print("Continuing pdb session in different process...")
+            key = b"RAY_PDB_" + active_session[len("RAY_PDB_CONTINUE_"):]
+            while True:
+                data = ray.experimental.internal_kv._internal_kv_get(key)
+                if data:
+                    session = json.loads(data)
+                    if "exit_debugger" in session:
+                        ray.experimental.internal_kv._internal_kv_del(key)
+                        return
+                    host, port = session["pdb_address"].split(":")
+                    with Telnet(host, int(port)) as tn:
+                        tn.interact()
+                    ray.experimental.internal_kv._internal_kv_del(key)
+                    continue_debug_session()
+                    return
+                time.sleep(1.0)
+
+
 @cli.command()
 @click.option(
     "--address",
@@ -158,30 +183,16 @@ def dashboard(cluster_config_file, cluster_name, port, remote_port):
     help="Override the address to connect to.")
 def debug(address):
     """Show all active breakpoints and exceptions in the Ray debugger."""
-    from telnetlib import Telnet
     if not address:
         address = services.get_ray_address_to_use_or_die()
     logger.info(f"Connecting to Ray instance at {address}.")
     ray.init(address=address, log_to_driver=False)
     while True:
+        continue_debug_session()
+
         active_sessions = ray.experimental.internal_kv._internal_kv_list(
             "RAY_PDB_")
 
-        for active_session in active_sessions:
-            if active_session.startswith(b"RAY_PDB_CONTINUE"):
-                print("Continuing pdb session in different process...")
-                key = b"RAY_PDB_" + active_session[len("RAY_PDB_CONTINUE_"):]
-                while True:
-                    data = ray.experimental.internal_kv._internal_kv_get(key)
-                    if data:
-                        session = json.loads(data)
-                        host, port = session["pdb_address"].split(":")
-                        with Telnet(host, int(port)) as tn:
-                            tn.interact()
-                        ray.experimental.internal_kv._internal_kv_del(key)
-                        continue
-                    time.sleep(1.0)
-
         print("Active breakpoints:")
         for i, active_session in enumerate(active_sessions):
             data = json.loads(

python/ray/tests/test_ray_debugger.py

@@ -64,6 +64,38 @@ def f():
     ray.get(result)
 
 
+def test_ray_debugger_recursive(shutdown_only):
+    ray.init(num_cpus=1)
+
+    @ray.remote
+    def fact(n):
+        if n < 1:
+            return n
+        ray.util.pdb.set_trace()
+        n_id = fact.remote(n - 1)
+        return n * ray.get(n_id)
+
+    result = fact.remote(5)
+
+    p = pexpect.spawn("ray debug")
+    p.expect("Enter breakpoint index or press enter to refresh: ")
+    p.sendline("0")
+    p.expect("(Pdb)")
+    p.sendline("remote")
+    p.expect("(Pdb)")
+    p.sendline("remote")
+    p.expect("(Pdb)")
+    p.sendline("remote")
+    p.expect("(Pdb)")
+    p.sendline("remote")
+    p.expect("(Pdb)")
+    p.sendline("remote")
+    p.expect("(Pdb)")
+    p.sendline("remote")
+
+    ray.get(result)
+
+
 if __name__ == "__main__":
     import pytest
     # Make subprocess happy in bazel.

python/ray/util/rpdb.py

@@ -214,11 +214,14 @@ def set_trace(breakpoint_uuid=None):
 
     Can be used within a Ray task or actor.
     """
-    frame = sys._getframe().f_back
-    rdb = connect_ray_pdb(
-        None, None, False, None,
-        breakpoint_uuid.decode() if breakpoint_uuid else None)
-    rdb.set_trace(frame=frame)
+    # If there is an active debugger already, we do not want to
+    # start another one, so "set_trace" is just a no-op in that case.
+    if ray.worker.global_worker.debugger_breakpoint == "":
+        frame = sys._getframe().f_back
+        rdb = connect_ray_pdb(
+            None, None, False, None,
+            breakpoint_uuid.decode() if breakpoint_uuid else None)
+        rdb.set_trace(frame=frame)
 
 
 def post_mortem():