Kernel manager is cleaned up and simplified. Still bugs though.

IPython/frontend/qt/console/frontend_widget.py

@@ -175,7 +175,7 b' class FrontendWidget(HistoryConsoleWidget):'
             xreq.object_info_reply.disconnect(self._handle_object_info_reply)
 
             # Handle the case where the old kernel manager is still listening.
-            if self._kernel_manager.is_listening:
+            if self._kernel_manager.channels_running:
                 self._stopped_listening()
 
         # Set the new kernel manager.
@@ -197,7 +197,7 b' class FrontendWidget(HistoryConsoleWidget):'
         
         # Handle the case where the kernel manager started listening before
         # we connected.
-        if kernel_manager.is_listening:
+        if kernel_manager.channels_running:
             self._started_listening()
 
     kernel_manager = property(_get_kernel_manager, _set_kernel_manager)

IPython/frontend/qt/console/ipython_widget.py

@@ -79,7 +79,7 b" if __name__ == '__main__':"
     # Create a KernelManager.
     kernel_manager = QtKernelManager()
     kernel_manager.start_kernel()
-    kernel_manager.start_listening()
+    kernel_manager.start_channels()
 
     # Don't let Qt or ZMQ swallow KeyboardInterupts.
     # FIXME: Gah, ZMQ swallows even custom signal handlers. So for now we leave 
@@ -92,7 +92,7 b" if __name__ == '__main__':"
     # Create the application, making sure to clean up nicely when we exit.
     app = QtGui.QApplication([])
     def quit_hook():
-        kernel_manager.stop_listening()
+        kernel_manager.stop_channels()
         kernel_manager.kill_kernel()
     app.aboutToQuit.connect(quit_hook)
 

IPython/frontend/qt/kernelmanager.py

@@ -128,14 +128,14 b' class QtKernelManager(KernelManager, QtCore.QObject):'
     # 'KernelManager' interface
     #---------------------------------------------------------------------------
     
-    def start_listening(self):
+    def start_channels(self):
         """ Reimplemented to emit signal.
         """
-        super(QtKernelManager, self).start_listening()
+        super(QtKernelManager, self).start_channels()
         self.started_listening.emit()
 
-    def stop_listening(self):
+    def stop_channels(self):
         """ Reimplemented to emit signal.
         """ 
-        super(QtKernelManager, self).stop_listening()
+        super(QtKernelManager, self).stop_channels()
         self.stopped_listening.emit()

IPython/zmq/kernelmanager.py

@@ -1,15 +1,27 b''
-"""Kernel frontend classes.
+"""Classes to manage the interaction with a running kernel.
 
-TODO: Create logger to handle debugging and console messages.
+Todo
+====
 
+* Create logger to handle debugging and console messages.
 """
 
+#-----------------------------------------------------------------------------
+#  Copyright (C) 2008-2010  The IPython Development Team
+#
+#  Distributed under the terms of the BSD License.  The full license is in
+#  the file COPYING, distributed as part of this software.
+#-----------------------------------------------------------------------------
+
+#-----------------------------------------------------------------------------
+# Imports
+#-----------------------------------------------------------------------------
+
 # Standard library imports.
 from Queue import Queue, Empty
 from subprocess import Popen
 from threading import Thread
 import time
-import traceback
 
 # System library imports.
 import zmq
@@ -17,66 +29,80 b' from zmq import POLLIN, POLLOUT, POLLERR'
 from zmq.eventloop import ioloop
 
 # Local imports.
-from IPython.utils.traitlets import HasTraits, Any, Bool, Int, Instance, Str, \
-    Type
+from IPython.utils.traitlets import HasTraits, Any, Instance, Type
 from kernel import launch_kernel
 from session import Session
 
-# Constants.
+#-----------------------------------------------------------------------------
+# Constants and exceptions
+#-----------------------------------------------------------------------------
+
 LOCALHOST = '127.0.0.1'
 
+class InvalidPortNumber(Exception):
+    pass
 
+#-----------------------------------------------------------------------------
+# ZMQ Socket Channel classes
+#-----------------------------------------------------------------------------
 
 class ZmqSocketChannel(Thread):
-    """ The base class for the channels that use ZMQ sockets.
+    """The base class for the channels that use ZMQ sockets.
     """
     context = None
     session = None
     socket = None
     ioloop = None
     iostate = None
