Plugged debugger

Author: JohanMabille

ipykernel/debugger.py

@@ -2,9 +2,10 @@
 import os
 
 from zmq.utils import jsonapi
-from traitlets import Instance
+#from traitlets import Instance
+#from traitlets.config.configurable import SingletonConfigurable
 
-from asyncio import Queue
+from asyncio import (Event, Queue)
 
 class DebugpyMessageQueue:
 
@@ -26,7 +27,6 @@ def _reset_tcp_pos(self):
         self.message_pos = -1
 
     def _put_message(self, raw_msg):
-        # TODO: forward to iopub if this is an event message
         msg = jsonapi.loads(raw_msg)
         if mes['type'] == 'event':
             self.event_callback(msg)
@@ -72,23 +72,23 @@ async def get_message(self):
 
 class DebugpyClient:
 
-    def __init__(self, debugpy_socket, debugpy_stream):
-        self.debugpy_socket = debugpy_socket
+    def __init__(self, debugpy_stream, event_callback):
         self.debugpy_stream = debugpy_stream
+        self.event_callback = event_callback
         self.message_queue = DebugpyMessageQueue(self._forward_event)
         self.wait_for_attach = True
-        self.init_event = asyncio.Event()
+        self.init_event = Event()
 
     def _forward_event(self, msg):
         if msg['event'] == 'initialized':
             self.init_event.set()
-        #TODO: send event to iopub
+        self.event_callback(msg)
 
     def _send_request(self, msg):
         content = jsonapi.dumps(msg)
         content_length = len(content)
         buf = DebugpyMessageQueue.HEADER + content_length + DebugpyMessageQueue.SEPARATOR + content_msg
-        self.debugpy_socket.send(buf) # TODO: pass routing_id 
+        self.debugpy_stream.send(buf) # TODO: pass routing_id 
 
     async def _wait_for_reponse(self):
         # Since events are never pushed to the message_queue
@@ -115,6 +115,9 @@ async def _handle_init_sequence(self):
         attach_rep = await self._wait_for_response()
         return attach_rep
 
+    def receive_dap_frame(self, frame):
+        self.message_queue.put_tcp_frame(frame)
+
     async def send_dap_request(self, msg):
         self._send_request(msg)
         if self.wait_for_attach and msg['command'] == 'attach':
@@ -140,19 +143,20 @@ class Debugger:
         'debugInfo', 'inspectVariables'
     ]
 
-    log = Instance(logging.Logger, allow_none=True)
+    #log = Instance(logging.Logger, allow_none=True)
 
-    def __init__(self):
+    def __init__(self, debugpy_stream, event_callback, shell_socket, session):
+        self.debugpy_client = DebugpyClient(debugpy_stream, event_callback)
+        self.shell_socket = shell_socket
+        self.session = session
         self.is_started = False
-
-        self.header = ''
 
         self.started_debug_handlers = {}
-        for msg_type in started_debug_msg_types:
+        for msg_type in Debugger.started_debug_msg_types:
             self.started_debug_handlers[msg_type] = getattr(self, msg_type)
 
         self.static_debug_handlers = {}
-        for msg_type in static_debug_msg_types:
+        for msg_type in Debugger.static_debug_msg_types:
             self.static_debug_handlers[msg_type] = getattr(self, msg_type)
 
         self.breakpoint_list = {}
@@ -161,10 +165,27 @@ def __init__(self):
     async def _forward_message(self, msg):
         return await self.debugpy_client.send_dap_request(msg)
 
+    @property
+    def tcp_client(self):
+        return self.debugpy_client
+
     def start(self):
+        endpoint = self.debugpy_client.debugpy_stream.socket.getsockopt(zmq.LAST_ENDPOINT)
+        index = endpoit.rfind(':')
+        port = endpoint[index+1:]
+        code = 'import debugpy;'
+        code += 'debugpy.listen(("127.0.0.1",' + port + '))'
+        content = {
+            'code': code,
+            'slient': True
+        }
+        self.session.send(self.shell_socket, 'execute_request', content,
+                          None, (self.shell_socket.getsockopt(zmq.ROUTING_ID)))
+                          
         return False
 
     def stop(self):
+        # TODO 
         pass
 
     def dumpCell(self, message):
@@ -198,7 +219,7 @@ def source(self, message):
 
     async def stackTrace(self, message):
         reply = await self._forward_message(message)
