Refactoring kernel restarting.

IPython/kernel/ioloopkernelmanager.py

@@ -0,0 +1,50 b''
+"""A kernel manager with ioloop based logic."""
+
+#-----------------------------------------------------------------------------
+#  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
+#-----------------------------------------------------------------------------
+
+from __future__ import absolute_import
+
+import zmq
+from zmq.eventloop import ioloop
+
+from IPython.utils.traitlets import (
+    Instance
+)
+
+from .blockingkernelmanager import BlockingKernelManager
+from .ioloopkernelrestarter import IOLoopKernelRestarter
+
+#-----------------------------------------------------------------------------
+# Code
+#-----------------------------------------------------------------------------
+
+class IOLoopKernelManager(BlockingKernelManager):
+
+    loop = Instance('zmq.eventloop.ioloop.IOLoop', allow_none=False)
+    def _loop_default(self):
+        return ioloop.IOLoop.instance()
+
+    _restarter = Instance('IPython.kernel.ioloopkernelrestarter.IOLoopKernelRestarter')
+
+    def start_restarter(self):
+        if self.autorestart and self.has_kernel:
+            if self._restarter is None:
+                self._restarter = IOLoopKernelRestarter(
+                    kernel_manager=self, loop=self.loop,
+                    config=self.config, log=self.log
+                )
+            self._restarter.start()
+
+    def stop_restarter(self):
+        if self.autorestart:
+            if self._restarter is not None:
+                self._restarter.stop()

IPython/frontend/html/notebook/handlers.py

@@ -407,6 +407,9 b' class ZMQStreamHandler(websocket.WebSocketHandler):'
         return jsonapi.dumps(msg, default=date_default)
 
     def _on_zmq_reply(self, msg_list):
+        # Sometimes this gets triggered when the on_close method is scheduled in the
+        # eventloop but hasn't been called.
+        if self.stream.closed(): return
         try:
             msg = self._reserialize_reply(msg_list)
         except Exception:

IPython/kernel/blockingkernelmanager.py

@@ -81,7 +81,7 b' class BlockingHBChannel(HBChannel):'
 
 
 class BlockingKernelManager(KernelManager):
-    
+
     # The classes to use for the various channels.
     shell_channel_class = Type(BlockingShellChannel)
     iopub_channel_class = Type(BlockingIOPubChannel)

IPython/kernel/ioloopkernelrestarter.py

@@ -30,7 +30,7 b' from IPython.utils.traitlets import ('
 # Code
 #-----------------------------------------------------------------------------
 
-class KernelRestarter(LoggingConfigurable):
+class IOLoopKernelRestarter(LoggingConfigurable):
     """Monitor and autorestart a kernel."""
 
     loop = Instance('zmq.eventloop.ioloop.IOLoop', allow_none=False)
@@ -45,14 +45,11 b' class KernelRestarter(LoggingConfigurable):'
 
     _pcallback = None
 
-    def __init__(self, **kwargs):
-        super(KernelRestarter, self).__init__(**kwargs)
-
     def start(self):
         """Start the polling of the kernel."""
         if self._pcallback is None:
             self._pcallback = ioloop.PeriodicCallback(
-                self._poll, 1000*self.time_to_dead, self.ioloop
+                self._poll, 1000*self.time_to_dead, self.loop
             )
         self._pcallback.start()
 
@@ -68,12 +65,10 b' class KernelRestarter(LoggingConfigurable):'
             self._pcallback = None
 
     def _poll(self):
+        self.log.info('Polling kernel...')
         if not self.kernel_manager.is_alive():
-            self.stop()
             # This restart event should leave the connection file in place so
             # the ports are the same. Because this takes place below the
             # MappingKernelManager, the kernel_id will also remain the same.
-            self.log('KernelRestarter: restarting kernel')
+            self.log.info('KernelRestarter: restarting kernel')
             self.kernel_manager.restart_kernel(now=True);
-            self.start()
-

IPython/kernel/kernelmanager.py

@@ -21,14 +21,12 b' from __future__ import absolute_import'
 import atexit
 import errno
 import json
-from subprocess import Popen
 import os
 import signal
 import sys
 from threading import Thread
 import time
 
-# System library imports
 import zmq
 # import ZMQError in top-level namespace, to avoid ugly attribute-error messages
 # during garbage collection of threads at exit:
@@ -37,9 +35,11 b' from zmq.eventloop import ioloop, zmqstream'
 
 # Local imports
 from IPython.config.configurable import Configurable
