pykernel.py
278 lines
| 10.2 KiB
| text/x-python
|
PythonLexer
|
r2598 | #!/usr/bin/env python | ||
"""A simple interactive kernel that talks to a frontend over 0MQ. | ||||
Things to do: | ||||
* Implement `set_parent` logic. Right before doing exec, the Kernel should | ||||
call set_parent on all the PUB objects with the message about to be executed. | ||||
* Implement random port and security key logic. | ||||
* Implement control messages. | ||||
* Implement event loop and poll version. | ||||
""" | ||||
|
r2712 | #----------------------------------------------------------------------------- | ||
# Imports | ||||
#----------------------------------------------------------------------------- | ||||
|
r2641 | # Standard library imports. | ||
|
r2598 | import __builtin__ | ||
|
r2708 | from code import CommandCompiler | ||
|
r2598 | import sys | ||
import time | ||||
import traceback | ||||
|
r2641 | # System library imports. | ||
|
r2598 | import zmq | ||
|
r2641 | # Local imports. | ||
|
r3970 | from IPython.utils.traitlets import HasTraits, Instance, Dict, Float | ||
|
r2598 | from completer import KernelCompleter | ||
|
r3970 | from entry_point import base_launch_kernel | ||
|
r2778 | from session import Session, Message | ||
|
r3970 | from kernelapp import KernelApp | ||
|
r2598 | |||
|
r2712 | #----------------------------------------------------------------------------- | ||
|
r2755 | # Main kernel class | ||
|
r2712 | #----------------------------------------------------------------------------- | ||
|
r2641 | |||
|
r2778 | class Kernel(HasTraits): | ||
|
r2756 | |||
|
r3028 | # Private interface | ||
|
r3935 | # 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) | ||||
|
r3028 | # 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() | ||
|
r3028 | |||
|
r2756 | #--------------------------------------------------------------------------- | ||
# Kernel interface | ||||
#--------------------------------------------------------------------------- | ||||
|
r2778 | session = Instance(Session) | ||
|
r3970 | shell_socket = Instance('zmq.Socket') | ||
iopub_socket = Instance('zmq.Socket') | ||||
stdin_socket = Instance('zmq.Socket') | ||||
|
r3981 | log = Instance('logging.Logger') | ||
|
r2778 | |||
def __init__(self, **kwargs): | ||||
super(Kernel, self).__init__(**kwargs) | ||||
|
r2598 | self.user_ns = {} | ||
self.history = [] | ||||
self.compiler = CommandCompiler() | ||||
self.completer = KernelCompleter(self.user_ns) | ||||
|
r2756 | |||
|
r2598 | # Build dict of handlers for message types | ||
|
r2612 | msg_types = [ 'execute_request', 'complete_request', | ||
|
r3088 | 'object_info_request', 'shutdown_request' ] | ||
|
r2598 | self.handlers = {} | ||
|
r2612 | for msg_type in msg_types: | ||
|
r2598 | self.handlers[msg_type] = getattr(self, msg_type) | ||
|
r2756 | def start(self): | ||
""" Start the kernel main loop. | ||||
""" | ||||
|
r2598 | while True: | ||
|
r3970 | ident,msg = self.session.recv(self.shell_socket,0) | ||
|
r3269 | assert ident is not None, "Missing message part." | ||
|
r2756 | omsg = Message(msg) | ||
|
r3981 | self.log.debug(str(omsg)) | ||
|
r2756 | handler = self.handlers.get(omsg.msg_type, None) | ||
if handler is None: | ||||
|
r3981 | self.log.error("UNKNOWN MESSAGE TYPE: %s"%omsg) | ||
|
r2598 | else: | ||
|
r2756 | handler(ident, omsg) | ||
|
r3970 | def record_ports(self, ports): | ||
|
r3028 | """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 | ||
|
r3028 | |||
|
r2756 | #--------------------------------------------------------------------------- | ||
# Kernel request handlers | ||||
#--------------------------------------------------------------------------- | ||||
|
r2598 | |||
def execute_request(self, ident, parent): | ||||
try: | ||||
code = parent[u'content'][u'code'] | ||||
except: | ||||
|
r3981 | self.log.error("Got bad msg: %s"%Message(parent)) | ||
|
r2598 | return | ||
|
r3970 | pyin_msg = self.session.send(self.iopub_socket, u'pyin',{u'code':code}, parent=parent) | ||
|
r2722 | |||
|
r2598 | try: | ||
comp_code = self.compiler(code, '<zmq-kernel>') | ||||
|
r2730 | |||
# Replace raw_input. Note that is not sufficient to replace | ||||
# raw_input in the user namespace. | ||||
|
r2756 | raw_input = lambda prompt='': self._raw_input(prompt, ident, parent) | ||
|
r2730 | __builtin__.raw_input = raw_input | ||
|
r2824 | # Set the parent message of the display hook and out streams. | ||
|
r2598 | sys.displayhook.set_parent(parent) | ||
|
r2824 | sys.stdout.set_parent(parent) | ||
sys.stderr.set_parent(parent) | ||||
|
r2730 | |||
|
r2598 | exec comp_code in self.user_ns, self.user_ns | ||
except: | ||||
etype, evalue, tb = sys.exc_info() | ||||
tb = traceback.format_exception(etype, evalue, tb) | ||||
exc_content = { | ||||
u'status' : u'error', | ||||
u'traceback' : tb, | ||||
|
r2716 | u'ename' : unicode(etype.__name__), | ||
|
r2598 | u'evalue' : unicode(evalue) | ||
} | ||||
|
r3970 | exc_msg = self.session.send(self.iopub_socket, u'pyerr', exc_content, parent) | ||
|
r2598 | reply_content = exc_content | ||
else: | ||||
|
r2778 | reply_content = { 'status' : 'ok', 'payload' : {} } | ||
|
r2722 | |||
# Flush output before sending the reply. | ||||
sys.stderr.flush() | ||||
sys.stdout.flush() | ||||
|
r3935 | # 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. | ||||
if self._execute_sleep: | ||||
time.sleep(self._execute_sleep) | ||||
|
r2722 | |||
# Send the reply. | ||||
|
r3970 | reply_msg = self.session.send(self.shell_socket, u'execute_reply', reply_content, parent, ident=ident) | ||
|
r3981 | self.log.debug(Message(reply_msg)) | ||
|
r2598 | if reply_msg['content']['status'] == u'error': | ||
|
r2756 | self._abort_queue() | ||
def complete_request(self, ident, parent): | ||||
|
r2867 | matches = {'matches' : self._complete(parent), | ||
|
r2778 | 'status' : 'ok'} | ||
|
r3970 | completion_msg = self.session.send(self.shell_socket, 'complete_reply', | ||
|
r2756 | matches, parent, ident) | ||
|
r3981 | self.log.debug(completion_msg) | ||
|
r2756 | |||
def object_info_request(self, ident, parent): | ||||
context = parent['content']['oname'].split('.') | ||||
object_info = self._object_info(context) | ||||
|
r3970 | msg = self.session.send(self.shell_socket, 'object_info_reply', | ||
|
r2756 | object_info, parent, ident) | ||
|
r3981 | self.log.debug(msg) | ||
|
r2756 | |||
|
r3088 | def shutdown_request(self, ident, parent): | ||
content = dict(parent['content']) | ||||
|
r3970 | msg = self.session.send(self.shell_socket, 'shutdown_reply', | ||
|
r3088 | content, parent, ident) | ||
|
r3970 | msg = self.session.send(self.iopub_socket, 'shutdown_reply', | ||
|
r3089 | content, parent, ident) | ||
|
r3981 | self.log.debug(msg) | ||
|
r3088 | time.sleep(0.1) | ||
sys.exit(0) | ||||
|
r2756 | #--------------------------------------------------------------------------- | ||
# Protected interface | ||||
#--------------------------------------------------------------------------- | ||||
def _abort_queue(self): | ||||
while True: | ||||
|
r3981 | ident,msg = self.session.recv(self.shell_socket, zmq.NOBLOCK) | ||
|
r3930 | if msg is None: | ||
|
r3981 | # msg=None on EAGAIN | ||
|
r3930 | break | ||
|
r2756 | else: | ||
|
r3981 | assert ident is not None, "Missing message part." | ||
self.log.debug("Aborting: %s"%Message(msg)) | ||||
|
r4230 | msg_type = msg['header']['msg_type'] | ||
|
r2756 | reply_type = msg_type.split('_')[0] + '_reply' | ||
|
r3970 | reply_msg = self.session.send(self.shell_socket, reply_type, {'status':'aborted'}, msg, ident=ident) | ||
|
r3981 | self.log.debug(Message(reply_msg)) | ||
|
r2756 | # We need to wait a bit for requests to come in. This can probably | ||
# be set shorter for true asynchronous clients. | ||||
time.sleep(0.1) | ||||
|
r2598 | |||
|
r2756 | def _raw_input(self, prompt, ident, parent): | ||
|
r2730 | # Flush output before making the request. | ||
sys.stderr.flush() | ||||
sys.stdout.flush() | ||||
# Send the input request. | ||||
content = dict(prompt=prompt) | ||||
|
r3970 | msg = self.session.send(self.stdin_socket, u'input_request', content, parent) | ||
|
r2730 | |||
# Await a response. | ||||
|
r3970 | ident,reply = self.session.recv(self.stdin_socket, 0) | ||
|
r2730 | try: | ||
value = reply['content']['value'] | ||||
except: | ||||
|
r3981 | self.log.error("Got bad raw_input reply: %s"%Message(parent)) | ||
|
r2730 | value = '' | ||
return value | ||||
|
r2778 | def _complete(self, msg): | ||
return self.completer.complete(msg.content.line, msg.content.text) | ||||
|
r2756 | def _object_info(self, context): | ||
symbol, leftover = self._symbol_from_context(context) | ||||
|
r2612 | if symbol is not None and not leftover: | ||
doc = getattr(symbol, '__doc__', '') | ||||
else: | ||||
doc = '' | ||||
object_info = dict(docstring = doc) | ||||
return object_info | ||||
|
r2756 | def _symbol_from_context(self, context): | ||
|
r2612 | if not context: | ||
return None, context | ||||
base_symbol_string = context[0] | ||||
symbol = self.user_ns.get(base_symbol_string, None) | ||||
if symbol is None: | ||||
symbol = __builtin__.__dict__.get(base_symbol_string, None) | ||||
if symbol is None: | ||||
return None, context | ||||
context = context[1:] | ||||
for i, name in enumerate(context): | ||||
new_symbol = getattr(symbol, name, None) | ||||
if new_symbol is None: | ||||
return symbol, context[i:] | ||||
else: | ||||
symbol = new_symbol | ||||
return symbol, [] | ||||
|
r2712 | #----------------------------------------------------------------------------- | ||
# Kernel main and launch functions | ||||
#----------------------------------------------------------------------------- | ||||
|
r3970 | def launch_kernel(*args, **kwargs): | ||
""" Launches a simple Python kernel, binding to the specified ports. | ||||
|
r3144 | |||
|
r3970 | This function simply calls entry_point.base_launch_kernel with the right first | ||
command to start a pykernel. See base_launch_kernel for arguments. | ||||
|
r2700 | |||
Returns | ||||
------- | ||||
A tuple of form: | ||||
|
r3970 | (kernel_process, xrep_port, pub_port, req_port, hb_port) | ||
|
r2702 | where kernel_process is a Popen object and the ports are integers. | ||
|
r2641 | """ | ||
|
r2778 | return base_launch_kernel('from IPython.zmq.pykernel import main; main()', | ||
|
r3970 | *args, **kwargs) | ||
|
r2700 | |||
|
r3970 | def main(): | ||
"""Run a PyKernel as an application""" | ||||
|
r3980 | app = KernelApp.instance() | ||
|
r3970 | app.initialize() | ||
app.start() | ||||
|
r2598 | |||
if __name__ == '__main__': | ||||
|
r2667 | main() | ||