-        reply['body']['stackFrames'] =
+        reply['body']['stackFrames'] = \
             [frame for frame in reply['body']['stackFrames'] if frame['source']['path'] != '<string>']
         return reply
 
@@ -215,7 +236,7 @@ async def attach(self, message):
         message['arguments']['logToFile'] = True
         return await self._forward_message(message)
 
-    def configurationDone(self, message):
+    async def configurationDone(self, message):
         reply = {
             'seq': message['seq'],
             'type': 'response',
@@ -225,7 +246,13 @@ def configurationDone(self, message):
         }
         return reply;
 
-    def debugInfo(self, message):
+    async def debugInfo(self, message):
+        breakpoint_list = []
+        for key, value in self.breakpoint_list.items():
+            breakpoint_list.append({
+                'source': key,
+                'breakpoints': value
+            })
         reply = {
             'type': 'response',
             'request_seq': message['seq'],
@@ -235,18 +262,21 @@ def debugInfo(self, message):
                 'isStarted': self.is_started,
                 'hashMethod': 'Murmur2',
                 'hashSeed': 0,
-                'tmpFilePrefix': 'coincoin',
+                'tmpFilePrefix': '/tmp/ipykernel_debugger',
                 'tmpFileSuffix': '.py',
-                'breakpoints': self.breakpoint_list,
+                'breakpoints': breakpoint_list,
                 'stoppedThreads': self.stopped_threads
             }
         }
+        #self.log.info("returning reply %s", reply)
+        print("DEBUGGER: ", reply)
         return reply
 
     def inspectVariables(self, message):
+        # TODO
         return {}
 
-    async def process_request(self, header, message):
+    async def process_request(self, message):
         reply = {}
 
         if message['command'] == 'initialize':
@@ -269,11 +299,10 @@ async def process_request(self, header, message):
         if handler is not None:
             reply = await handler(message)
         elif self.is_started:
-            self.header = header
             handler = self.started_debug_handlers.get(message['command'], None)
             if handler is not None:
                 reply = await handler(message)
-            else
+            else:
                 reply = await self._forward_message(message)
 
         if message['command'] == 'disconnect':

ipykernel/kernelapp.py

