Collapse Qt out of process channel class hierarchy

IPython/qt/client.py

@@ -1,31 +1,576 b''
 """ Defines a KernelClient that provides signals and slots.
 """
+import atexit
+import errno
+from threading import Thread
+import time
+
+import zmq
+# import ZMQError in top-level namespace, to avoid ugly attribute-error messages
+# during garbage collection of threads at exit:
+from zmq import ZMQError
+from zmq.eventloop import ioloop, zmqstream
+
+from IPython.external.qt import QtCore
 
 # Local imports
 from IPython.utils.traitlets import Type
-from IPython.kernel.channels import (
-    ShellChannel, IOPubChannel, StdInChannel, HBChannel
-)
+from IPython.kernel.channels import HBChannel
 from IPython.kernel import KernelClient
 
-from .kernel_mixins import (
-    QtShellChannelMixin, QtIOPubChannelMixin,
-    QtStdInChannelMixin, QtHBChannelMixin,
-    QtKernelClientMixin
-)
+from .kernel_mixins import (QtHBChannelMixin, QtKernelClientMixin)
+from .util import SuperQObject
 
-class QtShellChannel(QtShellChannelMixin, ShellChannel):
+class QtHBChannel(QtHBChannelMixin, HBChannel):
     pass
 
-class QtIOPubChannel(QtIOPubChannelMixin, IOPubChannel):
-    pass
+from IPython.core.release import kernel_protocol_version_info
 
-class QtStdInChannel(QtStdInChannelMixin, StdInChannel):
-    pass
+from IPython.kernel.channelsabc import (
+    ShellChannelABC, IOPubChannelABC, StdInChannelABC,
+)
+from IPython.utils.py3compat import string_types, iteritems
 
-class QtHBChannel(QtHBChannelMixin, HBChannel):
+major_protocol_version = kernel_protocol_version_info[0]
+
+class InvalidPortNumber(Exception):
     pass
 