+from IPython.utils.importstring import import_item
 from IPython.utils.localinterfaces import LOCALHOST, LOCAL_IPS
 from IPython.utils.traitlets import (
-    Any, Instance, Type, Unicode, List, Integer, Bool, CaselessStrEnum
+    Any, Instance, Type, Unicode, List, Integer, Bool,
+    CaselessStrEnum, DottedObjectName
 )
 from IPython.utils.py3compat import str_to_bytes
 from IPython.kernel import (
@@ -54,7 +54,6 b' from .kernelmanagerabc import ('
     KernelManagerABC
 )
 
-
 #-----------------------------------------------------------------------------
 # Constants and exceptions
 #-----------------------------------------------------------------------------
@@ -691,11 +690,11 b' class KernelManager(Configurable):'
         Override this if you have a custom 
         """
     )
+
     def _kernel_cmd_changed(self, name, old, new):
         self.ipython_kernel = False
 
     ipython_kernel = Bool(True)
-    
 
     # The addresses for the communication channels.
     connection_file = Unicode('')
@@ -708,6 +707,7 b' class KernelManager(Configurable):'
         Consoles on other machines will be able to connect
         to the Kernel, so be careful!"""
     )
+
     def _ip_default(self):
         if self.transport == 'ipc':
             if self.connection_file:
@@ -716,9 +716,11 b' class KernelManager(Configurable):'
                 return 'kernel-ipc'
         else:
             return LOCALHOST
+
     def _ip_changed(self, name, old, new):
         if new == '*':
             self.ip = '0.0.0.0'
+
     shell_port = Integer(0)
     iopub_port = Integer(0)
     stdin_port = Integer(0)
@@ -738,6 +740,10 b' class KernelManager(Configurable):'
     _hb_channel = Any
     _connection_file_written=Bool(False)
 
+    autorestart = Bool(False, config=True,
+        help="""Should we autorestart the kernel if it dies."""
+    )
+
     def __del__(self):
         self.cleanup_connection_file()
 
@@ -895,9 +901,19 b' class KernelManager(Configurable):'
         self._connection_file_written = True
 
     #--------------------------------------------------------------------------
+    # Kernel restarter
+    #--------------------------------------------------------------------------
+
+    def start_restarter(self):
+        pass
+
+    def stop_restarter(self):
+        pass
+
+    #--------------------------------------------------------------------------
     # Kernel management
     #--------------------------------------------------------------------------
-    
+
     def format_kernel_cmd(self, **kw):
         """format templated args (e.g. {connection_file})"""
         if self.kernel_cmd:
@@ -948,6 +964,7 b' class KernelManager(Configurable):'
         self.kernel = self._launch_kernel(kernel_cmd,
                                     ipython_kernel=self.ipython_kernel,
                                     **kw)
+        self.start_restarter()
 
     def shutdown_kernel(self, now=False, restart=False):
         """Attempts to the stop the kernel process cleanly. 
@@ -967,15 +984,19 b' class KernelManager(Configurable):'
             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()
-            return
 
         # Pause the heart beat channel if it exists.
         if self._hb_channel is not None:
             self._hb_channel.pause()
 
+        # Stop monitoring for restarting while we shutdown.
+        self.stop_restarter()
+
+        # FIXME: Shutdown does not work on Windows due to ZMQ errors!
+        if sys.platform == 'win32':
+            self._kill_kernel()
+            return
+
         if now:
             if self.has_kernel:
                 self._kill_kernel()
@@ -1047,9 +1068,6 b' class KernelManager(Configurable):'
         This is a private method, callers should use shutdown_kernel(now=True).
         """
         if self.has_kernel:
-            # Pause the heart beat channel if it exists.
-            if self._hb_channel is not None:
-                self._hb_channel.pause()
 
             # Signal the kernel to terminate (sends SIGKILL on Unix and calls
             # TerminateProcess() on Win32).

IPython/kernel/multikernelmanager.py

@@ -23,14 +23,12 b' import uuid'
 
 import zmq
 from zmq.eventloop.zmqstream import ZMQStream
-from zmq.eventloop import ioloop
 
 from IPython.config.configurable import LoggingConfigurable
 from IPython.utils.importstring import import_item
 from IPython.utils.traitlets import (
     Instance, Dict, Unicode, Any, DottedObjectName, Bool
 )
-# from IPython.kernel.kernelrestarter import KernelRestarter
 
 #-----------------------------------------------------------------------------
 # Classes