+    _address = None
+
+    def __init__(self, context, session, address):
+        """Create a channel
 
-    def __init__(self, context, session, address=None):
+        Parameters
+        ----------
+        context : zmq.Context
+            The ZMQ context to use.
+        session : session.Session
+            The session to use.
+        address : tuple
+            Standard (ip, port) tuple that the kernel is listening on.
+        """
         super(ZmqSocketChannel, self).__init__()
         self.daemon = True
 
         self.context = context
         self.session = session
-        self.address = address
+        if address[1] == 0:
+            raise InvalidPortNumber('The port number for a channel cannot be 0.')
+        self._address = address
 
     def stop(self):
-        """Stop the thread's activity. Returns when the thread terminates.
+        """Stop the channel's activity.
 
-        The thread will raise :class:`RuntimeError` if :method:`self.start`
-        is called again.
+        This calls :method:`Thread.join` and returns when the thread
+        terminates. :class:`RuntimeError` will be raised if 
+        :method:`self.start` is called again.
         """
         self.join()
 
-
-    def get_address(self):
-        """ Get the channel's address. By the default, a channel is on 
-            localhost with no port specified (a negative port number).
+    @property
+    def address(self):
+        """Get the channel's address as an (ip, port) tuple.
+        
+        By the default, the address is (localhost, 0), where 0 means a random
+        port.
         """
         return self._address
 
-    def set_adresss(self, address):
-        """ Set the channel's address. Should be a tuple of form:
-                (ip address [str], port [int]).
-            or None, in which case the address is reset to its default value.
-        """
-        # FIXME: Validate address.
-        if self.is_alive():  # This is Thread.is_alive
-            raise RuntimeError("Cannot set address on a running channel!")
-        else:
-            if address is None:
-                address = (LOCALHOST, 0)
-            self._address = address
-
-    address = property(get_address, set_adresss)
-
     def add_io_state(self, state):
         """Add IO state to the eventloop.
 
+        Parameters
+        ----------
+        state : zmq.POLLIN|zmq.POLLOUT|zmq.POLLERR
+            The IO state flag to set.
+
         This is thread safe as it uses the thread safe IOLoop.add_callback.
         """
         def add_io_state_callback():
@@ -88,6 +114,11 b' class ZmqSocketChannel(Thread):'
     def drop_io_state(self, state):
         """Drop IO state from the eventloop.
 
+        Parameters
+        ----------
+        state : zmq.POLLIN|zmq.POLLOUT|zmq.POLLERR
+            The IO state flag to set.
+
         This is thread safe as it uses the thread safe IOLoop.add_callback.
         """
         def drop_io_state_callback():
@@ -97,48 +128,30 b' class ZmqSocketChannel(Thread):'
         self.ioloop.add_callback(drop_io_state_callback)
 
 
-class SubSocketChannel(ZmqSocketChannel):
+class XReqSocketChannel(ZmqSocketChannel):
+    """The XREQ channel for issues request/replies to the kernel.
+    """
 
-    def __init__(self, context, session, address=None):
-        super(SubSocketChannel, self).__init__(context, session, address)
+    command_queue = None
+
+    def __init__(self, context, session, address):
+        self.command_queue = Queue()
+        super(XReqSocketChannel, self).__init__(context, session, address)
 
     def run(self):
-        self.socket = self.context.socket(zmq.SUB)
-        self.socket.setsockopt(zmq.SUBSCRIBE,'')
+        """The thread's main activity.  Call start() instead."""
+        self.socket = self.context.socket(zmq.XREQ)
         self.socket.setsockopt(zmq.IDENTITY, self.session.session)
         self.socket.connect('tcp://%s:%i' % self.address)
         self.ioloop = ioloop.IOLoop()
