Merge pull request #2775 from ellisonbg/kernelid

General cleanup of kernel manager code.

This does some general cleanup of MultiKernelManager to better reflect how it is actually being used.

Sometimes there is a need to create kernel_id's elsewhere in code.  This minor change allows a kernel_id to be created outside of the MultiKernelManager and passed in as a kwarg.  An exception is raised if the id is already used.

Tasks:

- [x] Fix cleanup of ipc files.
- [x] Allow kernel_id to be passed to MultiKernelManager.start_kernel.
- [x] Add ipc support to MultiKernelManager.
- [x] Add more tests for MultiKernelManager.
- [x] Rename sub channel to iopub channel everywhere.
- [x] Use consistent naming for all channel classes in zmq, inprocess and qt.
- [x] Move BlockingChannelMixin to zmq.blockingkernelmanager.
- [x] Create ABC for KernelManager.
- [x] Make the InProcessKernelManager a Configurable.
- [x] Cleanup docstrings in ABCs.
- [x] Add tests for KernelManager.
- [x] Check over MultiKernelManager.
- [x] Make KernelManager,kill_kernel private in ABC and implementations.
- [x] Find bug that is causing the kernel manager tests to hang unless the shell channel linger is set to 0.
- [x] Decide about critical logging in ipkernel.
- [x] Debug lack of stderr redirect in tests.

IPython/frontend/html/notebook/tests/test_kernelmanager.py

@@ -0,0 +1,70 b''
+"""Tests for the notebook kernel and session manager."""
+
+from subprocess import PIPE
+import time
+from unittest import TestCase
+
+from IPython.config.loader import Config
+from IPython.frontend.html.notebook.kernelmanager import MultiKernelManager
+from IPython.zmq.kernelmanager import KernelManager
+
+class TestKernelManager(TestCase):
+
+    def _get_tcp_km(self):
+        return MultiKernelManager()
+
+    def _get_ipc_km(self):
+        c = Config()
+        c.KernelManager.transport = 'ipc'
+        c.KernelManager.ip = 'test'
+        km = MultiKernelManager(config=c)
+        return km
+
+    def _run_lifecycle(self, km):
+        kid = km.start_kernel(stdout=PIPE, stderr=PIPE)
+        self.assertTrue(kid in km)
+        self.assertTrue(kid in km.list_kernel_ids())
+        self.assertEqual(len(km),1)
+        km.restart_kernel(kid)
+        self.assertTrue(kid in km.list_kernel_ids())
+        # We need a delay here to give the restarting kernel a chance to
+        # restart. Otherwise, the interrupt will kill it, causing the test
+        # suite to hang. The reason it *hangs* is that the shutdown
+        # message for the restart sometimes hasn't been sent to the kernel.
+        # Because linger is oo on the shell channel, the context can't
+        # close until the message is sent to the kernel, which is not dead.
+        time.sleep(1.0)
+        km.interrupt_kernel(kid)
+        k = km.get_kernel(kid)
+        self.assertTrue(isinstance(k, KernelManager))
+        km.shutdown_kernel(kid)
+        self.assertTrue(not kid in km)
+
+    def _run_cinfo(self, km, transport, ip):
+        kid = km.start_kernel(stdout=PIPE, stderr=PIPE)
+        k = km.get_kernel(kid)
+        cinfo = km.get_connection_info(kid)
+        self.assertEqual(transport, cinfo['transport'])
+        self.assertEqual(ip, cinfo['ip'])
+        self.assertTrue('stdin_port' in cinfo)
+        self.assertTrue('iopub_port' in cinfo)
+        self.assertTrue('shell_port' in cinfo)
+        self.assertTrue('hb_port' in cinfo)
+        km.shutdown_kernel(kid)
+
+    def test_tcp_lifecycle(self):
+        km = self._get_tcp_km()
+        self._run_lifecycle(km)
+    
+    def test_tcp_cinfo(self):
+        km = self._get_tcp_km()
+        self._run_cinfo(km, 'tcp', '127.0.0.1')
+
+    def test_ipc_lifecycle(self):
+        km = self._get_ipc_km()
+        self._run_lifecycle(km)
+    
+    def test_ipc_cinfo(self):
+        km = self._get_ipc_km()
+        self._run_cinfo(km, 'ipc', 'test')
+

IPython/zmq/kernelmanagerabc.py

@@ -0,0 +1,226 b''
+"""Abstract base classes for kernel manager and channels."""
+
+#-----------------------------------------------------------------------------
+#  Copyright (C) 2013  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.
+import abc
+
+#-----------------------------------------------------------------------------
+# Channels
+#-----------------------------------------------------------------------------
+
+
+class ChannelABC(object):
+    """A base class for all channel ABCs."""
+
+    __metaclass__ = abc.ABCMeta
+
+    @abc.abstractmethod
+    def start(self):
+        pass
+
+    @abc.abstractmethod
+    def stop(self):
+        pass
+
+    @abc.abstractmethod
+    def is_alive(self):
+        pass
+
+
+class ShellChannelABC(ChannelABC):
+    """ShellChannel ABC.
+
+    The docstrings for this class can be found in the base implementation:
+
+    `IPython.zmq.kernelmanager.ShellChannel`
+    """
+
+    @abc.abstractproperty
+    def allow_stdin(self):
+        pass
+
+    @abc.abstractmethod
+    def execute(self, code, silent=False, store_history=True,
+                user_variables=None, user_expressions=None, allow_stdin=None):
+        pass
+
+    @abc.abstractmethod
+    def complete(self, text, line, cursor_pos, block=None):
+        pass
+
+    @abc.abstractmethod
+    def object_info(self, oname, detail_level=0):
+        pass
+
+    @abc.abstractmethod
+    def history(self, raw=True, output=False, hist_access_type='range', **kwargs):
+        pass
+
+    @abc.abstractmethod
+    def kernel_info(self):
+        pass
+
+    @abc.abstractmethod
+    def shutdown(self, restart=False):
+        pass
+
+
+class IOPubChannelABC(ChannelABC):
+    """IOPubChannel ABC.
+
+    The docstrings for this class can be found in the base implementation:
+
+    `IPython.zmq.kernelmanager.IOPubChannel`
+    """
+
+    @abc.abstractmethod
+    def flush(self, timeout=1.0):
+        pass
+
+
+class StdInChannelABC(ChannelABC):
+    """StdInChannel ABC.
+
+    The docstrings for this class can be found in the base implementation:
+
+    `IPython.zmq.kernelmanager.StdInChannel`
+    """
+
+    @abc.abstractmethod
+    def input(self, string):
+        pass
+
+
+class HBChannelABC(ChannelABC):
+    """HBChannel ABC.
+
+    The docstrings for this class can be found in the base implementation:
+
+    `IPython.zmq.kernelmanager.HBChannel`
+    """
+
+    @abc.abstractproperty
+    def time_to_dead(self):
+        pass
+
+    @abc.abstractmethod
+    def pause(self):
+        pass
+
+    @abc.abstractmethod
+    def unpause(self):
+        pass
+
+    @abc.abstractmethod
+    def is_beating(self):
+        pass
+
+
+#-----------------------------------------------------------------------------
+# Main kernel manager class
+#-----------------------------------------------------------------------------
+
+class KernelManagerABC(object):
+    """KernelManager ABC.
+
+    The docstrings for this class can be found in the base implementation:
+
+    `IPython.zmq.kernelmanager.KernelManager`
+    """
+
+    __metaclass__ = abc.ABCMeta
+
+    @abc.abstractproperty
+    def kernel(self):
+        pass
+
+    @abc.abstractproperty
+    def shell_channel_class(self):
+        pass
+
+    @abc.abstractproperty
+    def iopub_channel_class(self):
+        pass
+
+    @abc.abstractproperty
+    def hb_channel_class(self):
+        pass
+
+    @abc.abstractproperty
+    def stdin_channel_class(self):
+        pass
+
+    #--------------------------------------------------------------------------
+    # Channel management methods
+    #--------------------------------------------------------------------------
+
+    @abc.abstractmethod
+    def start_channels(self, shell=True, iopub=True, stdin=True, hb=True):
+        pass
+
+    @abc.abstractmethod
+    def stop_channels(self):
+        pass
+
+    @abc.abstractproperty
+    def channels_running(self):
+        pass
+
+    @abc.abstractproperty
+    def shell_channel(self):
+        pass
+
+    @abc.abstractproperty
+    def iopub_channel(self):
+        pass
+
+    @abc.abstractproperty
+    def stdin_channel(self):
+        pass
+
+    @abc.abstractproperty
+    def hb_channel(self):
+        pass
+
+    #--------------------------------------------------------------------------
+    # Kernel management
+    #--------------------------------------------------------------------------
+
+    @abc.abstractmethod
+    def start_kernel(self, **kw):
+        pass
+
+    @abc.abstractmethod
+    def shutdown_kernel(self, now=False, restart=False):
+        pass
+
+    @abc.abstractmethod
+    def restart_kernel(self, now=False, **kw):
+        pass
+
+    @abc.abstractproperty
+    def has_kernel(self):
+        pass
+
+    @abc.abstractmethod
+    def interrupt_kernel(self):
+        pass
+
+    @abc.abstractmethod
+    def signal_kernel(self, signum):
+        pass
+
+    @abc.abstractproperty
+    def is_alive(self):
+        pass
+

IPython/zmq/tests/test_kernelmanager.py