@@ -44,7 +42,7 b' class MultiKernelManager(LoggingConfigurable):'
     """A class for managing multiple kernels."""
     
     kernel_manager_class = DottedObjectName(
-        "IPython.kernel.blockingkernelmanager.BlockingKernelManager", config=True,
+        "IPython.kernel.ioloopkernelmanager.IOLoopKernelManager", config=True,
         help="""The kernel manager class.  This is configurable to allow
         subclassing of the KernelManager for customized behavior.
         """
@@ -60,18 +58,9 b' class MultiKernelManager(LoggingConfigurable):'
     def _context_default(self):
         return zmq.Context.instance()
 
-    loop = Instance('zmq.eventloop.ioloop.IOLoop', allow_none=False)
-    def _loop_default(self):
-        return ioloop.IOLoop.instance()
-
-    autorestart = Bool(False, config=True,
-        help="""Should we autorestart kernels that die."""
-    )
-
     connection_dir = Unicode('')
 
     _kernels = Dict()
-    _restarters = Dict()
 
     def list_kernel_ids(self):
         """Return a list of the kernel ids of the active kernels."""
@@ -86,30 +75,6 b' class MultiKernelManager(LoggingConfigurable):'
     def __contains__(self, kernel_id):
         return kernel_id in self._kernels
 
-    def start_restarter(self, kernel_id):
-        km = self.get_kernel(kernel_id)
-        if self.autorestart:
-            kr = self._restarters.get(kernel_id, None)
-            if kr is None:
-                kr = KernelRestarter(
-                    kernel_manager=km, loop=self.loop,
-                    config=self.config, log=self.log
-                )
-                self._restarters[kernel_id] = kr
-            kr.start()
-
-    def stop_restarter(self, kernel_id):
-        if self.autorestart:
-            kr = self._restarters.get(kernel_id, None)
-            if kr is not None:
-                kr.stop()
-
-    def clear_restarter(self, kernel_id):
-        if self.autorestart:
-            kr = self._restarters.pop(kernel_id, None)
-            if kr is not None:
-                kr.stop()
-
     def start_kernel(self, **kwargs):
         """Start a new kernel.
 
@@ -129,13 +94,12 b' class MultiKernelManager(LoggingConfigurable):'
         # 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,
+                    config=self.config, autorestart=True, log=self.log
         )
         km.start_kernel(**kwargs)
         # start just the shell channel, needed for graceful restart
         km.start_channels(shell=True, iopub=False, stdin=False, hb=False)
         self._kernels[kernel_id] = km
-        self.start_restarter(kernel_id)
         return kernel_id
 
     def shutdown_kernel(self, kernel_id, now=False):
@@ -149,11 +113,9 b' class MultiKernelManager(LoggingConfigurable):'
             Should the kernel be shutdown forcibly using a signal.
         """
         k = self.get_kernel(kernel_id)
-        self.stop_restarter(kernel_id)
         k.shutdown_kernel(now=now)
         k.shell_channel.stop()
         del self._kernels[kernel_id]
-        self.clear_restarter(kernel_id)
 
     def shutdown_all(self, now=False):
         """Shutdown all kernels."""
@@ -192,9 +154,7 b' class MultiKernelManager(LoggingConfigurable):'
             The id of the kernel to interrupt.
         """
         km = self.get_kernel(kernel_id)
-        self.stop_restarter(kernel_id)
         km.restart_kernel()
-        self.start_restarter(kernel_id)
 
     def is_alive(self, kernel_id):
         """Is the kernel alive.

IPython/kernel/tests/test_kernelmanager.py

@@ -12,12 +12,16 b' from IPython.kernel.kernelmanager import KernelManager'
 class TestKernelManager(TestCase):
 
     def _get_tcp_km(self):
-        return KernelManager()
+        c = Config()
+        # c.KernelManager.autorestart=False
+        km = KernelManager(config=c)
+        return km
 
     def _get_ipc_km(self):
         c = Config()
         c.KernelManager.transport = 'ipc'
         c.KernelManager.ip = 'test'
+        # c.KernelManager.autorestart=False
         km = KernelManager(config=c)
         return km
 

IPython/kernel/tests/test_multikernelmanager.py

@@ -14,12 +14,16 b' from IPython.kernel.multikernelmanager import MultiKernelManager'
 class TestKernelManager(TestCase):
 
     def _get_tcp_km(self):
-        return MultiKernelManager()
+        c = Config()
+        # c.KernelManager.autorestart=False
+        km = MultiKernelManager(config=c)
+        return km
 
     def _get_ipc_km(self):
         c = Config()
         c.KernelManager.transport = 'ipc'
         c.KernelManager.ip = 'test'
+        # c.KernelManager.autorestart=False
         km = MultiKernelManager(config=c)
         return km