-        self.iostate = POLLIN|POLLERR
+        self.iostate = POLLERR|POLLIN
         self.ioloop.add_handler(self.socket, self._handle_events, 
                                 self.iostate)
         self.ioloop.start()
 
     def stop(self):
         self.ioloop.stop()
-        super(SubSocketChannel, self).stop()
-
-    def _handle_events(self, socket, events):
-        # Turn on and off POLLOUT depending on if we have made a request
-        if events & POLLERR:
-            self._handle_err()
-        if events & POLLIN:
-            self._handle_recv()
-
-    def _handle_err(self):
-        # We don't want to let this go silently, so eventually we should log.
-        raise zmq.ZMQError()
-
-    def _handle_recv(self):
-        # Get all of the messages we can
-        while True:
-            try:
-                msg = self.socket.recv_json(zmq.NOBLOCK)
-            except zmq.ZMQError:
-                # Check the errno?
-                # Will this tigger POLLERR?
-                break
-            else:
-                self.call_handlers(msg)
+        super(XReqSocketChannel, self).stop()
 
     def call_handlers(self, msg):
         """This method is called in the ioloop thread when a message arrives.
@@ -150,54 +163,65 b' class SubSocketChannel(ZmqSocketChannel):'
         """
         raise NotImplementedError('call_handlers must be defined in a subclass.')
 
-    def flush(self, timeout=1.0):
-        """Immediately processes all pending messages on the SUB channel.
-
-        This method is thread safe.
+    def execute(self, code):
+        """Execute code in the kernel.
 
         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 xrange(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):
-        """Called in this thread by the IOLoop to indicate that all events have
-        been processed.
-        """
-        self._flushed = True
+        code : str
+            A string of Python code.
 
+        Returns
+        -------
+        The msg_id of the message sent.
+        """
+        # Create class for content/msg creation. Related to, but possibly
+        # not in Session.
+        content = dict(code=code)
+        msg = self.session.msg('execute_request', content)
+        self._queue_request(msg)
+        return msg['header']['msg_id']
 
-class XReqSocketChannel(ZmqSocketChannel):
+    def complete(self, text, line, block=None):
+        """Tab complete text, line, block in the kernel's namespace.
 
-    command_queue = None
+        Parameters
+        ----------
+        text : str
+            The text to complete.
+        line : str
+            The full line of text that is the surrounding context for the 
+            text to complete.
+        block : str
+            The full block of code in which the completion is being requested.
+
+        Returns
+        -------
+        The msg_id of the message sent.
 
-    def __init__(self, context, session, address=None):
-        self.command_queue = Queue()
-        super(XReqSocketChannel, self).__init__(context, session, address)
+        """
+        content = dict(text=text, line=line)
+        msg = self.session.msg('complete_request', content)
+        self._queue_request(msg)
+        return msg['header']['msg_id']
 
-    def run(self):
-        self.socket = self.context.socket(zmq.XREQ)
-        self.socket.setsockopt(zmq.IDENTITY, self.session.session)
-        self.socket.connect('tcp://%s:%i' % self.address)
-        self.ioloop = ioloop.IOLoop()
-        self.iostate = POLLERR|POLLIN
-        self.ioloop.add_handler(self.socket, self._handle_events, 
-                                self.iostate)
-        self.ioloop.start()
+    def object_info(self, oname):
+        """Get metadata information about an object.
 
