kernelbase.py
699 lines
| 25.2 KiB
| text/x-python
|
PythonLexer
|
r16857 | """Base class for a kernel that talks to frontends over 0MQ.""" | ||
|
r2598 | |||
|
r16566 | # Copyright (c) IPython Development Team. | ||
# Distributed under the terms of the Modified BSD License. | ||||
|
r2856 | from __future__ import print_function | ||
|
r2712 | |||
|
r2598 | import sys | ||
import time | ||||
|
r3294 | import logging | ||
|
r6788 | import uuid | ||
from datetime import datetime | ||||
|
r5601 | from signal import ( | ||
|
r11130 | signal, default_int_handler, SIGINT | ||
|
r5601 | ) | ||
|
r6789 | |||
|
r2598 | import zmq | ||
|
r6790 | from zmq.eventloop import ioloop | ||
from zmq.eventloop.zmqstream import ZMQStream | ||||
|
r2598 | |||
|
r17987 | from IPython.config.configurable import SingletonConfigurable | ||
|
r4952 | from IPython.core.error import StdinNotImplementedError | ||
|
r9094 | from IPython.core import release | ||
|
r4772 | from IPython.utils import py3compat | ||
|
r16857 | from IPython.utils.py3compat import unicode_type, string_types | ||
|
r2948 | from IPython.utils.jsonutil import json_clean | ||
|
r3970 | from IPython.utils.traitlets import ( | ||
|
r16857 | Any, Instance, Float, Dict, List, Set, Integer, Unicode, Bool, | ||
|
r3970 | ) | ||
|
r3979 | |||
|
r13347 | from .session import Session | ||
|
r2598 | |||
|
r3314 | |||
|
r17987 | class Kernel(SingletonConfigurable): | ||
|
r2759 | |||
|
r2778 | #--------------------------------------------------------------------------- | ||
# Kernel interface | ||||
#--------------------------------------------------------------------------- | ||||
|
r5153 | # attribute to override with a GUI | ||
eventloop = Any(None) | ||||
|
r6790 | def _eventloop_changed(self, name, old, new): | ||
"""schedule call to eventloop from IOLoop""" | ||||
loop = ioloop.IOLoop.instance() | ||||
|
r15232 | loop.add_callback(self.enter_eventloop) | ||
|
r5153 | |||
|
r2778 | session = Instance(Session) | ||
|
r5261 | profile_dir = Instance('IPython.core.profiledir.ProfileDir') | ||
|
r6790 | shell_streams = List() | ||
control_stream = Instance(ZMQStream) | ||||
|
r6804 | iopub_socket = Instance(zmq.Socket) | ||
|
r6790 | stdin_socket = Instance(zmq.Socket) | ||
|
r3970 | log = Instance(logging.Logger) | ||
|
r2759 | |||
|
r6789 | # identities: | ||
int_id = Integer(-1) | ||||
ident = Unicode() | ||||
def _ident_default(self): | ||||
|
r13353 | return unicode_type(uuid.uuid4()) | ||
|
r6789 | |||
|
r18468 | # This should be overridden by wrapper kernels that implement any real | ||
# language. | ||||
language_info = {} | ||||
|
r19732 | |||
# any links that should go in the help menu | ||||
help_links = List() | ||||
|
r18468 | |||
|
r2940 | # Private interface | ||
|
r6785 | |||
|
r13583 | _darwin_app_nap = Bool(True, config=True, | ||
help="""Whether to use appnope for compatiblity with OS X App Nap. | ||||
Only affects OS X >= 10.9. | ||||
""" | ||||
) | ||||
|
r16574 | # track associations with current request | ||
_allow_stdin = Bool(False) | ||||
_parent_header = Dict() | ||||
_parent_ident = Any(b'') | ||||
|
r2940 | # Time to sleep after flushing the stdout/err buffers in each execute | ||
# cycle. While this introduces a hard limit on the minimal latency of the | ||||
# execute cycle, it helps prevent output synchronization problems for | ||||
# clients. | ||||
# Units are in seconds. The minimum zmq latency on local host is probably | ||||
# ~150 microseconds, set this to 500us for now. We may need to increase it | ||||
# a little if it's not enough after more interactive testing. | ||||
_execute_sleep = Float(0.0005, config=True) | ||||
# Frequency of the kernel's event loop. | ||||
# Units are in seconds, kernel subclasses for GUI toolkits may need to | ||||
# adapt to milliseconds. | ||||
_poll_interval = Float(0.05, config=True) | ||||
|
r2972 | |||
# If the shutdown was requested over the network, we leave here the | ||||
# necessary reply message so it can be sent by our registered atexit | ||||
# handler. This ensures that the reply is only sent to clients truly at | ||||
# the end of our shutdown process (which happens after the underlying | ||||
# IPython shell's own shutdown). | ||||
_shutdown_message = None | ||||
|
r3019 | |||
# This is a dict of port number that the kernel is listening on. It is set | ||||
# by record_ports and used by connect_request. | ||||
|
r3970 | _recorded_ports = Dict() | ||
|
r8940 | |||
|
r6785 | # set of aborted msg_ids | ||
aborted = Set() | ||||
|
r3970 | |||
|
r16857 | # Track execution count here. For IPython, we override this to use the | ||
# execution count we store in the shell. | ||||
execution_count = 0 | ||||
|
r3294 | |||
|
r2759 | def __init__(self, **kwargs): | ||
|
r17095 | super(Kernel, self).__init__(**kwargs) | ||
|
r2786 | |||
|
r2598 | # Build dict of handlers for message types | ||
|
r4872 | msg_types = [ 'execute_request', 'complete_request', | ||
|
r16587 | 'inspect_request', 'history_request', | ||
|
r8879 | 'kernel_info_request', | ||
|
r6782 | 'connect_request', 'shutdown_request', | ||
|
r17624 | 'apply_request', 'is_complete_request', | ||
|
r6782 | ] | ||
|
r6790 | self.shell_handlers = {} | ||
|
r2612 | for msg_type in msg_types: | ||
|
r6790 | self.shell_handlers[msg_type] = getattr(self, msg_type) | ||
|
r6783 | |||
|
r6799 | control_msg_types = msg_types + [ 'clear_request', 'abort_request' ] | ||
|
r6782 | self.control_handlers = {} | ||
for msg_type in control_msg_types: | ||||
self.control_handlers[msg_type] = getattr(self, msg_type) | ||||
|
r17099 | |||
|
r6790 | def dispatch_control(self, msg): | ||
"""dispatch control requests""" | ||||
idents,msg = self.session.feed_identities(msg, copy=False) | ||||
try: | ||||
|
r18330 | msg = self.session.deserialize(msg, content=True, copy=False) | ||
|
r6790 | except: | ||
self.log.error("Invalid Control Message", exc_info=True) | ||||
return | ||||
|
r4237 | |||
|
r6790 | self.log.debug("Control received: %s", msg) | ||
|
r17117 | |||
# Set the parent message for side effects. | ||||
self.set_parent(idents, msg) | ||||
self._publish_status(u'busy') | ||||
|
r6790 | header = msg['header'] | ||
msg_type = header['msg_type'] | ||||
|
r4872 | |||
|
r6790 | handler = self.control_handlers.get(msg_type, None) | ||
if handler is None: | ||||
self.log.error("UNKNOWN CONTROL MESSAGE TYPE: %r", msg_type) | ||||
else: | ||||
|
r6792 | try: | ||
handler(self.control_stream, idents, msg) | ||||
except Exception: | ||||
self.log.error("Exception in control handler:", exc_info=True) | ||||
|
r17117 | |||
sys.stdout.flush() | ||||
sys.stderr.flush() | ||||
self._publish_status(u'idle') | ||||
|
r6790 | |||
def dispatch_shell(self, stream, msg): | ||||
"""dispatch shell requests""" | ||||
# flush control requests first | ||||
if self.control_stream: | ||||
self.control_stream.flush() | ||||
idents,msg = self.session.feed_identities(msg, copy=False) | ||||
try: | ||||
|
r18330 | msg = self.session.deserialize(msg, content=True, copy=False) | ||
|
r6790 | except: | ||
self.log.error("Invalid Message", exc_info=True) | ||||
return | ||||
|
r17099 | # Set the parent message for side effects. | ||
self.set_parent(idents, msg) | ||||
self._publish_status(u'busy') | ||||
|
r6790 | header = msg['header'] | ||
msg_id = header['msg_id'] | ||||
msg_type = msg['header']['msg_type'] | ||||
|
r2926 | # Print some info about this message and leave a '--->' marker, so it's | ||
# easier to trace visually the message chain when debugging. Each | ||||
# handler prints its message at the end. | ||||
|
r6783 | self.log.debug('\n*** MESSAGE TYPE:%s***', msg_type) | ||
self.log.debug(' Content: %s\n --->\n ', msg['content']) | ||||
|
r2926 | |||
|
r6785 | if msg_id in self.aborted: | ||
self.aborted.remove(msg_id) | ||||
# is it safe to assume a msg_id will not be resubmitted? | ||||
reply_type = msg_type.split('_')[0] + '_reply' | ||||
status = {'status' : 'aborted'} | ||||
|
r7957 | md = {'engine' : self.ident} | ||
md.update(status) | ||||
|
r16857 | self.session.send(stream, reply_type, metadata=md, | ||
|
r6785 | content=status, parent=msg, ident=idents) | ||
return | ||||
|
r6790 | handler = self.shell_handlers.get(msg_type, None) | ||
|
r2868 | if handler is None: | ||
|
r6790 | self.log.error("UNKNOWN MESSAGE TYPE: %r", msg_type) | ||
|
r2868 | else: | ||
|
r6790 | # ensure default_int_handler during handler call | ||
sig = signal(SIGINT, default_int_handler) | ||||
|
r16580 | self.log.debug("%s: %s", msg_type, msg) | ||
|
r6790 | try: | ||
handler(stream, idents, msg) | ||||
|
r6792 | except Exception: | ||
self.log.error("Exception in message handler:", exc_info=True) | ||||
|
r6790 | finally: | ||
signal(SIGINT, sig) | ||||
|
r17099 | |||
sys.stdout.flush() | ||||
sys.stderr.flush() | ||||
self._publish_status(u'idle') | ||||
|
r6790 | |||
def enter_eventloop(self): | ||||
"""enter eventloop""" | ||||
|
r15232 | self.log.info("entering eventloop %s", self.eventloop) | ||
for stream in self.shell_streams: | ||||
# flush any pending replies, | ||||
# which may be skipped by entering the eventloop | ||||
stream.flush(zmq.POLLOUT) | ||||
|
r6790 | # restore default_int_handler | ||
signal(SIGINT, default_int_handler) | ||||
|
r6827 | while self.eventloop is not None: | ||
try: | ||||
self.eventloop(self) | ||||
except KeyboardInterrupt: | ||||
# Ctrl-C shouldn't crash the kernel | ||||
self.log.error("KeyboardInterrupt caught in kernel") | ||||
continue | ||||
else: | ||||
# eventloop exited cleanly, this means we should stop (right?) | ||||
self.eventloop = None | ||||
break | ||||
|
r6882 | self.log.info("exiting eventloop") | ||
|
r4872 | |||
|
r6790 | def start(self): | ||
"""register dispatchers for streams""" | ||||
if self.control_stream: | ||||
self.control_stream.on_recv(self.dispatch_control, copy=False) | ||||
|
r2950 | |||
|
r6790 | def make_dispatcher(stream): | ||
def dispatcher(msg): | ||||
return self.dispatch_shell(stream, msg) | ||||
return dispatcher | ||||
|
r2778 | |||
|
r6790 | for s in self.shell_streams: | ||
s.on_recv(make_dispatcher(s), copy=False) | ||||
|
r10338 | |||
# publish idle status | ||||
self._publish_status('starting') | ||||
|
r6790 | |||
def do_one_iteration(self): | ||||
"""step eventloop just once""" | ||||
if self.control_stream: | ||||
self.control_stream.flush() | ||||
for stream in self.shell_streams: | ||||
# handle at most one request per iteration | ||||
stream.flush(zmq.POLLIN, 1) | ||||
stream.flush(zmq.POLLOUT) | ||||
|
r5153 | |||
|
r3019 | |||
|
r3970 | def record_ports(self, ports): | ||
|
r3019 | """Record the ports that this kernel is using. | ||
The creator of the Kernel instance must call this methods if they | ||||
want the :meth:`connect_request` method to return the port numbers. | ||||
""" | ||||
|
r3970 | self._recorded_ports = ports | ||
|
r3019 | |||
|
r2778 | #--------------------------------------------------------------------------- | ||
# Kernel request handlers | ||||
#--------------------------------------------------------------------------- | ||||
|
r6807 | |||
|
r7957 | def _make_metadata(self, other=None): | ||
"""init metadata dict, for execute/apply_reply""" | ||||
new_md = { | ||||
|
r6807 | 'dependencies_met' : True, | ||
'engine' : self.ident, | ||||
'started': datetime.now(), | ||||
} | ||||
|
r7957 | if other: | ||
new_md.update(other) | ||||
return new_md | ||||
|
r6807 | |||
|
r16567 | def _publish_execute_input(self, code, parent, execution_count): | ||
"""Publish the code request on the iopub stream.""" | ||||
|
r2926 | |||
|
r16567 | self.session.send(self.iopub_socket, u'execute_input', | ||
|
r6794 | {u'code':code, u'execution_count': execution_count}, | ||
|
r16567 | parent=parent, ident=self._topic('execute_input') | ||
|
r6794 | ) | ||
|
r7491 | |||
def _publish_status(self, status, parent=None): | ||||
"""send status (busy/idle) on IOPub""" | ||||
|
r6804 | self.session.send(self.iopub_socket, | ||
|
r5471 | u'status', | ||
|
r7491 | {u'execution_state': status}, | ||
|
r17099 | parent=parent or self._parent_header, | ||
|
r6795 | ident=self._topic('status'), | ||
|
r6794 | ) | ||
|
r16573 | |||
|
r16574 | def set_parent(self, ident, parent): | ||
|
r16665 | """Set the current parent_header | ||
|
r16574 | |||
|
r16665 | Side effects (IOPub messages) and replies are associated with | ||
the request that caused them via the parent_header. | ||||
The parent identity is used to route input_request messages | ||||
on the stdin channel. | ||||
|
r16574 | """ | ||
self._parent_ident = ident | ||||
self._parent_header = parent | ||||
|
r16857 | |||
def send_response(self, stream, msg_or_type, content=None, ident=None, | ||||
buffers=None, track=False, header=None, metadata=None): | ||||
"""Send a response to the message we're currently processing. | ||||
This accepts all the parameters of :meth:`IPython.kernel.zmq.session.Session.send` | ||||
except ``parent``. | ||||
This relies on :meth:`set_parent` having been called for the current | ||||
message. | ||||
""" | ||||
return self.session.send(stream, msg_or_type, content, self._parent_header, | ||||
ident, buffers, track, header, metadata) | ||||
|
r16574 | |||
|
r7491 | def execute_request(self, stream, ident, parent): | ||
"""handle an execute_request""" | ||||
|
r2598 | try: | ||
|
r2926 | content = parent[u'content'] | ||
|
r13893 | code = py3compat.cast_unicode_py2(content[u'code']) | ||
|
r4872 | silent = content[u'silent'] | ||
|
r7981 | store_history = content.get(u'store_history', not silent) | ||
|
r16857 | user_expressions = content.get('user_expressions', {}) | ||
allow_stdin = content.get('allow_stdin', False) | ||||
|
r2598 | except: | ||
|
r3970 | self.log.error("Got bad msg: ") | ||
|
r6783 | self.log.error("%s", parent) | ||
|
r2598 | return | ||
|
r6807 | |||
|
r7957 | md = self._make_metadata(parent['metadata']) | ||
|
r16574 | |||
|
r2926 | # Re-broadcast our input for the benefit of listening clients, and | ||
# start computing output | ||||
if not silent: | ||||
|
r16857 | self.execution_count += 1 | ||
self._publish_execute_input(code, parent, self.execution_count) | ||||
|
r8579 | |||
|
r16857 | reply_content = self.do_execute(code, silent, store_history, | ||
user_expressions, allow_stdin) | ||||
|
r2987 | |||
|
r2938 | # Flush output before sending the reply. | ||
sys.stdout.flush() | ||||
sys.stderr.flush() | ||||
# FIXME: on rare occasions, the flush doesn't seem to make it to the | ||||
# clients... This seems to mitigate the problem, but we definitely need | ||||
# to better understand what's going on. | ||||
|
r2940 | if self._execute_sleep: | ||
time.sleep(self._execute_sleep) | ||||
|
r4872 | |||
|
r3333 | # Send the reply. | ||
|
r4782 | reply_content = json_clean(reply_content) | ||
|
r6807 | |||
|
r7957 | md['status'] = reply_content['status'] | ||
|
r6807 | if reply_content['status'] == 'error' and \ | ||
reply_content['ename'] == 'UnmetDependency': | ||||
|
r7957 | md['dependencies_met'] = False | ||
|
r6807 | |||
|
r6790 | reply_msg = self.session.send(stream, u'execute_reply', | ||
|
r7957 | reply_content, parent, metadata=md, | ||
|
r6807 | ident=ident) | ||
|
r6783 | self.log.debug("%s", reply_msg) | ||
|
r3333 | |||
|
r6803 | if not silent and reply_msg['content']['status'] == u'error': | ||
self._abort_queues() | ||||
|
r2778 | |||
|
r16858 | def do_execute(self, code, silent, store_history=True, | ||
user_experssions=None, allow_stdin=False): | ||||
"""Execute user code. Must be overridden by subclasses. | ||||
""" | ||||
raise NotImplementedError | ||||
|
r6790 | def complete_request(self, stream, ident, parent): | ||
|
r16580 | content = parent['content'] | ||
code = content['code'] | ||||
cursor_pos = content['cursor_pos'] | ||||
|
r16857 | matches = self.do_complete(code, cursor_pos) | ||
|
r4782 | matches = json_clean(matches) | ||
|
r6790 | completion_msg = self.session.send(stream, 'complete_reply', | ||
|
r2778 | matches, parent, ident) | ||
|
r6783 | self.log.debug("%s", completion_msg) | ||
|
r2778 | |||
|
r16858 | def do_complete(self, code, cursor_pos): | ||
"""Override in subclasses to find completions. | ||||
""" | ||||
return {'matches' : [], | ||||
'cursor_end' : cursor_pos, | ||||
'cursor_start' : cursor_pos, | ||||
'metadata' : {}, | ||||
'status' : 'ok'} | ||||
|
r16587 | def inspect_request(self, stream, ident, parent): | ||
|
r6556 | content = parent['content'] | ||
|
r16580 | |||
|
r16857 | reply_content = self.do_inspect(content['code'], content['cursor_pos'], | ||
content.get('detail_level', 0)) | ||||
|
r3051 | # Before we send this object over, we scrub it for JSON usage | ||
|
r16580 | reply_content = json_clean(reply_content) | ||
|
r16587 | msg = self.session.send(stream, 'inspect_reply', | ||
|
r16580 | reply_content, parent, ident) | ||
|
r6783 | self.log.debug("%s", msg) | ||
|
r2795 | |||
|
r16858 | def do_inspect(self, code, cursor_pos, detail_level=0): | ||
"""Override in subclasses to allow introspection. | ||||
""" | ||||
return {'status': 'ok', 'data':{}, 'metadata':{}, 'found':False} | ||||
|
r6790 | def history_request(self, stream, ident, parent): | ||
|
r16857 | content = parent['content'] | ||
|
r4872 | |||
|
r16973 | reply_content = self.do_history(**content) | ||
|
r16857 | |||
reply_content = json_clean(reply_content) | ||||
|
r6790 | msg = self.session.send(stream, 'history_reply', | ||
|
r16857 | reply_content, parent, ident) | ||
self.log.debug("%s", msg) | ||||
|
r3019 | |||
|
r16858 | def do_history(self, hist_access_type, output, raw, session=None, start=None, | ||
stop=None, n=None, pattern=None, unique=False): | ||||
"""Override in subclasses to access history. | ||||
""" | ||||
return {'history': []} | ||||
|
r6790 | def connect_request(self, stream, ident, parent): | ||
|
r3019 | if self._recorded_ports is not None: | ||
content = self._recorded_ports.copy() | ||||
else: | ||||
content = {} | ||||
|
r6790 | msg = self.session.send(stream, 'connect_reply', | ||
|
r3019 | content, parent, ident) | ||
|
r6783 | self.log.debug("%s", msg) | ||
|
r3019 | |||
|
r16857 | @property | ||
def kernel_info(self): | ||||
return { | ||||
'protocol_version': release.kernel_protocol_version, | ||||
'implementation': self.implementation, | ||||
'implementation_version': self.implementation_version, | ||||
|
r18468 | 'language_info': self.language_info, | ||
|
r16857 | 'banner': self.banner, | ||
|
r19732 | 'help_links': self.help_links, | ||
|
r8830 | } | ||
|
r16857 | |||
def kernel_info_request(self, stream, ident, parent): | ||||
|
r8879 | msg = self.session.send(stream, 'kernel_info_reply', | ||
|
r16857 | self.kernel_info, parent, ident) | ||
|
r8830 | self.log.debug("%s", msg) | ||
|
r6790 | def shutdown_request(self, stream, ident, parent): | ||
|
r16857 | content = self.do_shutdown(parent['content']['restart']) | ||
|
r6799 | self.session.send(stream, u'shutdown_reply', content, parent, ident=ident) | ||
# same content, but different msg_id for broadcasting on IOPub | ||||
|
r5471 | self._shutdown_message = self.session.msg(u'shutdown_reply', | ||
|
r6799 | content, parent | ||
|
r6790 | ) | ||
self._at_shutdown() | ||||
# call sys.exit after a short delay | ||||
|
r6826 | loop = ioloop.IOLoop.instance() | ||
loop.add_timeout(time.time()+0.1, loop.stop) | ||||
|
r3019 | |||
|
r16858 | def do_shutdown(self, restart): | ||
"""Override in subclasses to do things when the frontend shuts down the | ||||
kernel. | ||||
""" | ||||
return {'status': 'ok', 'restart': restart} | ||||
|
r17624 | |||
def is_complete_request(self, stream, ident, parent): | ||||
content = parent['content'] | ||||
code = content['code'] | ||||
reply_content = self.do_is_complete(code) | ||||
reply_content = json_clean(reply_content) | ||||
reply_msg = self.session.send(stream, 'is_complete_reply', | ||||
reply_content, parent, ident) | ||||
self.log.debug("%s", reply_msg) | ||||
def do_is_complete(self, code): | ||||
"""Override in subclasses to find completions. | ||||
""" | ||||
|
r17804 | return {'status' : 'unknown', | ||
|
r17624 | } | ||
|
r16858 | |||
|
r2778 | #--------------------------------------------------------------------------- | ||
|
r6782 | # Engine methods | ||
#--------------------------------------------------------------------------- | ||||
|
r6790 | def apply_request(self, stream, ident, parent): | ||
|
r6782 | try: | ||
content = parent[u'content'] | ||||
bufs = parent[u'buffers'] | ||||
msg_id = parent['header']['msg_id'] | ||||
except: | ||||
|
r6783 | self.log.error("Got bad msg: %s", parent, exc_info=True) | ||
|
r6782 | return | ||
|
r6834 | |||
|
r7957 | md = self._make_metadata(parent['metadata']) | ||
|
r6782 | |||
|
r16857 | reply_content, result_buf = self.do_apply(content, bufs, msg_id, md) | ||
|
r6782 | |||
# put 'ok'/'error' status in header, for scheduler introspection: | ||||
|
r7957 | md['status'] = reply_content['status'] | ||
|
r6782 | |||
# flush i/o | ||||
sys.stdout.flush() | ||||
sys.stderr.flush() | ||||
|
r6783 | |||
|
r16857 | self.session.send(stream, u'apply_reply', reply_content, | ||
|
r7957 | parent=parent, ident=ident,buffers=result_buf, metadata=md) | ||
|
r6782 | |||
|
r16858 | def do_apply(self, content, bufs, msg_id, reply_metadata): | ||
"""Override in subclasses to support the IPython parallel framework. | ||||
""" | ||||
raise NotImplementedError | ||||
|
r6782 | #--------------------------------------------------------------------------- | ||
# Control messages | ||||
#--------------------------------------------------------------------------- | ||||
|
r6790 | def abort_request(self, stream, ident, parent): | ||
|
r17741 | """abort a specific msg by id""" | ||
|
r6782 | msg_ids = parent['content'].get('msg_ids', None) | ||
|
r13353 | if isinstance(msg_ids, string_types): | ||
|
r6782 | msg_ids = [msg_ids] | ||
if not msg_ids: | ||||
|
r17741 | self._abort_queues() | ||
|
r6782 | for mid in msg_ids: | ||
self.aborted.add(str(mid)) | ||||
content = dict(status='ok') | ||||
|
r6790 | reply_msg = self.session.send(stream, 'abort_reply', content=content, | ||
|
r6782 | parent=parent, ident=ident) | ||
|
r6783 | self.log.debug("%s", reply_msg) | ||
|
r6782 | |||
|
r6790 | def clear_request(self, stream, idents, parent): | ||
|
r6782 | """Clear our namespace.""" | ||
|
r16857 | content = self.do_clear() | ||
self.session.send(stream, 'clear_reply', ident=idents, parent=parent, | ||||
content = content) | ||||
|
r6782 | |||
|
r16858 | def do_clear(self): | ||
"""Override in subclasses to clear the namespace | ||||
This is only required for IPython.parallel. | ||||
""" | ||||
raise NotImplementedError | ||||
|
r6782 | #--------------------------------------------------------------------------- | ||
|
r2778 | # Protected interface | ||
#--------------------------------------------------------------------------- | ||||
|
r2598 | |||
|
r6795 | def _topic(self, topic): | ||
"""prefixed topic for IOPub messages""" | ||||
if self.int_id >= 0: | ||||
base = "engine.%i" % self.int_id | ||||
else: | ||||
base = "kernel.%s" % self.ident | ||||
return py3compat.cast_bytes("%s.%s" % (base, topic)) | ||||
|
r6785 | def _abort_queues(self): | ||
|
r6790 | for stream in self.shell_streams: | ||
if stream: | ||||
self._abort_queue(stream) | ||||
|
r6785 | |||
|
r6790 | def _abort_queue(self, stream): | ||
|
r6799 | poller = zmq.Poller() | ||
poller.register(stream.socket, zmq.POLLIN) | ||||
|
r2778 | while True: | ||
|
r6790 | idents,msg = self.session.recv(stream, zmq.NOBLOCK, content=True) | ||
|
r3269 | if msg is None: | ||
|
r6785 | return | ||
|
r3322 | |||
|
r6785 | self.log.info("Aborting:") | ||
self.log.info("%s", msg) | ||||
|
r4230 | msg_type = msg['header']['msg_type'] | ||
|
r2778 | reply_type = msg_type.split('_')[0] + '_reply' | ||
|
r6807 | |||
status = {'status' : 'aborted'} | ||||
|
r7957 | md = {'engine' : self.ident} | ||
md.update(status) | ||||
|
r7978 | reply_msg = self.session.send(stream, reply_type, metadata=md, | ||
|
r6807 | content=status, parent=msg, ident=idents) | ||
|
r6783 | self.log.debug("%s", reply_msg) | ||
|
r2778 | # We need to wait a bit for requests to come in. This can probably | ||
# be set shorter for true asynchronous clients. | ||||
|
r6799 | poller.poll(50) | ||
|
r6785 | |||
|
r2778 | |||
|
r4952 | def _no_raw_input(self): | ||
|
r5471 | """Raise StdinNotImplentedError if active frontend doesn't support | ||
stdin.""" | ||||
raise StdinNotImplementedError("raw_input was called, but this " | ||||
"frontend does not support stdin.") | ||||
|
r16573 | |||
|
r16574 | def getpass(self, prompt=''): | ||
"""Forward getpass to frontends | ||||
Raises | ||||
------ | ||||
StdinNotImplentedError if active frontend doesn't support stdin. | ||||
""" | ||||
if not self._allow_stdin: | ||||
raise StdinNotImplementedError( | ||||
"getpass was called, but this frontend does not support input requests." | ||||
) | ||||
return self._input_request(prompt, | ||||
self._parent_ident, | ||||
self._parent_header, | ||||
password=True, | ||||
) | ||||
def raw_input(self, prompt=''): | ||||
"""Forward raw_input to frontends | ||||
Raises | ||||
------ | ||||
StdinNotImplentedError if active frontend doesn't support stdin. | ||||
""" | ||||
if not self._allow_stdin: | ||||
raise StdinNotImplementedError( | ||||
"raw_input was called, but this frontend does not support input requests." | ||||
) | ||||
return self._input_request(prompt, | ||||
self._parent_ident, | ||||
self._parent_header, | ||||
password=False, | ||||
) | ||||
def _input_request(self, prompt, ident, parent, password=False): | ||||
|
r2730 | # Flush output before making the request. | ||
sys.stderr.flush() | ||||
sys.stdout.flush() | ||||
|
r10367 | # flush the stdin socket, to purge stale replies | ||
while True: | ||||
try: | ||||
self.stdin_socket.recv_multipart(zmq.NOBLOCK) | ||||
except zmq.ZMQError as e: | ||||
if e.errno == zmq.EAGAIN: | ||||
break | ||||
else: | ||||
raise | ||||
|
r2730 | # Send the input request. | ||
|
r16573 | content = json_clean(dict(prompt=prompt, password=password)) | ||
|
r5471 | self.session.send(self.stdin_socket, u'input_request', content, parent, | ||
ident=ident) | ||||
|
r2730 | |||
# Await a response. | ||||
|
r4521 | while True: | ||
try: | ||||
ident, reply = self.session.recv(self.stdin_socket, 0) | ||||
except Exception: | ||||
self.log.warn("Invalid Message:", exc_info=True) | ||||
|
r11706 | except KeyboardInterrupt: | ||
# re-raise KeyboardInterrupt, to truncate traceback | ||||
raise KeyboardInterrupt | ||||
|
r4521 | else: | ||
break | ||||
|
r2730 | try: | ||
|
r11154 | value = py3compat.unicode_to_str(reply['content']['value']) | ||
|
r2730 | except: | ||
|
r16574 | self.log.error("Bad input_reply: %s", parent) | ||
|
r2730 | value = '' | ||
|
r5622 | if value == '\x04': | ||
# EOF | ||||
raise EOFError | ||||
|
r2730 | return value | ||
|
r4872 | |||
|
r2972 | def _at_shutdown(self): | ||
"""Actions taken at shutdown by the kernel, called by python's atexit. | ||||
""" | ||||
# io.rprint("Kernel at_shutdown") # dbg | ||||
if self._shutdown_message is not None: | ||||
|
r6804 | self.session.send(self.iopub_socket, self._shutdown_message, ident=self._topic('shutdown')) | ||
|
r6783 | self.log.debug("%s", self._shutdown_message) | ||
|
r6804 | [ s.flush(zmq.POLLOUT) for s in self.shell_streams ] | ||