@@ -0,0 +1,45 b''
+"""Tests for the notebook kernel and session manager"""
+
+from subprocess import PIPE
+import time
+from unittest import TestCase
+
+from IPython.config.loader import Config
+from IPython.zmq.kernelmanager import KernelManager
+
+class TestKernelManager(TestCase):
+
+    def _get_tcp_km(self):
+        return KernelManager()
+
+    def _get_ipc_km(self):
+        c = Config()
+        c.KernelManager.transport = 'ipc'
+        c.KernelManager.ip = 'test'
+        km = KernelManager(config=c)
+        return km
+
+    def _run_lifecycle(self, km):
+        km.start_kernel(stdout=PIPE, stderr=PIPE)
+        km.start_channels(shell=True, iopub=False, stdin=False, hb=False)
+        km.restart_kernel()
+        # We need a delay here to give the restarting kernel a chance to
+        # restart. Otherwise, the interrupt will kill it, causing the test
+        # suite to hang. The reason it *hangs* is that the shutdown
+        # message for the restart sometimes hasn't been sent to the kernel.
+        # Because linger is oo on the shell channel, the context can't
+        # close until the message is sent to the kernel, which is not dead.
+        time.sleep(1.0)
+        km.interrupt_kernel()
+        self.assertTrue(isinstance(km, KernelManager))
+        km.shutdown_kernel()
+        km.shell_channel.stop()
+
+    def test_tcp_lifecycle(self):
+        km = self._get_tcp_km()
+        self._run_lifecycle(km)
+
+    def testipc_lifecycle(self):
+        km = self._get_ipc_km()
+        self._run_lifecycle(km)
+

IPython/frontend/html/notebook/handlers.py

@@ -330,7 +330,7 b' class MainKernelHandler(AuthenticatedHandler):'
     @web.authenticated
     def get(self):
         km = self.application.kernel_manager
-        self.finish(jsonapi.dumps(km.kernel_ids))
+        self.finish(jsonapi.dumps(km.list_kernel_ids()))
 
     @web.authenticated
     def post(self):
@@ -364,9 +364,9 b' class KernelActionHandler(AuthenticatedHandler):'
             km.interrupt_kernel(kernel_id)
             self.set_status(204)
         if action == 'restart':
-            new_kernel_id = km.restart_kernel(kernel_id)
-            data = {'ws_url':self.ws_url,'kernel_id':new_kernel_id}
-            self.set_header('Location', '/'+new_kernel_id)
+            km.restart_kernel(kernel_id)
+            data = {'ws_url':self.ws_url, 'kernel_id':kernel_id}
+            self.set_header('Location', '/'+kernel_id)
             self.write(jsonapi.dumps(data))
         self.finish()
 
@@ -416,7 +416,7 b' class AuthenticatedZMQStreamHandler(ZMQStreamHandler):'
     def open(self, kernel_id):
         self.kernel_id = kernel_id.decode('ascii')
         try:
-            cfg = self.application.ipython_app.config
+            cfg = self.application.config
         except AttributeError:
             # protect from the case where this is run from something other than
             # the notebook app:
@@ -541,9 +541,13 b' class IOPubHandler(AuthenticatedZMQStreamHandler):'
             if not self.hb_stream.closed():
                 self.hb_stream.on_recv(None)
 
-    def kernel_died(self):
+    def _delete_kernel_data(self):
+        """Remove the kernel data and notebook mapping."""
         self.application.kernel_manager.delete_mapping_for_kernel(self.kernel_id)