-    def stop(self):
-        self.ioloop.stop()
-        super(XReqSocketChannel, self).stop()
+        Parameters
+        ----------
+        oname : str
+            A string specifying the object name.
+        
+        Returns
+        -------
+        The msg_id of the message sent.
+        """
+        print oname
+        content = dict(oname=oname)
+        msg = self.session.msg('object_info_request', content)
+        self._queue_request(msg)
+        return msg['header']['msg_id']
 
     def _handle_events(self, socket, events):
         if events & POLLERR:
@@ -225,10 +249,34 b' class XReqSocketChannel(ZmqSocketChannel):'
         # We don't want to let this go silently, so eventually we should log.
         raise zmq.ZMQError()
 
-    def _queue_request(self, msg, callback):
+    def _queue_request(self, msg):
         self.command_queue.put(msg)
         self.add_io_state(POLLOUT)
 
+
+class SubSocketChannel(ZmqSocketChannel):
+    """The SUB channel which listens for messages that the kernel publishes.
+    """
+
+    def __init__(self, context, session, address):
+        super(SubSocketChannel, self).__init__(context, session, address)
+
+    def run(self):
+        """The thread's main activity.  Call start() instead."""
+        self.socket = self.context.socket(zmq.SUB)
+        self.socket.setsockopt(zmq.SUBSCRIBE,'')
+        self.socket.setsockopt(zmq.IDENTITY, self.session.session)
+        self.socket.connect('tcp://%s:%i' % self.address)
+        self.ioloop = ioloop.IOLoop()
+        self.iostate = POLLIN|POLLERR
+        self.ioloop.add_handler(self.socket, self._handle_events, 
+                                self.iostate)
+        self.ioloop.start()
+
+    def stop(self):
+        self.ioloop.stop()
+        super(SubSocketChannel, self).stop()
+
     def call_handlers(self, msg):
         """This method is called in the ioloop thread when a message arrives.
 
@@ -239,33 +287,66 b' class XReqSocketChannel(ZmqSocketChannel):'
         """
         raise NotImplementedError('call_handlers must be defined in a subclass.')
 
-    def execute(self, code, callback=None):
-        # Create class for content/msg creation. Related to, but possibly
-        # not in Session.
-        content = dict(code=code)
-        msg = self.session.msg('execute_request', content)
-        self._queue_request(msg, callback)
-        return msg['header']['msg_id']
+    def flush(self, timeout=1.0):
+        """Immediately processes all pending messages on the SUB channel.
 
-    def complete(self, text, line, block=None, callback=None):
-        content = dict(text=text, line=line)
-        msg = self.session.msg('complete_request', content)
-        self._queue_request(msg, callback)
-        return msg['header']['msg_id']
+        This method is thread safe.
 
-    def object_info(self, oname, callback=None):
-        content = dict(oname=oname)
-        msg = self.session.msg('object_info_request', content)
-        self._queue_request(msg, callback)
-        return msg['header']['msg_id']
+        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 xrange(2):
+            self._flushed = False
+            self.ioloop.add_callback(self._flush)
+            while not self._flushed and time.time() < stop_time:
+                time.sleep(0.01)
+
+    def _handle_events(self, socket, events):
+        # Turn on and off POLLOUT depending on if we have made a request
+        if events & POLLERR:
+            self._handle_err()
+        if events & POLLIN:
+            self._handle_recv()
+
+    def _handle_err(self):
+        # We don't want to let this go silently, so eventually we should log.
+        raise zmq.ZMQError()
+
+    def _handle_recv(self):
+        # Get all of the messages we can
+        while True:
+            try:
+                msg = self.socket.recv_json(zmq.NOBLOCK)
+            except zmq.ZMQError:
+                # Check the errno?
+                # Will this tigger POLLERR?
+                break
+            else:
+                self.call_handlers(msg)
+
+    def _flush(self):
+        """Callback for :method:`self.flush`."""
+        self._flushed = True
 
 
 class RepSocketChannel(ZmqSocketChannel):
+    """A reply channel to handle raw_input requests that the kernel makes."""
 
     def on_raw_input(self):
         pass
 
 
+#-----------------------------------------------------------------------------
+# Main kernel manager class
+#-----------------------------------------------------------------------------
+
+
 class KernelManager(HasTraits):
     """ Manages a kernel for a frontend.
 
