Support re-entrant breakpoints

Keep an actor local (bool) flag which determines if there is already
a running debugger instance for the current process. If another task
tries to enter in this case, simply ignore it since allowing entry may
result in a deadlock where the new task will be sync waiting on the
parent stdio lock (a case that will never arrive due to the current
debugger's active use of it).

In the future we may want to allow FIFO queueing of local tasks where
instead of ignoring re-entrant breakpoints we allow tasks to async wait
for debugger release, though not sure the implications of that since
you'd likely want to support switching the debugger to the new task and
that could cause deadlocks where tasks are inter-dependent. It may be
more sane to just error on multiple breakpoint requests within an actor.

tractor/_debug.py

@@ -36,8 +36,8 @@ class TractorConfig(pdbpp.DefaultConfig):
     """
     sticky_by_default = True
 
-    def teardown(self, _pdb):
-        _pdb_release_hook(_pdb)
+    def teardown(self):
+        _pdb_release_hook()
 
 
 class PdbwTeardown(pdbpp.Pdb):
@@ -50,11 +50,11 @@ class PdbwTeardown(pdbpp.Pdb):
     # since that'll cause deadlock for us.
     def set_continue(self):
         super().set_continue()
-        self.config.teardown(self)
+        self.config.teardown()
 
     def set_quit(self):
         super().set_quit()
-        self.config.teardown(self)
+        self.config.teardown()
 
 
 # TODO: will be needed whenever we get to true remote debugging.
@@ -140,36 +140,48 @@ async def wait_for_parent_stdin_hijack(
 
         # TODO: need a more robust check for the "root" actor
         if actor._parent_chan:
-            async with tractor._portal.open_portal(
-                actor._parent_chan,
-                start_msg_loop=False,
-                shield=True,
-            ) as portal:
-                # with trio.fail_after(1):
-                agen = await portal.run(
-                    'tractor._debug',
-                    '_hijack_stdin_relay_to_child',
-                    subactor_uid=actor.uid,
-                )
-                async with aclosing(agen):
-                    async for val in agen:
-                        assert val == 'Locked'
-                        task_status.started()
-                        with trio.CancelScope(shield=True):
-                            await do_unlock.wait()
-
-                            # trigger cancellation of remote stream
-                            break
-
+            try:
+                async with tractor._portal.open_portal(
+                    actor._parent_chan,
+                    start_msg_loop=False,
+                    shield=True,
+                ) as portal:
+                    # with trio.fail_after(1):
+                    agen = await portal.run(
+                        'tractor._debug',
+                        '_hijack_stdin_relay_to_child',
+                        subactor_uid=actor.uid,
+                    )
+                    async with aclosing(agen):
+                        async for val in agen:
+                            assert val == 'Locked'
+                            task_status.started()
+                            with trio.CancelScope(shield=True):
+                                await do_unlock.wait()
+
+                                # trigger cancellation of remote stream
+                                break
+            finally:
+                log.debug(f"Exiting debugger for actor {actor}")
+                actor.statespace['_in_debug'] = False
                 log.debug(f"Child {actor} released parent stdio lock")
 
-    def unlock(_pdb):
-        do_unlock.set()
+    async def _bp():
+        """Async breakpoint which schedules a parent stdio lock, and once complete
+        enters the ``pdbpp`` debugging console.
+        """
+        in_debug = actor.statespace.setdefault('_in_debug', False)
 
-    global _pdb_release_hook
-    _pdb_release_hook = unlock
+        if in_debug:
+            log.warning(f"Actor {actor} already has a debug lock, skipping...")
+            return
+
+        # assign unlock callback for debugger teardown hooks
+        global _pdb_release_hook
+        _pdb_release_hook = do_unlock.set
+
+        actor.statespace['_in_debug'] = True
 
-    async def _bp():
         # this **must** be awaited by the caller and is done using the
         # root nursery so that the debugger can continue to run without
         # being restricted by the scope of a new task nursery.
@@ -179,7 +191,8 @@ async def _bp():
         # ``breakpoint()`` was awaited and begin handling stdio
         debug_func(actor)
 
-    return _bp()  # user code **must** await this!
+    # user code **must** await this!
+    return _bp()
 
 
 def _set_trace(actor):