-        self.application.log.error("Kernel %s failed to respond to heartbeat", self.kernel_id)
+
+    def kernel_died(self):
+        self._delete_kernel_data()
+        self.application.log.error("Kernel died: %s" % self.kernel_id)
         self.write_message(
             {'header': {'msg_type': 'status'},
              'parent_header': {},

IPython/frontend/html/notebook/kernelmanager.py

@@ -17,8 +17,6 b' Authors:'
 #-----------------------------------------------------------------------------
 
 import os
-import signal
-import sys
 import uuid
 
 import zmq
@@ -63,56 +61,65 b' class MultiKernelManager(LoggingConfigurable):'
 
     _kernels = Dict()
 
-    @property
-    def kernel_ids(self):
+    def list_kernel_ids(self):
         """Return a list of the kernel ids of the active kernels."""
-        return self._kernels.keys()
+        # Create a copy so we can iterate over kernels in operations
+        # that delete keys.
+        return list(self._kernels.keys())
 
     def __len__(self):
         """Return the number of running kernels."""
-        return len(self.kernel_ids)
+        return len(self.list_kernel_ids())
 
     def __contains__(self, kernel_id):
-        if kernel_id in self.kernel_ids:
-            return True
-        else:
-            return False
+        return kernel_id in self._kernels
 
     def start_kernel(self, **kwargs):
-        """Start a new kernel."""
-        kernel_id = unicode(uuid.uuid4())
-        # use base KernelManager for each Kernel
+        """Start a new kernel.
+
+        The caller can pick a kernel_id by passing one in as a keyword arg,
+        otherwise one will be picked using a uuid.
+
+        To silence the kernel's stdout/stderr, call this using::
+
+            km.start_kernel(stdout=PIPE, stderr=PIPE)
+
+        """
+        kernel_id = kwargs.pop('kernel_id', unicode(uuid.uuid4()))
+        if kernel_id in self:
+            raise DuplicateKernelError('Kernel already exists: %s' % kernel_id)
+        # kernel_manager_factory is the constructor for the KernelManager
+        # subclass we are using. It can be configured as any Configurable,
+        # including things like its transport and ip.
         km = self.kernel_manager_factory(connection_file=os.path.join(
                     self.connection_dir, "kernel-%s.json" % kernel_id),
                     config=self.config,
         )
         km.start_kernel(**kwargs)
         # start just the shell channel, needed for graceful restart
-        km.start_channels(shell=True, sub=False, stdin=False, hb=False)
+        km.start_channels(shell=True, iopub=False, stdin=False, hb=False)
         self._kernels[kernel_id] = km
         return kernel_id
 
-    def shutdown_kernel(self, kernel_id):
+    def shutdown_kernel(self, kernel_id, now=False):
         """Shutdown a kernel by its kernel uuid.
 
         Parameters
         ==========
         kernel_id : uuid
             The id of the kernel to shutdown.
+        now : bool
+            Should the kernel be shutdown forcibly using a signal.
         """
-        self.get_kernel(kernel_id).shutdown_kernel()
+        k = self.get_kernel(kernel_id)
+        k.shutdown_kernel(now=now)
+        k.shell_channel.stop()
         del self._kernels[kernel_id]
 
-    def kill_kernel(self, kernel_id):
-        """Kill a kernel by its kernel uuid.
-
-        Parameters
-        ==========
-        kernel_id : uuid
-            The id of the kernel to kill.
-        """
-        self.get_kernel(kernel_id).kill_kernel()
-        del self._kernels[kernel_id]
+    def shutdown_all(self, now=False):
+        """Shutdown all kernels."""
+        for kid in self.list_kernel_ids():
+            self.shutdown_kernel(kid, now=now)
 
     def interrupt_kernel(self, kernel_id):
         """Interrupt (SIGINT) the kernel by its uuid.
@@ -125,7 +132,7 b' class MultiKernelManager(LoggingConfigurable):'
         return self.get_kernel(kernel_id).interrupt_kernel()
 
     def signal_kernel(self, kernel_id, signum):
-        """ Sends a signal to the kernel by its uuid.
+        """Sends a signal to the kernel by its uuid.
 
         Note that since only SIGTERM is supported on Windows, this function
         is only useful on Unix systems.
@@ -161,8 +168,8 b' class MultiKernelManager(LoggingConfigurable):'
         else:
             raise KeyError("Kernel with id not found: %s" % kernel_id)
 
-    def get_kernel_ports(self, kernel_id):
-        """Return a dictionary of ports for a kernel.
+    def get_connection_info(self, kernel_id):
+        """Return a dictionary of connection data for a kernel.
 
         Parameters
         ==========
@@ -171,21 +178,39 b' class MultiKernelManager(LoggingConfigurable):'
 
         Returns
         =======
-        port_dict : dict
-            A dict of key, value pairs where the keys are the names
-            (stdin_port,iopub_port,shell_port) and the values are the
-            integer port numbers for those channels.
+        connection_dict : dict
+            A dict of the information needed to connect to a kernel.
+            This includes the ip address and the integer port
+            numbers of the different channels (stdin_port, iopub_port,
+            shell_port, hb_port).
         """
-        # this will raise a KeyError if not found:
         km = self.get_kernel(kernel_id)
-        return dict(shell_port=km.shell_port,
+        return dict(transport=km.transport,
+                    ip=km.ip,
+                    shell_port=km.shell_port,
                     iopub_port=km.iopub_port,
                     stdin_port=km.stdin_port,
                     hb_port=km.hb_port,
-        )
+        )  
+
+    def _make_url(self, transport, ip, port):
+        """Make a ZeroMQ URL for a given transport, ip and port."""
+        if transport == 'tcp':
+            return "tcp://%s:%i" % (ip, port)
+        else:
+            return "%s://%s-%s" % (transport, ip, port)
+
+    def _create_connected_stream(self, kernel_id, socket_type):
+        """Create a connected ZMQStream for a kernel."""
+        cinfo = self.get_connection_info(kernel_id)
+        url = self._make_url(cinfo['transport'], cinfo['ip'], cinfo['port'])
+        sock = self.context.socket(socket_type)
+        self.log.info("Connecting to: %s" % url)
+        sock.connect(url)
+        return ZMQStream(sock)
 
-    def get_kernel_ip(self, kernel_id):
-        """Return ip address for a kernel.
+    def create_iopub_stream(self, kernel_id):
+        """Return a ZMQStream object connected to the iopub channel.
 
         Parameters
         ==========
@@ -194,35 +219,40 b' class MultiKernelManager(LoggingConfigurable):'
 
         Returns
         =======
-        ip : str
-            The ip address of the kernel.
+        stream : ZMQStream
         """
-        return self.get_kernel(kernel_id).ip
-
-    def create_connected_stream(self, ip, port, socket_type):
-        sock = self.context.socket(socket_type)
-        addr = "tcp://%s:%i" % (ip, port)
-        self.log.info("Connecting to: %s" % addr)
-        sock.connect(addr)
-        return ZMQStream(sock)
-
-    def create_iopub_stream(self, kernel_id):
-        ip = self.get_kernel_ip(kernel_id)
-        ports = self.get_kernel_ports(kernel_id)
-        iopub_stream = self.create_connected_stream(ip, ports['iopub_port'], zmq.SUB)
+        iopub_stream = self._create_connected_stream(kernel_id, zmq.SUB)
         iopub_stream.socket.setsockopt(zmq.SUBSCRIBE, b'')
         return iopub_stream
 
     def create_shell_stream(self, kernel_id):
-        ip = self.get_kernel_ip(kernel_id)
-        ports = self.get_kernel_ports(kernel_id)
-        shell_stream = self.create_connected_stream(ip, ports['shell_port'], zmq.DEALER)
+        """Return a ZMQStream object connected to the shell channel.
+
+        Parameters
+        ==========
+        kernel_id : uuid
+            The id of the kernel.
+
+        Returns
+        =======
+        stream : ZMQStream
+        """
+        shell_stream = self._create_connected_stream(kernel_id, zmq.DEALER)
         return shell_stream
 
     def create_hb_stream(self, kernel_id):
-        ip = self.get_kernel_ip(kernel_id)
-        ports = self.get_kernel_ports(kernel_id)
-        hb_stream = self.create_connected_stream(ip, ports['hb_port'], zmq.REQ)
+        """Return a ZMQStream object connected to the hb channel.
+
+        Parameters
+        ==========
+        kernel_id : uuid
+            The id of the kernel.
+
+        Returns
+        =======
+        stream : ZMQStream
+        """
+        hb_stream = self._create_connected_stream(kernel_id, zmq.REQ)
         return hb_stream
 
 
@@ -289,20 +319,15 b' class MappingKernelManager(MultiKernelManager):'
             self.log.info("Using existing kernel: %s" % kernel_id)
         return kernel_id
 
-    def shutdown_kernel(self, kernel_id):
+    def shutdown_kernel(self, kernel_id, now=False):
         """Shutdown a kernel and remove its notebook association."""
         self._check_kernel_id(kernel_id)
-        super(MappingKernelManager, self).shutdown_kernel(kernel_id)
+        super(MappingKernelManager, self).shutdown_kernel(
+            kernel_id, now=now
+        )
         self.delete_mapping_for_kernel(kernel_id)
         self.log.info("Kernel shutdown: %s" % kernel_id)
 
-    def kill_kernel(self, kernel_id):
-        """Kill a kernel and remove its notebook association."""
-        self._check_kernel_id(kernel_id)
-        super(MappingKernelManager, self).kill_kernel(kernel_id)
-        self.delete_mapping_for_kernel(kernel_id)
-        self.log.info("Kernel killed: %s" % kernel_id)
-
     def interrupt_kernel(self, kernel_id):
         """Interrupt a kernel."""
         self._check_kernel_id(kernel_id)
@@ -314,7 +339,6 b' class MappingKernelManager(MultiKernelManager):'
         self._check_kernel_id(kernel_id)
         super(MappingKernelManager, self).restart_kernel(kernel_id)
         self.log.info("Kernel restarted: %s" % kernel_id)
-        return kernel_id
 
     def create_iopub_stream(self, kernel_id):
         """Create a new iopub stream."""

IPython/frontend/html/notebook/notebookapp.py

@@ -187,6 +187,7 b' class NotebookWebApplication(web.Application):'
         self.cluster_manager = cluster_manager
         self.ipython_app = ipython_app
         self.read_only = self.ipython_app.read_only
+        self.config = self.ipython_app.config
         self.log = log
         self.jinja2_env = Environment(loader=FileSystemLoader(os.path.join(os.path.dirname(__file__), "templates")))
 
@@ -591,16 +592,13 b' class NotebookApp(BaseIPythonApplication):'
         self.init_signal()
 
     def cleanup_kernels(self):
-        """shutdown all kernels
+        """Shutdown all kernels.
         
         The kernels will shutdown themselves when this process no longer exists,
         but explicit shutdown allows the KernelManagers to cleanup the connection files.
         """
         self.log.info('Shutting down kernels')
-        km = self.kernel_manager
-        # copy list, since shutdown_kernel deletes keys
-        for kid in list(km.kernel_ids):
-            km.shutdown_kernel(kid)
+        self.kernel_manager.shutdown_all()
 
     def start(self):
         ip = self.ip if self.ip else '[all ip addresses on your system]'

IPython/frontend/qt/base_frontend_mixin.py

@@ -30,7 +30,7 b' class BaseFrontendMixin(object):'
             old_manager.stopped_channels.disconnect(self._stopped_channels)
 
             # Disconnect the old kernel manager's channels.
-            old_manager.sub_channel.message_received.disconnect(self._dispatch)
+            old_manager.iopub_channel.message_received.disconnect(self._dispatch)
             old_manager.shell_channel.message_received.disconnect(self._dispatch)
             old_manager.stdin_channel.message_received.disconnect(self._dispatch)
             old_manager.hb_channel.kernel_died.disconnect(
@@ -51,7 +51,7 b' class BaseFrontendMixin(object):'
         kernel_manager.stopped_channels.connect(self._stopped_channels)
 
         # Connect the new kernel manager's channels.
-        kernel_manager.sub_channel.message_received.connect(self._dispatch)
+        kernel_manager.iopub_channel.message_received.connect(self._dispatch)
         kernel_manager.shell_channel.message_received.connect(self._dispatch)
         kernel_manager.stdin_channel.message_received.connect(self._dispatch)
         kernel_manager.hb_channel.kernel_died.connect(self._handle_kernel_died)

IPython/frontend/qt/base_kernelmanager.py

@@ -69,7 +69,7 b' class QtShellChannelMixin(ChannelQObject):'
     _handlers_called = False
 
     #---------------------------------------------------------------------------
-    # 'ShellSocketChannel' interface
+    # 'ShellChannel' interface
     #---------------------------------------------------------------------------
 
     def call_handlers(self, msg):
@@ -98,7 +98,7 b' class QtShellChannelMixin(ChannelQObject):'
         self._handlers_called = False
 
 
-class QtSubChannelMixin(ChannelQObject):
+class QtIOPubChannelMixin(ChannelQObject):
 
     # Emitted when any message is received.
     message_received = QtCore.Signal(object)
@@ -126,7 +126,7 b' class QtSubChannelMixin(ChannelQObject):'
     shutdown_reply_received = QtCore.Signal(object)
 
     #---------------------------------------------------------------------------
-    # 'SubSocketChannel' interface
+    # 'IOPubChannel' interface
     #---------------------------------------------------------------------------
 
     def call_handlers(self, msg):
@@ -145,7 +145,7 b' class QtSubChannelMixin(ChannelQObject):'
     def flush(self):
         """ Reimplemented to ensure that signals are dispatched immediately.
         """
-        super(QtSubChannelMixin, self).flush()
+        super(QtIOPubChannelMixin, self).flush()
         QtCore.QCoreApplication.instance().processEvents()
 
 
@@ -158,7 +158,7 b' class QtStdInChannelMixin(ChannelQObject):'
     input_requested = QtCore.Signal(object)
 
     #---------------------------------------------------------------------------
-    # 'StdInSocketChannel' interface
+    # 'StdInChannel' interface
     #---------------------------------------------------------------------------
 
     def call_handlers(self, msg):
@@ -179,7 +179,7 b' class QtHBChannelMixin(ChannelQObject):'
     kernel_died = QtCore.Signal(object)
 
     #---------------------------------------------------------------------------
-    # 'HBSocketChannel' interface
+    # 'HBChannel' interface
     #---------------------------------------------------------------------------
 
     def call_handlers(self, since_last_heartbeat):
@@ -205,7 +205,7 b' class QtKernelManagerMixin(HasTraits, SuperQObject):'
     stopped_channels = QtCore.Signal()
 
     # Use Qt-specific channel classes that emit signals.
-    sub_channel_class = Type(QtSubChannelMixin)
+    iopub_channel_class = Type(QtIOPubChannelMixin)
     shell_channel_class = Type(QtShellChannelMixin)
     stdin_channel_class = Type(QtStdInChannelMixin)
     hb_channel_class = Type(QtHBChannelMixin)

IPython/frontend/qt/console/frontend_widget.py

@@ -396,7 +396,7 b' class FrontendWidget(HistoryConsoleWidget, BaseFrontendMixin):'
         if info and info.kind == 'user' and not self._hidden:
             # Make sure that all output from the SUB channel has been processed
             # before writing a new prompt.
-            self.kernel_manager.sub_channel.flush()
+            self.kernel_manager.iopub_channel.flush()
 
             # Reset the ANSI style information to prevent bad text in stdout
             # from messing up our colors. We're not a true terminal so we're
@@ -431,7 +431,7 b' class FrontendWidget(HistoryConsoleWidget, BaseFrontendMixin):'
 
         # Make sure that all output from the SUB channel has been processed
         # before entering readline mode.
-        self.kernel_manager.sub_channel.flush()
+        self.kernel_manager.iopub_channel.flush()
 
         def callback(line):
             self.kernel_manager.stdin_channel.input(line)

IPython/frontend/qt/inprocess_kernelmanager.py

@@ -3,23 +3,23 b''
 
 # Local imports.
 from IPython.inprocess.kernelmanager import \
-    ShellInProcessChannel, SubInProcessChannel, StdInInProcessChannel, \
-    HBInProcessChannel, InProcessKernelManager
+    InProcessShellChannel, InProcessIOPubChannel, InProcessStdInChannel, \
+    InProcessHBChannel, InProcessKernelManager
 from IPython.utils.traitlets import Type
-from base_kernelmanager import QtShellChannelMixin, QtSubChannelMixin, \
+from base_kernelmanager import QtShellChannelMixin, QtIOPubChannelMixin, \
     QtStdInChannelMixin, QtHBChannelMixin, QtKernelManagerMixin
 
 
-class QtShellInProcessChannel(QtShellChannelMixin, ShellInProcessChannel):
+class QtInProcessShellChannel(QtShellChannelMixin, InProcessShellChannel):
     pass
 
-class QtSubInProcessChannel(QtSubChannelMixin, SubInProcessChannel):
+class QtInProcessIOPubChannel(QtIOPubChannelMixin, InProcessIOPubChannel):
     pass
 
-class QtStdInInProcessChannel(QtStdInChannelMixin, StdInInProcessChannel):
+class QtInProcessStdInChannel(QtStdInChannelMixin, InProcessStdInChannel):
     pass
 
-class QtHBInProcessChannel(QtHBChannelMixin, HBInProcessChannel):
+class QtInProcessHBChannel(QtHBChannelMixin, InProcessHBChannel):
     pass
 
 
@@ -27,7 +27,7 b' class QtInProcessKernelManager(QtKernelManagerMixin, InProcessKernelManager):'
     """ An in-process KernelManager with signals and slots.
     """
 
-    sub_channel_class = Type(QtSubInProcessChannel)
-    shell_channel_class = Type(QtShellInProcessChannel)
-    stdin_channel_class = Type(QtStdInInProcessChannel)
-    hb_channel_class = Type(QtHBInProcessChannel)
+    iopub_channel_class = Type(QtInProcessIOPubChannel)
+    shell_channel_class = Type(QtInProcessShellChannel)
+    stdin_channel_class = Type(QtInProcessStdInChannel)
+    hb_channel_class = Type(QtInProcessHBChannel)

IPython/frontend/qt/kernelmanager.py

@@ -3,22 +3,22 b''
 
 # Local imports.
 from IPython.utils.traitlets import Type
-from IPython.zmq.kernelmanager import ShellSocketChannel, SubSocketChannel, \
-    StdInSocketChannel, HBSocketChannel, KernelManager
-from base_kernelmanager import QtShellChannelMixin, QtSubChannelMixin, \
+from IPython.zmq.kernelmanager import ShellChannel, IOPubChannel, \
+    StdInChannel, HBChannel, KernelManager
+from base_kernelmanager import QtShellChannelMixin, QtIOPubChannelMixin, \
     QtStdInChannelMixin, QtHBChannelMixin, QtKernelManagerMixin
 
 
-class QtShellSocketChannel(QtShellChannelMixin, ShellSocketChannel):
+class QtShellChannel(QtShellChannelMixin, ShellChannel):
     pass
 
-class QtSubSocketChannel(QtSubChannelMixin, SubSocketChannel):
+class QtIOPubChannel(QtIOPubChannelMixin, IOPubChannel):
     pass
 
-class QtStdInSocketChannel(QtStdInChannelMixin, StdInSocketChannel):
+class QtStdInChannel(QtStdInChannelMixin, StdInChannel):
     pass
 
-class QtHBSocketChannel(QtHBChannelMixin, HBSocketChannel):
+class QtHBChannel(QtHBChannelMixin, HBChannel):
     pass
 
 
@@ -26,7 +26,7 b' class QtKernelManager(QtKernelManagerMixin, KernelManager):'
     """ A KernelManager that provides signals and slots.
     """
 
-    sub_channel_class = Type(QtSubSocketChannel)
-    shell_channel_class = Type(QtShellSocketChannel)
-    stdin_channel_class = Type(QtStdInSocketChannel)
-    hb_channel_class = Type(QtHBSocketChannel)
+    iopub_channel_class = Type(QtIOPubChannel)
+    shell_channel_class = Type(QtShellChannel)
+    stdin_channel_class = Type(QtStdInChannel)
+    hb_channel_class = Type(QtHBChannel)

IPython/frontend/terminal/console/interactiveshell.py

@@ -211,8 +211,8 b' class ZMQTerminalInteractiveShell(TerminalInteractiveShell):'
            sub_msg:  message receive from kernel in the sub socket channel
                      capture by kernel manager.
         """
-        while self.km.sub_channel.msg_ready():
-            sub_msg = self.km.sub_channel.get_msg()
+        while self.km.iopub_channel.msg_ready():
+            sub_msg = self.km.iopub_channel.get_msg()
             msg_type = sub_msg['header']['msg_type']
             parent = sub_msg["parent_header"]
             if (not parent) or self.session_id == parent['session']:

IPython/inprocess/blockingkernelmanager.py

@@ -14,58 +14,25 b' Useful for test suites and blocking terminal interfaces.'
 #-----------------------------------------------------------------------------
 from __future__ import print_function
 
-# Standard library imports.
-import Queue
-from threading import Event
-
 # Local imports.
 from IPython.utils.io import raw_print
 from IPython.utils.traitlets import Type
-from kernelmanager import InProcessKernelManager, ShellInProcessChannel, \
-    SubInProcessChannel, StdInInProcessChannel
-
-#-----------------------------------------------------------------------------
-# Utility classes
-#-----------------------------------------------------------------------------
+from kernelmanager import InProcessKernelManager, InProcessShellChannel, \
+    InProcessIOPubChannel, InProcessStdInChannel
+from IPython.zmq.blockingkernelmanager import BlockingChannelMixin
 
-class BlockingChannelMixin(object):
-    
-    def __init__(self, *args, **kwds):
-        super(BlockingChannelMixin, self).__init__(*args, **kwds)
-        self._in_queue = Queue.Queue()
-        
-    def call_handlers(self, msg):
-        self._in_queue.put(msg)
-        
-    def get_msg(self, block=True, timeout=None):
-        """ Gets a message if there is one that is ready. """
-        return self._in_queue.get(block, timeout)
-        
-    def get_msgs(self):
-        """ Get all messages that are currently ready. """
-        msgs = []
-        while True:
-            try:
-                msgs.append(self.get_msg(block=False))
-            except Queue.Empty:
-                break
-        return msgs
-    
-    def msg_ready(self):
-        """ Is there a message that has been received? """
-        return not self._in_queue.empty()
 
 #-----------------------------------------------------------------------------
 # Blocking kernel manager
 #-----------------------------------------------------------------------------
 
-class BlockingShellInProcessChannel(BlockingChannelMixin, ShellInProcessChannel):
+class BlockingInProcessShellChannel(BlockingChannelMixin, InProcessShellChannel):
     pass
 
-class BlockingSubInProcessChannel(BlockingChannelMixin, SubInProcessChannel):
+class BlockingInProcessIOPubChannel(BlockingChannelMixin, InProcessIOPubChannel):
     pass
 
-class BlockingStdInInProcessChannel(BlockingChannelMixin, StdInInProcessChannel):
+class BlockingInProcessStdInChannel(BlockingChannelMixin, InProcessStdInChannel):
     
     def call_handlers(self, msg):
         """ Overridden for the in-process channel.
@@ -82,6 +49,6 b' class BlockingStdInInProcessChannel(BlockingChannelMixin, StdInInProcessChannel)'
 class BlockingInProcessKernelManager(InProcessKernelManager):
 
     # The classes to use for the various channels.
-    shell_channel_class = Type(BlockingShellInProcessChannel)
-    sub_channel_class = Type(BlockingSubInProcessChannel)
-    stdin_channel_class = Type(BlockingStdInInProcessChannel)
+    shell_channel_class = Type(BlockingInProcessShellChannel)
+    iopub_channel_class = Type(BlockingInProcessIOPubChannel)
+    stdin_channel_class = Type(BlockingInProcessStdInChannel)

IPython/inprocess/ipkernel.py

@@ -123,7 +123,7 b' class InProcessKernel(Kernel):'
         """
         ident, msg = self.session.recv(self.iopub_socket, copy=False)
         for frontend in self.frontends:
-            frontend.sub_channel.call_handlers(msg)
+            frontend.iopub_channel.call_handlers(msg)
         
     #------ Trait initializers -----------------------------------------------
 

IPython/inprocess/kernelmanager.py

@@ -12,17 +12,21 b''
 #-----------------------------------------------------------------------------
 
 # Local imports.
-from IPython.config.loader import Config
+from IPython.config.configurable import Configurable
 from IPython.inprocess.socket import DummySocket
-from IPython.utils.traitlets import HasTraits, Any, Instance, Type
+from IPython.utils.traitlets import Any, Instance, Type
+from IPython.zmq.kernelmanagerabc import (
+    ShellChannelABC, IOPubChannelABC,
+    HBChannelABC, StdInChannelABC,
+    KernelManagerABC
+)
 
 #-----------------------------------------------------------------------------
 # Channel classes
 #-----------------------------------------------------------------------------
 
 class InProcessChannel(object):
-    """ Base class for in-process channels.
-    """
+    """Base class for in-process channels."""
 
     def __init__(self, manager):
         super(InProcessChannel, self).__init__()
@@ -71,9 +75,8 b' class InProcessChannel(object):'
         raise NotImplementedError
 
 
-class ShellInProcessChannel(InProcessChannel):
-    """The DEALER channel for issues request/replies to the kernel.
-    """
+class InProcessShellChannel(InProcessChannel):
+    """See `IPython.zmq.kernelmanager.ShellChannel` for docstrings."""
 
     # flag for whether execute requests should be allowed to call raw_input
     allow_stdin = True
@@ -84,42 +87,6 b' class ShellInProcessChannel(InProcessChannel):'
 
     def execute(self, code, silent=False, store_history=True,
                 user_variables=[], user_expressions={}, 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_variables : list, optional
-            A list of variable names to pull from the user's namespace.  They
-            will come back as a dict with these names as keys and their
-            :func:`repr` as values.
-
-        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 allow_stdin is None:
             allow_stdin = self.allow_stdin
         content = dict(code=code, silent=silent, store_history=store_history,
@@ -131,81 +98,18 b' class ShellInProcessChannel(InProcessChannel):'
         return msg['header']['msg_id']
 
     def complete(self, text, line, cursor_pos, block=None):
-        """Tab complete text in the kernel's namespace.
-
-        Parameters
-        ----------
-        text : str
-            The text to complete.
-        line : str
-            The full line of text that is the surrounding context for the
-            text to complete.
-        cursor_pos : int
-            The position of the cursor in the line where the completion was
-            requested.
-        block : str, optional
-            The full block of code in which the completion is being requested.
-
-        Returns
-        -------
-        The msg_id of the message sent.
-        """
         content = dict(text=text, line=line, block=block, cursor_pos=cursor_pos)
         msg = self.manager.session.msg('complete_request', content)
         self._dispatch_to_kernel(msg)
         return msg['header']['msg_id']
 
     def object_info(self, oname, detail_level=0):
-        """Get metadata information about an object.
-
-        Parameters
-        ----------
-        oname : str
-            A string specifying the object name.
-        detail_level : int, optional
-            The level of detail for the introspection (0-2)
-
-        Returns
-        -------
-        The msg_id of the message sent.
-        """
         content = dict(oname=oname, detail_level=detail_level)
         msg = self.manager.session.msg('object_info_request', content)
         self._dispatch_to_kernel(msg)
         return msg['header']['msg_id']
 
     def history(self, raw=True, output=False, hist_access_type='range', **kwds):
-        """Get entries from the 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, **kwds)
         msg = self.manager.session.msg('history_request', content)
@@ -213,10 +117,6 b' class ShellInProcessChannel(InProcessChannel):'
         return msg['header']['msg_id']
 
     def shutdown(self, restart=False):
-        """ Request an immediate kernel shutdown.
-
-        A dummy method for the in-process kernel.
-        """
         # FIXME: What to do here?
         raise NotImplementedError('Cannot shutdown in-process kernel')
 
@@ -240,49 +140,39 b' class ShellInProcessChannel(InProcessChannel):'
         self.call_handlers_later(reply_msg)
 
 
-class SubInProcessChannel(InProcessChannel):
-    """The SUB channel which listens for messages that the kernel publishes. 
-    """
+class InProcessIOPubChannel(InProcessChannel):
+    """See `IPython.zmq.kernelmanager.IOPubChannel` for docstrings."""
 
     def flush(self, timeout=1.0):
-        """ Immediately processes all pending messages on the SUB channel.
-
-        A dummy method for the in-process kernel.
-        """
         pass
 
 
-class StdInInProcessChannel(InProcessChannel):
-    """ A reply channel to handle raw_input requests that the kernel makes. """
+class InProcessStdInChannel(InProcessChannel):
+    """See `IPython.zmq.kernelmanager.StdInChannel` for docstrings."""
 
     def input(self, string):
-        """ Send a string of raw input to the kernel. 
-        """
         kernel = self.manager.kernel
         if kernel is None:
             raise RuntimeError('Cannot send input reply. No kernel exists.')
         kernel.raw_input_str = string
 
 
-class HBInProcessChannel(InProcessChannel):
-    """ A dummy heartbeat channel. """
+class InProcessHBChannel(InProcessChannel):
+    """See `IPython.zmq.kernelmanager.HBChannel` for docstrings."""
 
     time_to_dead = 3.0
 
     def __init__(self, *args, **kwds):
-        super(HBInProcessChannel, self).__init__(*args, **kwds)
+        super(InProcessHBChannel, self).__init__(*args, **kwds)
         self._pause = True
 
     def pause(self):
-        """ Pause the heartbeat. """
         self._pause = True
 
     def unpause(self):
-        """ Unpause the heartbeat. """
         self._pause = False
 
     def is_beating(self):
-        """ Is the heartbeat running and responsive (and not paused). """
         return not self._pause
 
 
@@ -290,15 +180,15 b' class HBInProcessChannel(InProcessChannel):'
 # Main kernel manager class
 #-----------------------------------------------------------------------------
 
-class InProcessKernelManager(HasTraits):
-    """ A manager for an in-process kernel.
+class InProcessKernelManager(Configurable):
+    """A manager for an in-process kernel.
 
-    This class implements most of the interface of
-    ``IPython.zmq.kernelmanager.KernelManager`` and allows (asynchronous)
-    frontends to be used seamlessly with an in-process kernel.
+    This class implements the interface of
+    `IPython.zmq.kernelmanagerabc.KernelManagerABC` and allows
+    (asynchronous) frontends to be used seamlessly with an in-process kernel.
+    
+    See `IPython.zmq.kernelmanager.KernelManager` for docstrings.
     """
-    # Config object for passing to child configurables
-    config = Instance(Config)
 
     # The Session to use for building messages.
     session = Instance('IPython.zmq.session.Session')
@@ -310,28 +200,26 b' class InProcessKernelManager(HasTraits):'
     kernel = Instance('IPython.inprocess.ipkernel.InProcessKernel')
 
     # The classes to use for the various channels.
-    shell_channel_class = Type(ShellInProcessChannel)
-    sub_channel_class = Type(SubInProcessChannel)
-    stdin_channel_class = Type(StdInInProcessChannel)
-    hb_channel_class = Type(HBInProcessChannel)
+    shell_channel_class = Type(InProcessShellChannel)
+    iopub_channel_class = Type(InProcessIOPubChannel)
+    stdin_channel_class = Type(InProcessStdInChannel)
+    hb_channel_class = Type(InProcessHBChannel)
 
     # Protected traits.
     _shell_channel = Any
-    _sub_channel = Any
+    _iopub_channel = Any
     _stdin_channel = Any
     _hb_channel = Any
 
     #--------------------------------------------------------------------------
-    # Channel management methods:
+    # Channel management methods.
     #--------------------------------------------------------------------------
 
-    def start_channels(self, shell=True, sub=True, stdin=True, hb=True):
-        """ Starts the channels for this kernel.
-        """
+    def start_channels(self, shell=True, iopub=True, stdin=True, hb=True):
         if shell:
             self.shell_channel.start()
-        if sub:
-            self.sub_channel.start()
+        if iopub:
+            self.iopub_channel.start()
         if stdin:
             self.stdin_channel.start()
             self.shell_channel.allow_stdin = True
@@ -341,12 +229,10 b' class InProcessKernelManager(HasTraits):'
             self.hb_channel.start()
 
     def stop_channels(self):
-        """ Stops all the running channels for this kernel.
-        """
         if self.shell_channel.is_alive():
             self.shell_channel.stop()
-        if self.sub_channel.is_alive():
-            self.sub_channel.stop()
+        if self.iopub_channel.is_alive():
+            self.iopub_channel.stop()
         if self.stdin_channel.is_alive():
             self.stdin_channel.stop()
         if self.hb_channel.is_alive():
@@ -354,90 +240,74 b' class InProcessKernelManager(HasTraits):'
 
     @property
     def channels_running(self):
-        """ Are any of the channels created and running? """
-        return (self.shell_channel.is_alive() or self.sub_channel.is_alive() or
+        return (self.shell_channel.is_alive() or self.iopub_channel.is_alive() or
                 self.stdin_channel.is_alive() or self.hb_channel.is_alive())
 
+    @property
+    def shell_channel(self):
+        if self._shell_channel is None:
+            self._shell_channel = self.shell_channel_class(self)
+        return self._shell_channel
+
+    @property
+    def iopub_channel(self):
+        if self._iopub_channel is None:
+            self._iopub_channel = self.iopub_channel_class(self)
+        return self._iopub_channel
+
+    @property
+    def stdin_channel(self):
+        if self._stdin_channel is None:
+            self._stdin_channel = self.stdin_channel_class(self)
+        return self._stdin_channel
+
+    @property
+    def hb_channel(self):
+        if self._hb_channel is None:
+            self._hb_channel = self.hb_channel_class(self)
+        return self._hb_channel
+
     #--------------------------------------------------------------------------
     # Kernel management methods:
     #--------------------------------------------------------------------------
     
     def start_kernel(self, **kwds):
-        """ Starts a kernel process and configures the manager to use it.
-        """
         from IPython.inprocess.ipkernel import InProcessKernel
         self.kernel = InProcessKernel()
         self.kernel.frontends.append(self)
 
     def shutdown_kernel(self):
-        """ Attempts to the stop the kernel process cleanly. If the kernel
-        cannot be stopped and the kernel is local, it is killed.
-        """
-        self.kill_kernel()
+        self._kill_kernel()
 
     def restart_kernel(self, now=False, **kwds):
-        """ Restarts a kernel with the arguments that were used to launch it.
-
-        The 'now' parameter is ignored.
-        """
         self.shutdown_kernel()
         self.start_kernel(**kwds)
 
     @property
     def has_kernel(self):
-        """ Returns whether a kernel process has been specified for the kernel
-        manager.
-        """
         return self.kernel is not None
 
-    def kill_kernel(self):
-        """ Kill the running kernel. 
-        """
+    def _kill_kernel(self):
         self.kernel.frontends.remove(self)
         self.kernel = None
 
     def interrupt_kernel(self):
-        """ Interrupts the kernel. """
         raise NotImplementedError("Cannot interrupt in-process kernel.")
 
     def signal_kernel(self, signum):
-        """ Sends a signal to the kernel. """
         raise NotImplementedError("Cannot signal in-process kernel.")
 
     @property
     def is_alive(self):
-        """ Is the kernel process still running? """
         return True
 
-    #--------------------------------------------------------------------------
-    # Channels used for communication with the kernel:
-    #--------------------------------------------------------------------------
 
-    @property
-    def shell_channel(self):
-        """Get the REQ socket channel object to make requests of the kernel."""
-        if self._shell_channel is None:
-            self._shell_channel = self.shell_channel_class(self)
-        return self._shell_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)
-        return self._sub_channel
-
-    @property
-    def stdin_channel(self):
-        """Get the REP socket channel object to handle stdin (raw_input)."""
-        if self._stdin_channel is None:
-            self._stdin_channel = self.stdin_channel_class(self)
-        return self._stdin_channel
+#-----------------------------------------------------------------------------
+# ABC Registration
+#-----------------------------------------------------------------------------
 
-    @property
-    def hb_channel(self):
-        """Get the heartbeat socket channel object to check that the
-        kernel is alive."""
-        if self._hb_channel is None:
-            self._hb_channel = self.hb_channel_class(self)
-        return self._hb_channel
+ShellChannelABC.register(InProcessShellChannel)
+IOPubChannelABC.register(InProcessIOPubChannel)
+HBChannelABC.register(InProcessHBChannel)
+StdInChannelABC.register(InProcessStdInChannel)
+KernelManagerABC.register(InProcessKernelManager)

IPython/inprocess/tests/test_kernel.py

@@ -80,7 +80,7 b' def get_stream_message(kernel_manager, timeout=5):'
     """ Gets a single stream message synchronously from the sub channel.
     """
     while True:
-        msg = kernel_manager.sub_channel.get_msg(timeout=timeout)
+        msg = kernel_manager.iopub_channel.get_msg(timeout=timeout)
         if msg['header']['msg_type'] == 'stream':
             return msg
 

IPython/zmq/blockingkernelmanager.py

@@ -13,26 +13,58 b' Useful for test suites and blocking terminal interfaces.'
 # Imports
 #-----------------------------------------------------------------------------
 
-# Local imports.
-from IPython.inprocess.blockingkernelmanager import BlockingChannelMixin
+import Queue
+
 from IPython.utils.traitlets import Type
-from kernelmanager import KernelManager, SubSocketChannel, HBSocketChannel, \
-    ShellSocketChannel, StdInSocketChannel
+from kernelmanager import KernelManager, IOPubChannel, HBChannel, \
+    ShellChannel, StdInChannel
 
 #-----------------------------------------------------------------------------
 # Blocking kernel manager
 #-----------------------------------------------------------------------------
 
-class BlockingSubSocketChannel(BlockingChannelMixin, SubSocketChannel):
+
+class BlockingChannelMixin(object):
+    
+    def __init__(self, *args, **kwds):
+        super(BlockingChannelMixin, self).__init__(*args, **kwds)
+        self._in_queue = Queue.Queue()
+        
+    def call_handlers(self, msg):
+        self._in_queue.put(msg)
+        
+    def get_msg(self, block=True, timeout=None):
+        """ Gets a message if there is one that is ready. """
+        return self._in_queue.get(block, timeout)
+        
+    def get_msgs(self):
+        """ Get all messages that are currently ready. """
+        msgs = []
+        while True:
+            try:
+                msgs.append(self.get_msg(block=False))
+            except Queue.Empty:
+                break
+        return msgs
+    
+    def msg_ready(self):
+        """ Is there a message that has been received? """
+        return not self._in_queue.empty()
+
+
+class BlockingIOPubChannel(BlockingChannelMixin, IOPubChannel):
     pass
 
-class BlockingShellSocketChannel(BlockingChannelMixin, ShellSocketChannel):
+
+class BlockingShellChannel(BlockingChannelMixin, ShellChannel):
     pass
 
-class BlockingStdInSocketChannel(BlockingChannelMixin, StdInSocketChannel):
+
+class BlockingStdInChannel(BlockingChannelMixin, StdInChannel):
     pass
 
-class BlockingHBSocketChannel(HBSocketChannel):
+
+class BlockingHBChannel(HBChannel):
     
     # This kernel needs quicker monitoring, shorten to 1 sec.
     # less than 0.5s is unreliable, and will get occasional
@@ -43,11 +75,12 b' class BlockingHBSocketChannel(HBSocketChannel):'
         """ Pause beating on missed heartbeat. """
         pass
 
+
 class BlockingKernelManager(KernelManager):
     
     # The classes to use for the various channels.
-    shell_channel_class = Type(BlockingShellSocketChannel)
-    sub_channel_class = Type(BlockingSubSocketChannel)
-    stdin_channel_class = Type(BlockingStdInSocketChannel)
-    hb_channel_class = Type(BlockingHBSocketChannel)
-  
+    shell_channel_class = Type(BlockingShellChannel)
+    iopub_channel_class = Type(BlockingIOPubChannel)
+    stdin_channel_class = Type(BlockingStdInChannel)
+    hb_channel_class = Type(BlockingHBChannel)
+

IPython/zmq/kernelapp.py

@@ -210,9 +210,9 b' class KernelApp(BaseIPythonApplication):'
         except (IOError, OSError):
             pass
         
-        self._cleanup_ipc_files()
+        self.cleanup_ipc_files()
     
-    def _cleanup_ipc_files(self):
+    def cleanup_ipc_files(self):
         """cleanup ipc files if we wrote them"""
         if self.transport != 'ipc':
             return

IPython/zmq/kernelmanager.py

@@ -34,14 +34,20 b' from zmq import ZMQError'
 from zmq.eventloop import ioloop, zmqstream
 
 # Local imports.
-from IPython.config.loader import Config
+from IPython.config.configurable import Configurable
 from IPython.utils.localinterfaces import LOCALHOST, LOCAL_IPS
 from IPython.utils.traitlets import (
-    HasTraits, Any, Instance, Type, Unicode, Integer, Bool, CaselessStrEnum
+    Any, Instance, Type, Unicode, Integer, Bool, CaselessStrEnum
 )
 from IPython.utils.py3compat import str_to_bytes
 from IPython.zmq.entry_point import write_connection_file
 from session import Session
+from IPython.zmq.kernelmanagerabc import (
+    ShellChannelABC, IOPubChannelABC,
+    HBChannelABC, StdInChannelABC,
+    KernelManagerABC
+)
+
 
 #-----------------------------------------------------------------------------
 # Constants and exceptions
@@ -84,8 +90,7 b' def validate_string_dict(dct):'
 #-----------------------------------------------------------------------------
 
 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
@@ -95,7 +100,7 b' class ZMQSocketChannel(Thread):'
     _exiting = False
 
     def __init__(self, context, session, address):
-        """Create a channel
+        """Create a channel.
 
         Parameters
         ----------
@@ -141,7 +146,7 b' class ZMQSocketChannel(Thread):'
                 break
 
     def stop(self):
-        """Stop the channel's activity.
+        """Stop the channel's event loop and join its thread.
 
         This calls :method:`Thread.join` and returns when the thread
         terminates. :class:`RuntimeError` will be raised if
@@ -151,7 +156,9 b' class ZMQSocketChannel(Thread):'
 
     @property
     def address(self):
-        """Get the channel's address as a zmq url string ('tcp://127.0.0.1:5555').
+        """Get the channel's address as a zmq url string.
+
+        These URLS have the form: 'tcp://127.0.0.1:5555'.
         """
         return self._address
 
@@ -170,25 +177,24 b' class ZMQSocketChannel(Thread):'
         self.ioloop.add_callback(thread_send)
 
     def _handle_recv(self, msg):
-        """callback for stream.on_recv
-        
-        unpacks message, and calls handlers with it.
+        """Callback for stream.on_recv.
+
+        Unpacks message, and calls handlers with it.
         """
         ident,smsg = self.session.feed_identities(msg)
         self.call_handlers(self.session.unserialize(smsg))
     
 
 
-class ShellSocketChannel(ZMQSocketChannel):
-    """The DEALER channel for issues request/replies to the kernel.
-    """
+class ShellChannel(ZMQSocketChannel):
+    """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
 
     def __init__(self, context, session, address):
-        super(ShellSocketChannel, self).__init__(context, session, address)
+        super(ShellChannel, self).__init__(context, session, address)
         self.ioloop = ioloop.IOLoop()
 
     def run(self):
@@ -205,8 +211,9 b' class ShellSocketChannel(ZMQSocketChannel):'
             pass
 
     def stop(self):
+        """Stop the channel's event loop and join its thread."""
         self.ioloop.stop()
-        super(ShellSocketChannel, self).stop()
+        super(ShellChannel, self).stop()
 
     def call_handlers(self, msg):
         """This method is called in the ioloop thread when a message arrives.
@@ -307,7 +314,7 b' class ShellSocketChannel(ZMQSocketChannel):'
         return msg['header']['msg_id']
 
     def object_info(self, oname, detail_level=0):
-        """Get metadata information about an object.
+        """Get metadata information about an object in the kernel's namespace.
 
         Parameters
         ----------
@@ -326,7 +333,7 b' class ShellSocketChannel(ZMQSocketChannel):'
         return msg['header']['msg_id']
 
     def history(self, raw=True, output=False, hist_access_type='range', **kwargs):
-        """Get entries from the history list.
+        """Get entries from the kernel's history list.
 
         Parameters
         ----------
@@ -388,12 +395,14 b' class ShellSocketChannel(ZMQSocketChannel):'
 
 
 
-class SubSocketChannel(ZMQSocketChannel):
-    """The SUB channel which listens for messages that the kernel publishes.
+class IOPubChannel(ZMQSocketChannel):
+    """The iopub channel which listens for messages that the kernel publishes.
+
+    This channel is where all output is published to frontends.
     """
 
     def __init__(self, context, session, address):
-        super(SubSocketChannel, self).__init__(context, session, address)
+        super(IOPubChannel, self).__init__(context, session, address)
         self.ioloop = ioloop.IOLoop()
 
     def run(self):
@@ -411,8 +420,9 b' class SubSocketChannel(ZMQSocketChannel):'
             pass
 
     def stop(self):
+        """Stop the channel's event loop and join its thread."""
         self.ioloop.stop()
-        super(SubSocketChannel, self).stop()
+        super(IOPubChannel, self).stop()
 
     def call_handlers(self, msg):
         """This method is called in the ioloop thread when a message arrives.
@@ -425,7 +435,7 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.
+        """Immediately processes all pending messages on the iopub channel.
 
         Callers should use this method to ensure that :method:`call_handlers`
         has been called for all messages that have been received on the
@@ -454,13 +464,13 b' class SubSocketChannel(ZMQSocketChannel):'
         self._flushed = True
 
 
-class StdInSocketChannel(ZMQSocketChannel):
-    """A reply channel to handle raw_input requests that the kernel makes."""
+class StdInChannel(ZMQSocketChannel):
+    """The stdin channel to handle raw_input requests that the kernel makes."""
 
     msg_queue = None
 
     def __init__(self, context, session, address):
-        super(StdInSocketChannel, self).__init__(context, session, address)
+        super(StdInChannel, self).__init__(context, session, address)
         self.ioloop = ioloop.IOLoop()
 
     def run(self):
@@ -475,11 +485,11 b' class StdInSocketChannel(ZMQSocketChannel):'
             self.socket.close()
         except:
             pass
-        
 
     def stop(self):
+        """Stop the channel's event loop and join its thread."""
         self.ioloop.stop()
-        super(StdInSocketChannel, self).stop()
+        super(StdInChannel, self).stop()
 
     def call_handlers(self, msg):
         """This method is called in the ioloop thread when a message arrives.
@@ -498,7 +508,7 b' class StdInSocketChannel(ZMQSocketChannel):'
         self._queue_send(msg)
 
 
-class HBSocketChannel(ZMQSocketChannel):
+class HBChannel(ZMQSocketChannel):
     """The heartbeat channel which monitors the kernel heartbeat.
 
     Note that the heartbeat channel is paused by default. As long as you start
@@ -514,7 +524,7 b' class HBSocketChannel(ZMQSocketChannel):'
     _beating = None
 
     def __init__(self, context, session, address):
-        super(HBSocketChannel, self).__init__(context, session, address)
+        super(HBChannel, self).__init__(context, session, address)
         self._running = False
         self._pause =True
         self.poller = zmq.Poller()
@@ -531,7 +541,7 b' class HBSocketChannel(ZMQSocketChannel):'
         self.poller.register(self.socket, zmq.POLLIN)
     
     def _poll(self, start_time):
-        """poll for heartbeat replies until we reach self.time_to_dead
+        """poll for heartbeat replies until we reach self.time_to_dead.
         
         Ignores interrupts, and returns the result of poll(), which
         will be an empty list if no messages arrived before the timeout,
@@ -620,8 +630,9 b' class HBSocketChannel(ZMQSocketChannel):'
             return False
 
     def stop(self):
+        """Stop the channel's event loop and join its thread."""
         self._running = False
-        super(HBSocketChannel, self).stop()
+        super(HBChannel, self).stop()
 
     def call_handlers(self, since_last_heartbeat):
         """This method is called in the ioloop thread when a message arrives.
@@ -638,20 +649,23 b' class HBSocketChannel(ZMQSocketChannel):'
 # Main kernel manager class
 #-----------------------------------------------------------------------------
 
-class KernelManager(HasTraits):
-    """ Manages a kernel for a frontend.
+class KernelManager(Configurable):
+    """Manages a single kernel on this host along with its channels.
 
-    The SUB channel is for the frontend to receive messages published by the
-    kernel.
+    There are four channels associated with each kernel:
 
-    The REQ channel is for the frontend to make requests of the kernel.
+    * shell: for request/reply calls to the kernel.
+    * iopub: for the kernel to publish results to frontends.
+    * hb: for monitoring the kernel's heartbeat.
+    * stdin: for frontends to reply to raw_input calls in the kernel.
 
-    The REP channel is for the kernel to request stdin (raw_input) from the
-    frontend.
-    """
-    # config object for passing to child configurables
-    config = Instance(Config)
+    The usage of the channels that this class manages is optional. It is
+    entirely possible to connect to the kernels directly using ZeroMQ
+    sockets. These channels are useful primarily for talking to a kernel
+    whose :class:`KernelManager` is in the same process.
 
+    This version manages kernels started using Popen.
+    """
     # The PyZMQ Context to use for communication with the kernel.
     context = Instance(zmq.Context)
     def _context_default(self):
@@ -668,10 +682,9 b' class KernelManager(HasTraits):'
     # The addresses for the communication channels.
     connection_file = Unicode('')
     
-    transport = CaselessStrEnum(['tcp', 'ipc'], default_value='tcp')
-    
-    
-    ip = Unicode(LOCALHOST)
+    transport = CaselessStrEnum(['tcp', 'ipc'], default_value='tcp', config=True)
+
+    ip = Unicode(LOCALHOST, config=True)
     def _ip_changed(self, name, old, new):
         if new == '*':
             self.ip = '0.0.0.0'
@@ -681,38 +694,39 b' class KernelManager(HasTraits):'
     hb_port = Integer(0)
 
     # The classes to use for the various channels.
-    shell_channel_class = Type(ShellSocketChannel)
-    sub_channel_class = Type(SubSocketChannel)
-    stdin_channel_class = Type(StdInSocketChannel)
-    hb_channel_class = Type(HBSocketChannel)
+    shell_channel_class = Type(ShellChannel)
+    iopub_channel_class = Type(IOPubChannel)
+    stdin_channel_class = Type(StdInChannel)
+    hb_channel_class = Type(HBChannel)
 
     # Protected traits.
     _launch_args = Any
     _shell_channel = Any
-    _sub_channel = Any
+    _iopub_channel = Any
     _stdin_channel = Any
     _hb_channel = Any
     _connection_file_written=Bool(False)
-    
-    def __del__(self):
+ 	
+    def __del__(self):	  	
         self.cleanup_connection_file()
-    
+
     #--------------------------------------------------------------------------
     # Channel management methods:
     #--------------------------------------------------------------------------
 
-    def start_channels(self, shell=True, sub=True, stdin=True, hb=True):
+    def start_channels(self, shell=True, iopub=True, stdin=True, hb=True):
         """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.
+        them (their activity runs in a thread). 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 shell:
             self.shell_channel.start()
-        if sub:
-            self.sub_channel.start()
+        if iopub:
+            self.iopub_channel.start()
         if stdin:
             self.stdin_channel.start()
             self.shell_channel.allow_stdin = True
@@ -723,11 +737,13 b' class KernelManager(HasTraits):'
 
     def stop_channels(self):
         """Stops all the running channels for this kernel.
+
+        This stops their event loops and joins their threads.
         """
         if self.shell_channel.is_alive():
             self.shell_channel.stop()
-        if self.sub_channel.is_alive():
-            self.sub_channel.stop()
+        if self.iopub_channel.is_alive():
+            self.iopub_channel.stop()
         if self.stdin_channel.is_alive():
             self.stdin_channel.stop()
         if self.hb_channel.is_alive():
@@ -736,15 +752,64 b' class KernelManager(HasTraits):'
     @property
     def channels_running(self):
         """Are any of the channels created and running?"""
-        return (self.shell_channel.is_alive() or self.sub_channel.is_alive() or
+        return (self.shell_channel.is_alive() or self.iopub_channel.is_alive() or
                 self.stdin_channel.is_alive() or self.hb_channel.is_alive())
 
+    def _make_url(self, port):
+        """Make a zmq url with a port.
+
+        There are two cases that this handles:
+
+        * tcp: tcp://ip:port
+        * ipc: ipc://ip-port
+        """
+        if self.transport == 'tcp':
+            return "tcp://%s:%i" % (self.ip, port)
+        else:
+            return "%s://%s-%s" % (self.transport, self.ip, port)
+
+    @property
+    def shell_channel(self):
+        """Get the shell channel object for this kernel."""
+        if self._shell_channel is None:
+            self._shell_channel = self.shell_channel_class(
+                self.context, self.session, self._make_url(self.shell_port)
+            )
+        return self._shell_channel
+
+    @property
+    def iopub_channel(self):
+        """Get the iopub channel object for this kernel."""
+        if self._iopub_channel is None:
+            self._iopub_channel = self.iopub_channel_class(
+                self.context, self.session, self._make_url(self.iopub_port)
+            )
+        return self._iopub_channel
+
+    @property
+    def stdin_channel(self):
+        """Get the stdin channel object for this kernel."""
+        if self._stdin_channel is None:
+            self._stdin_channel = self.stdin_channel_class(
+                self.context, self.session, self._make_url(self.stdin_port)
+            )
+        return self._stdin_channel
+
+    @property
+    def hb_channel(self):
+        """Get the hb channel object for this kernel."""
+        if self._hb_channel is None:
+            self._hb_channel = self.hb_channel_class(
+                self.context, self.session, self._make_url(self.hb_port)
+            )
+        return self._hb_channel
+
     #--------------------------------------------------------------------------
-    # Kernel process management methods:
+    # Connection and ipc file management
     #--------------------------------------------------------------------------
     
     def cleanup_connection_file(self):
-        """cleanup connection file *if we wrote it*
+        """Cleanup connection file *if we wrote it*
         
         Will not raise if the connection file was already removed somehow.
         """
@@ -755,11 +820,9 b' class KernelManager(HasTraits):'
                 os.remove(self.connection_file)
             except (IOError, OSError):
                 pass
-            
-            self._cleanup_ipc_files()
     
-    def _cleanup_ipc_files(self):
-        """cleanup ipc files if we wrote them"""
+    def cleanup_ipc_files(self):
+        """Cleanup ipc files if we wrote them."""
         if self.transport != 'ipc':
             return
         for port in (self.shell_port, self.iopub_port, self.stdin_port, self.hb_port):
@@ -768,12 +831,12 b' class KernelManager(HasTraits):'
                 os.remove(ipcfile)
             except (IOError, OSError):
                 pass
-    
+
     def load_connection_file(self):
-        """load connection info from JSON dict in self.connection_file"""
+        """Load connection info from JSON dict in self.connection_file."""
         with open(self.connection_file) as f:
             cfg = json.loads(f.read())
-        
+
         from pprint import pprint
         pprint(cfg)
         self.transport = cfg.get('transport', 'tcp')
@@ -785,7 +848,7 b' class KernelManager(HasTraits):'
         self.session.key = str_to_bytes(cfg['key'])
     
     def write_connection_file(self):
-        """write connection info to JSON dict in self.connection_file"""
+        """Write connection info to JSON dict in self.connection_file."""
         if self._connection_file_written:
             return
         self.connection_file,cfg = write_connection_file(self.connection_file,
@@ -799,9 +862,13 b' class KernelManager(HasTraits):'
         self.hb_port = cfg['hb_port']
         
         self._connection_file_written = True
-    
+
+    #--------------------------------------------------------------------------
+    # Kernel management
+    #--------------------------------------------------------------------------
+
     def start_kernel(self, **kw):
-        """Starts a kernel process and configures the manager to use it.
+        """Starts a kernel on this host in a separate process.
 
         If random ports (port=0) are being used, this method must be called
         before the channels are created.
@@ -814,7 +881,8 b' class KernelManager(HasTraits):'
              it should not be necessary to use this parameter.
 
         **kw : optional
-             See respective options for IPython and Python kernels.
+             keyword arguments that are passed down into the launcher
+             callable.
         """
         if self.transport == 'tcp' and self.ip not in LOCAL_IPS:
             raise RuntimeError("Can only launch a kernel on a local interface. "
@@ -832,47 +900,62 b' class KernelManager(HasTraits):'
             from ipkernel import launch_kernel
         self.kernel = launch_kernel(fname=self.connection_file, **kw)
 
-    def shutdown_kernel(self, restart=False):
-        """ Attempts to the stop the kernel process cleanly. 
+    def shutdown_kernel(self, now=False, restart=False):
+        """Attempts to the stop the kernel process cleanly. 
+
+        This attempts to shutdown the kernels cleanly by:
+
+        1. Sending it a shutdown message over the shell channel.
+        2. If that fails, the kernel is shutdown forcibly by sending it
+           a signal.
 
-        If the kernel cannot be stopped and the kernel is local, it is killed.
+        Parameters:
+        -----------
+        now : bool
+            Should the kernel be forcible killed *now*. This skips the
+            first, nice shutdown attempt.
+        restart: bool
+            Will this kernel be restarted after it is shutdown. When this
+            is True, connection files will not be cleaned up.
         """
         # FIXME: Shutdown does not work on Windows due to ZMQ errors!
         if sys.platform == 'win32':
-            self.kill_kernel()
+            self._kill_kernel()
             return
 
         # Pause the heart beat channel if it exists.
         if self._hb_channel is not None:
             self._hb_channel.pause()
 
-        # Don't send any additional kernel kill messages immediately, to give
-        # the kernel a chance to properly execute shutdown actions. Wait for at
-        # most 1s, checking every 0.1s.
-        self.shell_channel.shutdown(restart=restart)
-        for i in range(10):
-            if self.is_alive:
-                time.sleep(0.1)
-            else:
-                break
-        else:
-            # OK, we've waited long enough.
+        if now:
             if self.has_kernel:
-                self.kill_kernel()
+                self._kill_kernel()
+        else:
+            # Don't send any additional kernel kill messages immediately, to give
+            # the kernel a chance to properly execute shutdown actions. Wait for at
+            # most 1s, checking every 0.1s.
+            self.shell_channel.shutdown(restart=restart)
+            for i in range(10):
+                if self.is_alive:
+                    time.sleep(0.1)
+                else:
+                    break
+            else:
+                # OK, we've waited long enough.
+                if self.has_kernel:
+                    self._kill_kernel()
 
-        if not restart and self._connection_file_written:
-            # cleanup connection files on full shutdown of kernel we started
-            self._connection_file_written = False
-            try:
-                os.remove(self.connection_file)
-            except IOError:
-                pass
+        if not restart:
+            self.cleanup_connection_file()
+            self.cleanup_ipc_files()
+        else:
+            self.cleanup_ipc_files()
 
     def restart_kernel(self, now=False, **kw):
         """Restarts a kernel with the arguments that were used to launch it.
 
         If the old kernel was launched with random ports, the same ports will be
-        used for the new kernel.
+        used for the new kernel. The same connection file is used again.
 
         Parameters
         ----------
@@ -885,7 +968,7 b' class KernelManager(HasTraits):'
             it is given a chance to perform a clean shutdown or not.
 
         **kw : optional
-            Any options specified here will replace those used to launch the
+            Any options specified here will overwrite those used to launch the
             kernel.
         """
         if self._launch_args is None:
@@ -893,11 +976,7 b' class KernelManager(HasTraits):'
                                "No previous call to 'start_kernel'.")
         else:
             # Stop currently running kernel.
-            if self.has_kernel:
-                if now:
-                    self.kill_kernel()
-                else:
-                    self.shutdown_kernel(restart=True)
+            self.shutdown_kernel(now=now, restart=True)
 
             # Start new kernel.
             self._launch_args.update(kw)
@@ -910,15 +989,13 b' class KernelManager(HasTraits):'
 
     @property
     def has_kernel(self):
-        """Returns whether a kernel process has been specified for the kernel
-        manager.
-        """
+        """Has a kernel been started that we are managing."""
         return self.kernel is not None
 
-    def kill_kernel(self):
-        """ Kill the running kernel.
+    def _kill_kernel(self):
+        """Kill the running kernel.
 
-        This method blocks until the kernel process has terminated.
+        This is a private method, callers should use shutdown_kernel(now=True).
         """
         if self.has_kernel:
             # Pause the heart beat channel if it exists.
@@ -949,7 +1026,7 b' class KernelManager(HasTraits):'
             raise RuntimeError("Cannot kill kernel. No kernel is running!")
 
     def interrupt_kernel(self):
-        """ Interrupts the kernel.
+        """Interrupts the kernel by sending it a signal.
 
         Unlike ``signal_kernel``, this operation is well supported on all
         platforms.
@@ -964,7 +1041,7 b' class KernelManager(HasTraits):'
             raise RuntimeError("Cannot interrupt kernel. No kernel is running!")
 
     def signal_kernel(self, signum):
-        """ Sends a signal to the kernel.
+        """Sends a signal to the kernel.
 
         Note that since only SIGTERM is supported on Windows, this function is
         only useful on Unix systems.
@@ -991,54 +1068,14 b' class KernelManager(HasTraits):'
             # so naively return True
             return True
 
-    #--------------------------------------------------------------------------
-    # Channels used for communication with the kernel:
-    #--------------------------------------------------------------------------
 
-    def _make_url(self, port):
-        """make a zmq url with a port"""
-        if self.transport == 'tcp':
-            return "tcp://%s:%i" % (self.ip, port)
-        else:
-            return "%s://%s-%s" % (self.transport, self.ip, port)
-
-    @property
-    def shell_channel(self):
-        """Get the REQ socket channel object to make requests of the kernel."""
-        if self._shell_channel is None:
-            self._shell_channel = self.shell_channel_class(self.context,
-                                                           self.session,
-                                                           self._make_url(self.shell_port),
-            )
-        return self._shell_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._make_url(self.iopub_port),
-            )
-        return self._sub_channel
+#-----------------------------------------------------------------------------
+# ABC Registration
+#-----------------------------------------------------------------------------
 
-    @property
-    def stdin_channel(self):
-        """Get the REP socket channel object to handle stdin (raw_input)."""
-        if self._stdin_channel is None:
-            self._stdin_channel = self.stdin_channel_class(self.context,
-                                                           self.session,
-                                                           self._make_url(self.stdin_port),
-            )
-        return self._stdin_channel
+ShellChannelABC.register(ShellChannel)
+IOPubChannelABC.register(IOPubChannel)
+HBChannelABC.register(HBChannel)
+StdInChannelABC.register(StdInChannel)
+KernelManagerABC.register(KernelManager)
 
-    @property
-    def hb_channel(self):
-        """Get the heartbeat socket channel object to check that the
-        kernel is alive."""
-        if self._hb_channel is None:
-            self._hb_channel = self.hb_channel_class(self.context,
-                                                     self.session,
-                                                     self._make_url(self.hb_port),
-            )
-        return self._hb_channel

IPython/zmq/tests/test_message_spec.py

@@ -48,7 +48,7 b' def teardown():'
 
 def flush_channels():
     """flush any messages waiting on the queue"""
-    for channel in (KM.shell_channel, KM.sub_channel):
+    for channel in (KM.shell_channel, KM.iopub_channel):
         while True:
             try:
                 msg = channel.get_msg(block=True, timeout=0.1)
@@ -61,7 +61,7 b' def flush_channels():'
 def execute(code='', **kwargs):
     """wrapper for doing common steps for validating an execution request"""
     shell = KM.shell_channel
-    sub = KM.sub_channel
+    sub = KM.iopub_channel
     
     msg_id = shell.execute(code=code, **kwargs)
     reply = shell.get_msg(timeout=2)
@@ -310,23 +310,23 b' def test_execute_silent():'
     msg_id, reply = execute(code='x=1', silent=True)
     
     # flush status=idle
-    status = KM.sub_channel.get_msg(timeout=2)
+    status = KM.iopub_channel.get_msg(timeout=2)
     for tst in validate_message(status, 'status', msg_id):
         yield tst
     nt.assert_equal(status['content']['execution_state'], 'idle')
 
-    yield nt.assert_raises(Empty, KM.sub_channel.get_msg, timeout=0.1)
+    yield nt.assert_raises(Empty, KM.iopub_channel.get_msg, timeout=0.1)
     count = reply['execution_count']
     
     msg_id, reply = execute(code='x=2', silent=True)
     
     # flush status=idle
-    status = KM.sub_channel.get_msg(timeout=2)
+    status = KM.iopub_channel.get_msg(timeout=2)
     for tst in validate_message(status, 'status', msg_id):
         yield tst
     yield nt.assert_equal(status['content']['execution_state'], 'idle')
     
-    yield nt.assert_raises(Empty, KM.sub_channel.get_msg, timeout=0.1)
+    yield nt.assert_raises(Empty, KM.iopub_channel.get_msg, timeout=0.1)
     count_2 = reply['execution_count']
     yield nt.assert_equal(count_2, count)
 
@@ -339,7 +339,7 b' def test_execute_error():'
     yield nt.assert_equal(reply['status'], 'error')
     yield nt.assert_equal(reply['ename'], 'ZeroDivisionError')
     
-    pyerr = KM.sub_channel.get_msg(timeout=2)
+    pyerr = KM.iopub_channel.get_msg(timeout=2)
     for tst in validate_message(pyerr, 'pyerr', msg_id):
         yield tst
 
@@ -475,7 +475,7 b' def test_stream():'
 
     msg_id, reply = execute("print('hi')")
 
-    stdout = KM.sub_channel.get_msg(timeout=2)
+    stdout = KM.iopub_channel.get_msg(timeout=2)
     for tst in validate_message(stdout, 'stream', msg_id):
         yield tst
     content = stdout['content']
@@ -489,7 +489,7 b' def test_display_data():'
 
     msg_id, reply = execute("from IPython.core.display import display; display(1)")
     
-    display = KM.sub_channel.get_msg(timeout=2)
+    display = KM.iopub_channel.get_msg(timeout=2)
     for tst in validate_message(display, 'display_data', parent=msg_id):
         yield tst
     data = display['content']['data']