@@ -277,59 +358,66 b' class KernelManager(HasTraits):'
     The REP channel is for the kernel to request stdin (raw_input) from the
     frontend.
     """
-
-    # Whether the kernel manager is currently listening on its channels.
-    is_listening = Bool(False)
-
     # The PyZMQ Context to use for communication with the kernel.
-    context = Instance(zmq.Context, ())
+    context = Instance(zmq.Context)
 
     # The Session to use for communication with the kernel.
-    session = Instance(Session, ())
+    session = Instance(Session)
 
     # The classes to use for the various channels.
-    sub_channel_class = Type(SubSocketChannel)
     xreq_channel_class = Type(XReqSocketChannel)
+    sub_channel_class = Type(SubSocketChannel)
     rep_channel_class = Type(RepSocketChannel)
     
     # Protected traits.
     _kernel = Instance(Popen)
-    _sub_channel = Any
+    _xreq_address = Any
+    _sub_address = Any
+    _rep_address = Any
     _xreq_channel = Any
+    _sub_channel = Any
     _rep_channel = Any
 
+    def __init__(self, xreq_address=None, sub_address=None, rep_address=None,
+                 context=None, session=None):
+        self._xreq_address = (LOCALHOST, 0) if xreq_address is None else xreq_address
+        self._sub_address = (LOCALHOST, 0) if sub_address is None else sub_address
+        self._rep_address = (LOCALHOST, 0) if rep_address is None else rep_address
+        self.context = zmq.Context() if context is None else context
+        self.session = Session() if session is None else session
+
     #--------------------------------------------------------------------------
     # Channel management methods:
     #--------------------------------------------------------------------------
 
-    def start_listening(self):
-        """Starts listening on the specified ports. If already listening, raises
-        a RuntimeError.
+    def start_channels(self):
+        """Starts the channels for this kernel.
+
+        This will create the channels if they do not exist and then start
+        them. If port numbers of 0 are being used (random ports) then you
+        must first call :method:`start_kernel`. If the channels have been
+        stopped and you call this, :class:`RuntimeError` will be raised.
         """
-        if self.is_listening:
-            raise RuntimeError("Cannot start listening. Already listening!")
-        else:
-            self.is_listening = True
-            self.sub_channel.start()
-            self.xreq_channel.start()
-            self.rep_channel.start()
+        self.xreq_channel.start()
+        self.sub_channel.start()
+        self.rep_channel.start()
 
-    @property
-    def is_alive(self):
-        """ Returns whether the kernel is alive. """
-        if self.is_listening:
-            # TODO: check if alive.
-            return True
-        else:
-            return False
+    def stop_channels(self):
+        """Stops the channels for this kernel.
+
+        This stops the channels by joining their threads. If the channels
+        were not started, :class:`RuntimeError` will be raised.
+        """
+        self.xreq_channel.stop()
+        self.sub_channel.stop()
+        self.rep_channel.stop()
 
-    def stop_listening(self):
-        """Stops listening. If not listening, does nothing. """
-        if self.is_listening:
-            self.is_listening = False
-            self.sub_channel.stop()
-            self.xreq_channel.stop()
-            self.rep_channel.stop()
+    @property
+    def channels_running(self):
+        """Are all of the channels created and running?"""
+        return self.xreq_channel.is_alive() \
+            and self.sub_channel.is_alive() \
+            and self.rep_channel.is_alive()
 
     #--------------------------------------------------------------------------
     # Kernel process management methods:
@@ -338,9 +426,8 b' class KernelManager(HasTraits):'
     def start_kernel(self):
         """Starts a kernel process and configures the manager to use it.
 
-        If ports have been specified via the address attributes, they are used.
-        Otherwise, open ports are chosen by the OS and the channel port
-        attributes are configured as appropriate.
+        If random ports (port=0) are being used, this method must be called
+        before the channels are created.
         """
         xreq, sub = self.xreq_address, self.sub_address
         if xreq[0] != LOCALHOST or sub[0] != LOCALHOST:
@@ -349,24 +436,14 b' class KernelManager(HasTraits):'
                                "configured properly.")
 
         kernel, xrep, pub = launch_kernel(xrep_port=xreq[1], pub_port=sub[1])
-        self.set_kernel(kernel)
-        self.xreq_address = (LOCALHOST, xrep)
-        self.sub_address = (LOCALHOST, pub)
-
-    def set_kernel(self, kernel):
-        """Sets the kernel manager's kernel to an existing kernel process.
-
-        It is *not* necessary to a set a kernel to communicate with it via the
-        channels, and those objects must be configured separately. It
-        *is* necessary to set a kernel if you want to use the manager (or
-        frontends that use the manager) to signal and/or kill the kernel.
-
-        Parameters:
-        -----------
-        kernel : Popen
-            An existing kernel process.
-        """
         self._kernel = kernel
