debugger_r.py

"""Support for remote Python debugging.

Some ASCII art to describe the structure:

 IN PYTHON SUBPROCESS # IN IDLE PROCESS
 #
 # oid='gui_adapter'
 +----------+ # +------------+ +-----+
 | GUIProxy |--remote#call-->| GUIAdapter |--calls-->| GUI |
+-----+--calls-->+----------+ # +------------+ +-----+
| Idb | # /
+-----+<-calls--+------------+ # +----------+<--calls-/
 | IdbAdapter |<--remote#call--| IdbProxy |
 +------------+ # +----------+
 oid='idb_adapter' #

The purpose of the Proxy and Adapter classes is to translate certain
arguments and return values that cannot be transported through the RPC
barrier, in particular frame and traceback objects.

"""
importreprlib
importtypes
fromidlelibimportdebugger

debugging=0

idb_adap_oid="idb_adapter"
gui_adap_oid="gui_adapter"

#=======================================
#
# In the PYTHON subprocess:

frametable= {}
dicttable= {}
codetable= {}
tracebacktable= {}

defwrap_frame(frame):
fid=id(frame)
frametable[fid] =frame
returnfid

defwrap_info(info):
"replace info[2], a traceback instance, by its ID"
ifinfoisNone:
returnNone
else:
traceback=info[2]
assertisinstance(traceback, types.TracebackType)
traceback_id=id(traceback)
tracebacktable[traceback_id] =traceback
modified_info= (info[0], info[1], traceback_id)
returnmodified_info

classGUIProxy:

def__init__(self, conn, gui_adap_oid):
self.conn=conn
self.oid=gui_adap_oid

definteraction(self, message, frame, info=None):
# calls rpc.SocketIO.remotecall() via run.MyHandler instance
# pass frame and traceback object IDs instead of the objects themselves
self.conn.remotecall(self.oid, "interaction",
 (message, wrap_frame(frame), wrap_info(info)),
 {})

classIdbAdapter:

def__init__(self, idb):
self.idb=idb

#----------called by an IdbProxy----------

defset_step(self):
self.idb.set_step()

defset_quit(self):
self.idb.set_quit()

defset_continue(self):
self.idb.set_continue()

defset_next(self, fid):
frame=frametable[fid]
self.idb.set_next(frame)

defset_return(self, fid):
frame=frametable[fid]
self.idb.set_return(frame)

defget_stack(self, fid, tbid):
frame=frametable[fid]
iftbidisNone:
tb=None
else:
tb=tracebacktable[tbid]
stack, i=self.idb.get_stack(frame, tb)
stack= [(wrap_frame(frame2), k) forframe2, kinstack]
returnstack, i

defrun(self, cmd):
import__main__
self.idb.run(cmd, __main__.__dict__)

defset_break(self, filename, lineno):
msg=self.idb.set_break(filename, lineno)
returnmsg

defclear_break(self, filename, lineno):
msg=self.idb.clear_break(filename, lineno)
returnmsg

defclear_all_file_breaks(self, filename):
msg=self.idb.clear_all_file_breaks(filename)
returnmsg

#----------called by a FrameProxy----------

defframe_attr(self, fid, name):
frame=frametable[fid]
returngetattr(frame, name)

defframe_globals(self, fid):
frame=frametable[fid]
gdict=frame.f_globals
did=id(gdict)
dicttable[did] =gdict
returndid

defframe_locals(self, fid):
frame=frametable[fid]
ldict=frame.f_locals
did=id(ldict)
dicttable[did] =ldict
returndid

defframe_code(self, fid):
frame=frametable[fid]
code=frame.f_code
cid=id(code)
codetable[cid] =code
returncid

#----------called by a CodeProxy----------

defcode_name(self, cid):
code=codetable[cid]
returncode.co_name

defcode_filename(self, cid):
code=codetable[cid]
returncode.co_filename

#----------called by a DictProxy----------

defdict_keys(self, did):
raiseNotImplementedError("dict_keys not public or pickleable")
## return dicttable[did].keys()

### Needed until dict_keys type is finished and pickleable.
# xxx finished. pickleable?
### Will probably need to extend rpc.py:SocketIO._proxify at that time.
defdict_keys_list(self, did):
returnlist(dicttable[did].keys())

defdict_item(self, did, key):
value=dicttable[did][key]
returnreprlib.repr(value) # Can't pickle module 'builtins'.

#----------end class IdbAdapter----------


defstart_debugger(rpchandler, gui_adap_oid):
"""Start the debugger and its RPC link in the Python subprocess

 Start the subprocess side of the split debugger and set up that side of the
 RPC link by instantiating the GUIProxy, Idb debugger, and IdbAdapter
 objects and linking them together. Register the IdbAdapter with the
 RPCServer to handle RPC requests from the split debugger GUI via the
 IdbProxy.

 """
gui_proxy=GUIProxy(rpchandler, gui_adap_oid)
idb=debugger.Idb(gui_proxy)
idb_adap=IdbAdapter(idb)
rpchandler.register(idb_adap_oid, idb_adap)
returnidb_adap_oid


#=======================================
#
# In the IDLE process:


classFrameProxy:

def__init__(self, conn, fid):
self._conn=conn
self._fid=fid
self._oid="idb_adapter"
self._dictcache= {}

def__getattr__(self, name):
ifname[:1] =="_":
raiseAttributeError(name)
ifname=="f_code":
returnself._get_f_code()
ifname=="f_globals":
returnself._get_f_globals()
ifname=="f_locals":
returnself._get_f_locals()
returnself._conn.remotecall(self._oid, "frame_attr",
 (self._fid, name), {})