+# some utilities to validate message structure, these might get moved elsewhere
+# if they prove to have more generic utility
+
+def validate_string_list(lst):
+    """Validate that the input is a list of strings.
+
+    Raises ValueError if not."""
+    if not isinstance(lst, list):
+        raise ValueError('input %r must be a list' % lst)
+    for x in lst:
+        if not isinstance(x, string_types):
+            raise ValueError('element %r in list must be a string' % x)
+
+
+def validate_string_dict(dct):
+    """Validate that the input is a dict with string keys and values.
+
+    Raises ValueError if not."""
+    for k,v in iteritems(dct):
+        if not isinstance(k, string_types):
+            raise ValueError('key %r in dict must be a string' % k)
+        if not isinstance(v, string_types):
+            raise ValueError('value %r in dict must be a string' % v)
+
+
+
+class QtZMQSocketChannel(SuperQObject, Thread):
+    """The base class for the channels that use ZMQ sockets."""
+    context = None
+    session = None
+    socket = None
+    ioloop = None
+    stream = None
+    _address = None
+    _exiting = False
+    proxy_methods = []
+
+    # Emitted when the channel is started.
+    started = QtCore.Signal()
+
+    # Emitted when the channel is stopped.
+    stopped = QtCore.Signal()
+
+    message_received = QtCore.Signal(object)
+
+    #---------------------------------------------------------------------------
+    # InProcessChannel interface
+    #---------------------------------------------------------------------------
+
+    def call_handlers_later(self, *args, **kwds):
+        """ Call the message handlers later.
+        """
+        do_later = lambda: self.call_handlers(*args, **kwds)
+        QtCore.QTimer.singleShot(0, do_later)
+
+    def process_events(self):
+        """ Process any pending GUI events.
+        """
+        QtCore.QCoreApplication.instance().processEvents()
+
+    def __init__(self, context, session, address):
+        """Create a channel.
+
+        Parameters
+        ----------
+        context : :class:`zmq.Context`
+            The ZMQ context to use.
+        session : :class:`session.Session`
+            The session to use.
+        address : zmq url
+            Standard (ip, port) tuple that the kernel is listening on.
+        """
+        super(QtZMQSocketChannel, self).__init__()
+        self.daemon = True
+
+        self.context = context
+        self.session = session
+        if isinstance(address, tuple):
+            if address[1] == 0:
+                message = 'The port number for a channel cannot be 0.'
+                raise InvalidPortNumber(message)
+            address = "tcp://%s:%i" % address
+        self._address = address
+        atexit.register(self._notice_exit)
+
+    def _notice_exit(self):
+        self._exiting = True
+
+    def _run_loop(self):
+        """Run my loop, ignoring EINTR events in the poller"""
+        while True:
+            try:
+                self.ioloop.start()
+            except ZMQError as e:
+                if e.errno == errno.EINTR:
+                    continue
+                else:
+                    raise
+            except Exception:
+                if self._exiting:
+                    break
+                else:
+                    raise
+            else:
+                break
+
+    def start(self):
+        """ Reimplemented to emit signal.
+        """
+        super(QtZMQSocketChannel, self).start()
+        self.started.emit()
+
+    def stop(self):
+        """Stop the channel's event loop and join its thread.
+
+        This calls :meth:`~threading.Thread.join` and returns when the thread
+        terminates. :class:`RuntimeError` will be raised if
+        :meth:`~threading.Thread.start` is called again.
+        """
+        if self.ioloop is not None:
+            self.ioloop.stop()
+        self.join()
+        self.close()
+        self.stopped.emit()
+
+    def close(self):
+        if self.ioloop is not None:
+            try:
+                self.ioloop.close(all_fds=True)
+            except Exception:
+                pass
+        if self.socket is not None:
+            try:
+                self.socket.close(linger=0)
+            except Exception:
+                pass
+            self.socket = None
+
+    @property
+    def address(self):
+        """Get the channel's address as a zmq url string.
+
+        These URLS have the form: 'tcp://127.0.0.1:5555'.
+        """
+        return self._address
+
+    def _queue_send(self, msg):
+        """Queue a message to be sent from the IOLoop's thread.
+
+        Parameters
+        ----------
+        msg : message to send
+
+        This is threadsafe, as it uses IOLoop.add_callback to give the loop's
+        thread control of the action.
+        """
+        def thread_send():
+            self.session.send(self.stream, msg)
+        self.ioloop.add_callback(thread_send)
+
+    def _handle_recv(self, msg):
+        """Callback for stream.on_recv.
+
+        Unpacks message, and calls handlers with it.
+        """
+        ident,smsg = self.session.feed_identities(msg)
+        msg = self.session.deserialize(smsg)
+        self.call_handlers(msg)
+
+    def call_handlers(self, msg):
+        """This method is called in the ioloop thread when a message arrives.
+
+        Subclasses should override this method to handle incoming messages.
+        It is important to remember that this method is called in the thread
+        so that some logic must be done to ensure that the application level
+        handlers are called in the application thread.
+        """
+        # Emit the generic signal.
+        self.message_received.emit(msg)
+
+
+class QtShellChannel(QtZMQSocketChannel):
+    """The shell channel for issuing request/replies to the kernel."""
+
+    command_queue = None
+    # flag for whether execute requests should be allowed to call raw_input:
+    allow_stdin = True
+    proxy_methods = [
+        'execute',
+        'complete',
+        'inspect',
+        'history',
+        'kernel_info',
+        'shutdown',
+        'is_complete',
+    ]
+
+    # Emitted when a reply has been received for the corresponding request type.
+    execute_reply = QtCore.Signal(object)
+    complete_reply = QtCore.Signal(object)
+    inspect_reply = QtCore.Signal(object)
+    history_reply = QtCore.Signal(object)
+    kernel_info_reply = QtCore.Signal(object)
+
+    def __init__(self, context, session, address):
+        super(QtShellChannel, self).__init__(context, session, address)
+        self.ioloop = ioloop.IOLoop()
+
+    def run(self):
+        """The thread's main activity.  Call start() instead."""
+        self.socket = self.context.socket(zmq.DEALER)
+        self.socket.linger = 1000
+        self.socket.setsockopt(zmq.IDENTITY, self.session.bsession)
+        self.socket.connect(self.address)
+        self.stream = zmqstream.ZMQStream(self.socket, self.ioloop)
+        self.stream.on_recv(self._handle_recv)
+        self._run_loop()
+
+    def call_handlers(self, msg):
+        super(QtShellChannel, self).call_handlers(msg)
+
+        # Catch kernel_info_reply for message spec adaptation
+        msg_type = msg['header']['msg_type']
+        if msg_type == 'kernel_info_reply':
+            self._handle_kernel_info_reply(msg)
+
+        # Emit specific signals
+        signal = getattr(self, msg_type, None)
+        if signal:
+            signal.emit(msg)
+
+    def execute(self, code, silent=False, store_history=True,
+                user_expressions=None, allow_stdin=None):
+        """Execute code in the kernel.
+
+        Parameters
+        ----------
+        code : str
+            A string of Python code.
+
+        silent : bool, optional (default False)
+            If set, the kernel will execute the code as quietly possible, and
+            will force store_history to be False.
+
+        store_history : bool, optional (default True)
+            If set, the kernel will store command history.  This is forced
+            to be False if silent is True.
+
+        user_expressions : dict, optional
+            A dict mapping names to expressions to be evaluated in the user's
+            dict. The expression values are returned as strings formatted using
+            :func:`repr`.
+
+        allow_stdin : bool, optional (default self.allow_stdin)
+            Flag for whether the kernel can send stdin requests to frontends.
+
+            Some frontends (e.g. the Notebook) do not support stdin requests.
+            If raw_input is called from code executed from such a frontend, a
+            StdinNotImplementedError will be raised.
+
+        Returns
+        -------
+        The msg_id of the message sent.
+        """
+        if user_expressions is None:
+            user_expressions = {}
+        if allow_stdin is None:
+            allow_stdin = self.allow_stdin
+
+
+        # Don't waste network traffic if inputs are invalid
+        if not isinstance(code, string_types):
+            raise ValueError('code %r must be a string' % code)
+        validate_string_dict(user_expressions)
+
+        # Create class for content/msg creation. Related to, but possibly
+        # not in Session.
+        content = dict(code=code, silent=silent, store_history=store_history,
+                       user_expressions=user_expressions,
+                       allow_stdin=allow_stdin,
+                       )
+        msg = self.session.msg('execute_request', content)
+        self._queue_send(msg)
+        return msg['header']['msg_id']
+
+    def complete(self, code, cursor_pos=None):
+        """Tab complete text in the kernel's namespace.
+
+        Parameters
+        ----------
+        code : str
+            The context in which completion is requested.
+            Can be anything between a variable name and an entire cell.
+        cursor_pos : int, optional
+            The position of the cursor in the block of code where the completion was requested.
+            Default: ``len(code)``
+
+        Returns
+        -------
+        The msg_id of the message sent.
+        """
+        if cursor_pos is None:
+            cursor_pos = len(code)
+        content = dict(code=code, cursor_pos=cursor_pos)
+        msg = self.session.msg('complete_request', content)
+        self._queue_send(msg)
+        return msg['header']['msg_id']
+
+    def inspect(self, code, cursor_pos=None, detail_level=0):
+        """Get metadata information about an object in the kernel's namespace.
+
+        It is up to the kernel to determine the appropriate object to inspect.
+
+        Parameters
+        ----------
+        code : str
+            The context in which info is requested.
+            Can be anything between a variable name and an entire cell.
+        cursor_pos : int, optional
+            The position of the cursor in the block of code where the info was requested.
+            Default: ``len(code)``
+        detail_level : int, optional
+            The level of detail for the introspection (0-2)
+
+        Returns
+        -------
+        The msg_id of the message sent.
+        """
+        if cursor_pos is None:
+            cursor_pos = len(code)
+        content = dict(code=code, cursor_pos=cursor_pos,
+            detail_level=detail_level,
+        )
+        msg = self.session.msg('inspect_request', content)
+        self._queue_send(msg)
+        return msg['header']['msg_id']
+
+    def history(self, raw=True, output=False, hist_access_type='range', **kwargs):
+        """Get entries from the kernel's history list.
+
+        Parameters
+        ----------
+        raw : bool
+            If True, return the raw input.
+        output : bool
+            If True, then return the output as well.
+        hist_access_type : str
+            'range' (fill in session, start and stop params), 'tail' (fill in n)
+             or 'search' (fill in pattern param).
+
+        session : int
+            For a range request, the session from which to get lines. Session
+            numbers are positive integers; negative ones count back from the
+            current session.
+        start : int
+            The first line number of a history range.
+        stop : int
+            The final (excluded) line number of a history range.
+
+        n : int
+            The number of lines of history to get for a tail request.
+
+        pattern : str
+            The glob-syntax pattern for a search request.
+
+        Returns
+        -------
+        The msg_id of the message sent.
+        """
+        content = dict(raw=raw, output=output, hist_access_type=hist_access_type,
+                                                                    **kwargs)
+        msg = self.session.msg('history_request', content)
+        self._queue_send(msg)
+        return msg['header']['msg_id']
+
+    def kernel_info(self):
+        """Request kernel info."""
+        msg = self.session.msg('kernel_info_request')
+        self._queue_send(msg)
+        return msg['header']['msg_id']
+
+    def _handle_kernel_info_reply(self, msg):
+        """handle kernel info reply
+
+        sets protocol adaptation version
+        """
+        adapt_version = int(msg['content']['protocol_version'].split('.')[0])
+        if adapt_version != major_protocol_version:
+            self.session.adapt_version = adapt_version
+
+    def shutdown(self, restart=False):
+        """Request an immediate kernel shutdown.
+
+        Upon receipt of the (empty) reply, client code can safely assume that
+        the kernel has shut down and it's safe to forcefully terminate it if
+        it's still alive.
+
+        The kernel will send the reply via a function registered with Python's
+        atexit module, ensuring it's truly done as the kernel is done with all
+        normal operation.
+        """
+        # Send quit message to kernel. Once we implement kernel-side setattr,
+        # this should probably be done that way, but for now this will do.
+        msg = self.session.msg('shutdown_request', {'restart':restart})
+        self._queue_send(msg)
+        return msg['header']['msg_id']
+
+    def is_complete(self, code):
+        msg = self.session.msg('is_complete_request', {'code': code})
+        self._queue_send(msg)
+        return msg['header']['msg_id']
+
+
+class QtIOPubChannel(QtZMQSocketChannel):
+    """The iopub channel which listens for messages that the kernel publishes.
+
+    This channel is where all output is published to frontends.
+    """
+    # Emitted when a message of type 'stream' is received.
+    stream_received = QtCore.Signal(object)
+
+    # Emitted when a message of type 'execute_input' is received.
+    execute_input_received = QtCore.Signal(object)
+
+    # Emitted when a message of type 'execute_result' is received.
+    execute_result_received = QtCore.Signal(object)
+
+    # Emitted when a message of type 'error' is received.
+    error_received = QtCore.Signal(object)
+
+    # Emitted when a message of type 'display_data' is received
+    display_data_received = QtCore.Signal(object)
+
+    # Emitted when a crash report message is received from the kernel's
+    # last-resort sys.excepthook.
+    crash_received = QtCore.Signal(object)
+
+    # Emitted when a shutdown is noticed.
+    shutdown_reply_received = QtCore.Signal(object)
+
+    def __init__(self, context, session, address):
+        super(QtIOPubChannel, self).__init__(context, session, address)
+        self.ioloop = ioloop.IOLoop()
+
+    def run(self):
+        """The thread's main activity.  Call start() instead."""
+        self.socket = self.context.socket(zmq.SUB)
+        self.socket.linger = 1000
+        self.socket.setsockopt(zmq.SUBSCRIBE,b'')
+        self.socket.setsockopt(zmq.IDENTITY, self.session.bsession)
+        self.socket.connect(self.address)
+        self.stream = zmqstream.ZMQStream(self.socket, self.ioloop)
+        self.stream.on_recv(self._handle_recv)
+        self._run_loop()
+
+    def call_handlers(self, msg):
+        super(QtIOPubChannel, self).call_handlers(msg)
+
+        # Emit signals for specialized message types.
+        msg_type = msg['header']['msg_type']
+        signal = getattr(self, msg_type + '_received', None)
+        if signal:
+            signal.emit(msg)
+
+    def flush(self, timeout=1.0):
+        """Immediately processes all pending messages on the iopub channel.
+
+        Callers should use this method to ensure that :meth:`call_handlers`
+        has been called for all messages that have been received on the
+        0MQ SUB socket of this channel.
+
+        This method is thread safe.
+
+        Parameters
+        ----------
+        timeout : float, optional
+            The maximum amount of time to spend flushing, in seconds. The
+            default is one second.
+        """
+        # We do the IOLoop callback process twice to ensure that the IOLoop
+        # gets to perform at least one full poll.
+        stop_time = time.time() + timeout
+        for i in range(2):
+            self._flushed = False
+            self.ioloop.add_callback(self._flush)
+            while not self._flushed and time.time() < stop_time:
+                time.sleep(0.01)
+
+    def _flush(self):
+        """Callback for :method:`self.flush`."""
+        self.stream.flush()
+        self._flushed = True
+
+
+class QtStdInChannel(QtZMQSocketChannel):
+    """The stdin channel to handle raw_input requests that the kernel makes."""
+
+    msg_queue = None
+    proxy_methods = ['input']
+
+    # Emitted when an input request is received.
+    input_requested = QtCore.Signal(object)
+
+    def __init__(self, context, session, address):
+        super(QtStdInChannel, self).__init__(context, session, address)
+        self.ioloop = ioloop.IOLoop()
+
+    def run(self):
+        """The thread's main activity.  Call start() instead."""
+        self.socket = self.context.socket(zmq.DEALER)
+        self.socket.linger = 1000
+        self.socket.setsockopt(zmq.IDENTITY, self.session.bsession)
+        self.socket.connect(self.address)
+        self.stream = zmqstream.ZMQStream(self.socket, self.ioloop)
+        self.stream.on_recv(self._handle_recv)
+        self._run_loop()
+
+    def call_handlers(self, msg):
+        super(QtStdInChannel, self).call_handlers(msg)
+
+        # Emit signals for specialized message types.
+        msg_type = msg['header']['msg_type']
+        if msg_type == 'input_request':
+            self.input_requested.emit(msg)
+
+    def input(self, string):
+        """Send a string of raw input to the kernel."""
+        content = dict(value=string)
+        msg = self.session.msg('input_reply', content)
+        self._queue_send(msg)
+
+ShellChannelABC.register(QtShellChannel)
+IOPubChannelABC.register(QtIOPubChannel)
+StdInChannelABC.register(QtStdInChannel)
+
 
 class QtKernelClient(QtKernelClientMixin, KernelClient):
     """ A KernelClient that provides signals and slots.