@@ -122,6 +122,8 @@ class IPKernelApp(BaseIPythonApplication, InteractiveShellApp,
     context = Any()
     shell_socket = Any()
     control_socket = Any()
+    debugpy_socket = Any()
+    debug_shell_socket = Any()
     stdin_socket = Any()
     iopub_socket = Any()
     iopub_thread = Any()
@@ -294,6 +296,16 @@ def init_control(self, context):
         self.control_port = self._bind_socket(self.control_socket, self.control_port)
         self.log.debug("control ROUTER Channel on port: %i" % self.control_port)
 
+        self.debugpy_socket = context.socket(zmq.STREAM)
+        self.debugpy_socket.linger = 1000
+        self.debugpy_port = 0
+        self.debugpy_port = self._bind_socket(self.debugpy_socket, self.debugpy_port)
+        self.log.debug("debugpy STREAM Channel on port: %i" % self.debugpy_port)
+
+        self.debug_shell_socket = context.socket(zmq.DEALER)
+        self.debug_shell_socket.linger = 1000
+        self.debug_shell_socket.connect(self.shell_socket.getsockopt(zmq.LAST_ENDPOINT))
+
         if hasattr(zmq, 'ROUTER_HANDOVER'):
             # set router-handover to workaround zeromq reconnect problems
             # in certain rare circumstances
@@ -335,6 +347,9 @@ def close(self):
             self.log.debug("Closing iopub channel")
             self.iopub_thread.stop()
             self.iopub_thread.close()
+
+        self.debug_shell_socket.close()
+
         for channel in ('shell', 'control', 'stdin'):
             self.log.debug("Closing %s channel", channel)
             socket = getattr(self, channel + "_socket", None)
@@ -449,11 +464,14 @@ def init_kernel(self):
         """Create the Kernel object itself"""
         shell_stream = ZMQStream(self.shell_socket)
         control_stream = ZMQStream(self.control_socket, self.control_thread.io_loop)
+        debugpy_stream = ZMQStream(self.debugpy_socket, self.control_thread.io_loop)
         self.control_thread.start()
         kernel_factory = self.kernel_class.instance
 
         kernel = kernel_factory(parent=self, session=self.session,
                                 control_stream=control_stream,
+                                debugpy_stream=debugpy_stream,
+                                debug_shell_socket=self.debug_shell_socket,
                                 shell_stream=shell_stream, 
                                 iopub_thread=self.iopub_thread,
                                 iopub_socket=self.iopub_socket,

ipykernel/kernelbase.py

@@ -12,6 +12,7 @@
 import time
 import uuid
 import warnings
+import asyncio
 
 try:
     # jupyter_client >= 5, use tz-aware now
@@ -39,6 +40,7 @@
 
 from ._version import kernel_protocol_version
 
+from .debugger import Debugger
 
 class Kernel(SingletonConfigurable):
 
@@ -69,6 +71,10 @@ def shell_streams(self):
         return [shell_stream]
 
     control_stream = Instance(ZMQStream, allow_none=True)
+    debugpy_stream = Instance(ZMQStream, allow_none=True)
+
+    debug_shell_socket = Any()
+
     iopub_socket = Any()
     iopub_thread = Any()
     stdin_socket = Any()
@@ -160,7 +166,7 @@ def _default_ident(self):
         'apply_request',
     ]
     # add deprecated ipyparallel control messages
-    control_msg_types = msg_types + ['clear_request', 'abort_request']
+    control_msg_types = msg_types + ['clear_request', 'abort_request', 'debug_request']
 
     def __init__(self, **kwargs):
         super(Kernel, self).__init__(**kwargs)
@@ -173,6 +179,16 @@ def __init__(self, **kwargs):
         for msg_type in self.control_msg_types:
             self.control_handlers[msg_type] = getattr(self, msg_type)
 
+        self.debugger = Debugger(self.debugpy_stream,
+                                 self._publish_debug_event,
+                                 self.debug_shell_socket,
+                                 self.session)
+
+    @gen.coroutine
+    def dispatch_debugpy(self, msg):
+        self.log.debug("Debugpy received: %s", msg)
+        selg.debugger.tcp_client.receive_dap_frame(msg)
+
     @gen.coroutine
     def dispatch_control(self, msg):
         """dispatch control requests"""
@@ -366,7 +382,12 @@ def dispatch_queue(self):
         Ensures that only one message is processing at a time,
         even when the handler is async
         """
+
         while True:
+            # ensure control stream is flushed before processing shell messages
+            if self.control_stream:
+                self.control_stream.flush()
+            # receive the next message and handle it
             try:
                 yield self.process_one()
             except Exception:
@@ -401,6 +422,7 @@ def start(self):
         self.io_loop.add_callback(self.dispatch_queue)
 
         self.control_stream.on_recv(self.dispatch_control, copy=False)
+        self.debugpy_stream.on_recv(self.dispatch_debugpy, copy=False)
 
         self.shell_stream.on_recv(
             partial(
@@ -442,6 +464,13 @@ def _publish_status(self, status, channel, parent=None):
                           parent=parent or self._parent_header[channel],
                           ident=self._topic('status'),
                           )
+    def _publish_debug_event(self, event):
+        self.session.send(self.iopub_socket,
+                          'debug_event',
+                          event,
+                          parent=self._parent_header['control'],
+                          ident=self._topic('debug_event')
+        )
 
     def set_parent(self, ident, parent, channel='shell'):
         """Set the current parent_header
@@ -519,6 +548,7 @@ def execute_request(self, stream, ident, parent):
             )
         )
 
+        self.log.debug("EXECUTE_REPLY: %s", reply_content)
         # Flush output before sending the reply.
         sys.stdout.flush()
         sys.stderr.flush()
@@ -693,6 +723,20 @@ def do_is_complete(self, code):
         return {'status' : 'unknown',
                 }
 
+    @gen.coroutine
+    def debug_request(self, stream, ident, parent):
+        content = parent['content']
+
+        reply_content = yield gen.maybe_future(self.do_debug_request(content))
+        reply_content = json_clean(reply_content)
+        reply_msg = self.session.send(stream, 'debug_reply', reply_content,
+                                      parent, ident)
+        self.log.debug("%s", reply_msg)
+
+    @gen.coroutine
+    def do_debug_request(self, msg):
+        return (yield self.debugger.process_request(msg))
+
     #---------------------------------------------------------------------------
     # Engine methods (DEPRECATED)
     #---------------------------------------------------------------------------

ipykernel/kernelspec.py

@@ -55,6 +55,7 @@ def get_kernel_dict(extra_arguments=None):
         'argv': make_ipkernel_cmd(extra_arguments=extra_arguments),
         'display_name': 'Python %i' % sys.version_info[0],
         'language': 'python',
+        'metadata': { 'debugger': True}
     }