def_get_f_code(self):
cid=self._conn.remotecall(self._oid, "frame_code", (self._fid,), {})
returnCodeProxy(self._conn, self._oid, cid)

def_get_f_globals(self):
did=self._conn.remotecall(self._oid, "frame_globals",
 (self._fid,), {})
returnself._get_dict_proxy(did)

def_get_f_locals(self):
did=self._conn.remotecall(self._oid, "frame_locals",
 (self._fid,), {})
returnself._get_dict_proxy(did)

def_get_dict_proxy(self, did):
ifdidinself._dictcache:
returnself._dictcache[did]
dp=DictProxy(self._conn, self._oid, did)
self._dictcache[did] =dp
returndp


classCodeProxy:

def__init__(self, conn, oid, cid):
self._conn=conn
self._oid=oid
self._cid=cid

def__getattr__(self, name):
ifname=="co_name":
returnself._conn.remotecall(self._oid, "code_name",
 (self._cid,), {})
ifname=="co_filename":
returnself._conn.remotecall(self._oid, "code_filename",
 (self._cid,), {})


classDictProxy:

def__init__(self, conn, oid, did):
self._conn=conn
self._oid=oid
self._did=did

## def keys(self):
## return self._conn.remotecall(self._oid, "dict_keys", (self._did,), {})

# 'temporary' until dict_keys is a pickleable built-in type
defkeys(self):
returnself._conn.remotecall(self._oid,
"dict_keys_list", (self._did,), {})

def__getitem__(self, key):
returnself._conn.remotecall(self._oid, "dict_item",
 (self._did, key), {})

def__getattr__(self, name):
##print("*** Failed DictProxy.__getattr__:", name)
raiseAttributeError(name)


classGUIAdapter:

def__init__(self, conn, gui):
self.conn=conn
self.gui=gui

definteraction(self, message, fid, modified_info):
##print("*** Interaction: (%s, %s, %s)" % (message, fid, modified_info))
frame=FrameProxy(self.conn, fid)
self.gui.interaction(message, frame, modified_info)


classIdbProxy:

def__init__(self, conn, shell, oid):
self.oid=oid
self.conn=conn
self.shell=shell

defcall(self, methodname, /, *args, **kwargs):
##print("*** IdbProxy.call %s %s %s" % (methodname, args, kwargs))
value=self.conn.remotecall(self.oid, methodname, args, kwargs)
##print("*** IdbProxy.call %s returns %r" % (methodname, value))
returnvalue

defrun(self, cmd, locals):
# Ignores locals on purpose!
seq=self.conn.asyncqueue(self.oid, "run", (cmd,), {})
self.shell.interp.active_seq=seq

defget_stack(self, frame, tbid):
# passing frame and traceback IDs, not the objects themselves
stack, i=self.call("get_stack", frame._fid, tbid)
stack= [(FrameProxy(self.conn, fid), k) forfid, kinstack]
returnstack, i

defset_continue(self):
self.call("set_continue")

defset_step(self):
self.call("set_step")

defset_next(self, frame):
self.call("set_next", frame._fid)

defset_return(self, frame):
self.call("set_return", frame._fid)

defset_quit(self):
self.call("set_quit")

defset_break(self, filename, lineno):
msg=self.call("set_break", filename, lineno)
returnmsg

defclear_break(self, filename, lineno):
msg=self.call("clear_break", filename, lineno)
returnmsg

defclear_all_file_breaks(self, filename):
msg=self.call("clear_all_file_breaks", filename)
returnmsg

defstart_remote_debugger(rpcclt, pyshell):
"""Start the subprocess debugger, initialize the debugger GUI and RPC link

 Request the RPCServer start the Python subprocess debugger and link. Set
 up the Idle side of the split debugger by instantiating the IdbProxy,
 debugger GUI, and debugger GUIAdapter objects and linking them together.

 Register the GUIAdapter with the RPCClient to handle debugger GUI
 interaction requests coming from the subprocess debugger via the GUIProxy.

 The IdbAdapter will pass execution and environment requests coming from the
 Idle debugger GUI to the subprocess debugger via the IdbProxy.

 """
globalidb_adap_oid

idb_adap_oid=rpcclt.remotecall("exec", "start_the_debugger",\
 (gui_adap_oid,), {})
idb_proxy=IdbProxy(rpcclt, pyshell, idb_adap_oid)
gui=debugger.Debugger(pyshell, idb_proxy)
gui_adap=GUIAdapter(rpcclt, gui)
rpcclt.register(gui_adap_oid, gui_adap)
returngui

defclose_remote_debugger(rpcclt):
"""Shut down subprocess debugger and Idle side of debugger RPC link

 Request that the RPCServer shut down the subprocess debugger and link.
 Unregister the GUIAdapter, which will cause a GC on the Idle process
 debugger and RPC link objects. (The second reference to the debugger GUI
 is deleted in pyshell.close_remote_debugger().)

 """
close_subprocess_debugger(rpcclt)
rpcclt.unregister(gui_adap_oid)

defclose_subprocess_debugger(rpcclt):
rpcclt.remotecall("exec", "stop_the_debugger", (idb_adap_oid,), {})

defrestart_subprocess_debugger(rpcclt):
idb_adap_oid_ret=rpcclt.remotecall("exec", "start_the_debugger",\
 (gui_adap_oid,), {})
assertidb_adap_oid_ret==idb_adap_oid, 'Idb restarted with different oid'


if__name__=="__main__":
fromunittestimportmain
main('idlelib.idle_test.test_debugger_r', verbosity=2, exit=False)