Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Fetching contributors…

Cannot retrieve contributors at this time

231 lines (194 sloc) 8.559 kb
"""KernelStarter class that intercepts Control Queue messages, and handles process management.
Authors:
* Min RK
"""
#-----------------------------------------------------------------------------
# Copyright (C) 2010-2011 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.
#-----------------------------------------------------------------------------
from zmq.eventloop import ioloop
from IPython.zmq.session import Session
class KernelStarter(object):
"""Object for resetting/killing the Kernel."""
def __init__(self, session, upstream, downstream, *kernel_args, **kernel_kwargs):
self.session = session
self.upstream = upstream
self.downstream = downstream
self.kernel_args = kernel_args
self.kernel_kwargs = kernel_kwargs
self.handlers = {}
for method in 'shutdown_request shutdown_reply'.split():
self.handlers[method] = getattr(self, method)
def start(self):
self.upstream.on_recv(self.dispatch_request)
self.downstream.on_recv(self.dispatch_reply)
#--------------------------------------------------------------------------
# Dispatch methods
#--------------------------------------------------------------------------
def dispatch_request(self, raw_msg):
idents, msg = self.session.feed_identities()
try:
msg = self.session.unserialize(msg, content=False)
except:
print ("bad msg: %s"%msg)
msgtype = msg['header']['msg_type']
handler = self.handlers.get(msgtype, None)
if handler is None:
self.downstream.send_multipart(raw_msg, copy=False)
else:
handler(msg)
def dispatch_reply(self, raw_msg):
idents, msg = self.session.feed_identities()
try:
msg = self.session.unserialize(msg, content=False)
except:
print ("bad msg: %s"%msg)
msgtype = msg['header']['msg_type']
handler = self.handlers.get(msgtype, None)
if handler is None:
self.upstream.send_multipart(raw_msg, copy=False)
else:
handler(msg)
#--------------------------------------------------------------------------
# Handlers
#--------------------------------------------------------------------------
def shutdown_request(self, msg):
""""""
self.downstream.send_multipart(msg)
#--------------------------------------------------------------------------
# Kernel process management methods, from KernelManager:
#--------------------------------------------------------------------------
def _check_local(addr):
if isinstance(addr, tuple):
addr = addr[0]
return addr in LOCAL_IPS
def start_kernel(self, **kw):
"""Starts a kernel process and configures the manager to use it.
If random ports (port=0) are being used, this method must be called
before the channels are created.
Parameters:
-----------
ipython : bool, optional (default True)
Whether to use an IPython kernel instead of a plain Python kernel.
"""
self.kernel = Process(target=make_kernel, args=self.kernel_args,
kwargs=self.kernel_kwargs)
def shutdown_kernel(self, restart=False):
""" Attempts to the stop the kernel process cleanly. If the kernel
cannot be stopped, it is killed, if possible.
"""
# FIXME: Shutdown does not work on Windows due to ZMQ errors!
if sys.platform == 'win32':
self.kill_kernel()
return
# 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.xreq_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()
def restart_kernel(self, now=False):
"""Restarts a kernel with the same 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.
Parameters
----------
now : bool, optional
If True, the kernel is forcefully restarted *immediately*, without
having a chance to do any cleanup action. Otherwise the kernel is
given 1s to clean up before a forceful restart is issued.
In all cases the kernel is restarted, the only difference is whether
it is given a chance to perform a clean shutdown or not.
"""
if self._launch_args is None:
raise RuntimeError("Cannot restart the kernel. "
"No previous call to 'start_kernel'.")
else:
if self.has_kernel:
if now:
self.kill_kernel()
else:
self.shutdown_kernel(restart=True)
self.start_kernel(**self._launch_args)
# FIXME: Messages get dropped in Windows due to probable ZMQ bug
# unless there is some delay here.
if sys.platform == 'win32':
time.sleep(0.2)
@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. """
if self.has_kernel:
# Pause the heart beat channel if it exists.
if self._hb_channel is not None:
self._hb_channel.pause()
# Attempt to kill the kernel.
try:
self.kernel.kill()
except OSError, e:
# In Windows, we will get an Access Denied error if the process
# has already terminated. Ignore it.
if not (sys.platform == 'win32' and e.winerror == 5):
raise
self.kernel = None
else:
raise RuntimeError("Cannot kill kernel. No kernel is running!")
def interrupt_kernel(self):
""" Interrupts the kernel. Unlike ``signal_kernel``, this operation is
well supported on all platforms.
"""
if self.has_kernel:
if sys.platform == 'win32':
from parentpoller import ParentPollerWindows as Poller
Poller.send_interrupt(self.kernel.win32_interrupt_event)
else:
self.kernel.send_signal(signal.SIGINT)
else:
raise RuntimeError("Cannot interrupt kernel. No kernel is running!")
def signal_kernel(self, signum):
""" Sends a signal to the kernel. Note that since only SIGTERM is
supported on Windows, this function is only useful on Unix systems.
"""
if self.has_kernel:
self.kernel.send_signal(signum)
else:
raise RuntimeError("Cannot signal kernel. No kernel is running!")
@property
def is_alive(self):
"""Is the kernel process still running?"""
# FIXME: not using a heartbeat means this method is broken for any
# remote kernel, it's only capable of handling local kernels.
if self.has_kernel:
if self.kernel.poll() is None:
return True
else:
return False
else:
# We didn't start the kernel with this KernelManager so we don't
# know if it is running. We should use a heartbeat for this case.
return True
def make_starter(up_addr, down_addr, *args, **kwargs):
"""entry point function for launching a kernelstarter in a subprocess"""
loop = ioloop.IOLoop.instance()
ctx = zmq.Context()
session = Session()
upstream = zmqstream.ZMQStream(ctx.socket(zmq.XREQ),loop)
upstream.connect(up_addr)
downstream = zmqstream.ZMQStream(ctx.socket(zmq.XREQ),loop)
downstream.connect(down_addr)
starter = KernelStarter(session, upstream, downstream, *args, **kwargs)
starter.start()
loop.start()
Jump to Line
Something went wrong with that request. Please try again.