+        print xrep, pub
+        self._xreq_address = (LOCALHOST, xrep)
+        self._sub_address = (LOCALHOST, pub)
+        # The rep channel is not fully working yet, but its base class makes
+        # sure the port is not 0. We set to -1 for now until the rep channel
+        # is fully working.
+        self._rep_address = (LOCALHOST, -1)
 
     @property
     def has_kernel(self):
@@ -379,7 +456,7 b' class KernelManager(HasTraits):'
 
     def kill_kernel(self):
         """ Kill the running kernel. """
-        if self._kernel:
+        if self._kernel is not None:
             self._kernel.kill()
             self._kernel = None
         else:
@@ -387,67 +464,65 b' class KernelManager(HasTraits):'
 
     def signal_kernel(self, signum):
         """ Sends a signal to the kernel. """
-        if self._kernel:
+        if self._kernel is not None:
             self._kernel.send_signal(signum)
         else:
             raise RuntimeError("Cannot signal kernel. No kernel is running!")
 
+    @property
+    def is_alive(self):
+        """Is the kernel process still running?"""
+        if self._kernel is not None:
+            if self._kernel.poll() is None:
+                return True
+            else:
+                return False
+        else:
+            # We didn't start the kernel with this KernelManager so we don't
+            # know if it is running. We should use a heartbeat for this case.
+            return True
+
     #--------------------------------------------------------------------------
     # Channels used for communication with the kernel:
     #--------------------------------------------------------------------------
 
     @property
-    def sub_channel(self):
-        """Get the SUB socket channel object."""
-        if self._sub_channel is None:
-            self._sub_channel = self.sub_channel_class(self.context,
-                                                       self.session)
-        return self._sub_channel
-
-    @property
     def xreq_channel(self):
         """Get the REQ socket channel object to make requests of the kernel."""
         if self._xreq_channel is None:
             self._xreq_channel = self.xreq_channel_class(self.context, 
-                                                         self.session)
+                                                         self.session,
+                                                         self.xreq_address)
         return self._xreq_channel
 
     @property
+    def sub_channel(self):
+        """Get the SUB socket channel object."""
+        if self._sub_channel is None:
+            self._sub_channel = self.sub_channel_class(self.context,
+                                                       self.session,
+                                                       self.sub_address)
+        return self._sub_channel
+
+    @property
     def rep_channel(self):
         """Get the REP socket channel object to handle stdin (raw_input)."""
         if self._rep_channel is None:
             self._rep_channel = self.rep_channel_class(self.context, 
-                                                       self.session)
+                                                       self.session,
+                                                       self.rep_address)
         return self._rep_channel
 
-    #--------------------------------------------------------------------------
-    # Delegates for the Channel address attributes:
-    #--------------------------------------------------------------------------
-
-    def get_sub_address(self):
-        return self.sub_channel.address
-
-    def set_sub_address(self, address):
-        self.sub_channel.address = address
-
-    sub_address = property(get_sub_address, set_sub_address,
-                           doc="The address used by SUB socket channel.")
-
-    def get_xreq_address(self):
-        return self.xreq_channel.address
-
-    def set_xreq_address(self, address):
-        self.xreq_channel.address = address
+    @property
+    def xreq_address(self):
+        return self._xreq_address
 
-    xreq_address = property(get_xreq_address, set_xreq_address,
-                            doc="The address used by XREQ socket channel.")
-    
-    def get_rep_address(self):
-        return self.rep_channel.address
+    @property
+    def sub_address(self):
+        return self._sub_address
 
-    def set_rep_address(self, address):
-        self.rep_channel.address = address
+    @property
+    def rep_address(self):
+        return self._rep_address
 
-    rep_address = property(get_rep_address, set_rep_address,
-                           doc="The address used by REP socket channel.")