make parent_header available from the Shell object

IPython/kernel/zmq/ipkernel.py

@@ -1,798 +1,789 b''
 #!/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.
-"""
+"""An interactive kernel that talks to frontends over 0MQ."""
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 from __future__ import print_function
 
 # Standard library imports
 import __builtin__
 import sys
 import time
 import traceback
 import logging
 import uuid
 
 from datetime import datetime
 from signal import (
         signal, default_int_handler, SIGINT
 )
 
 # System library imports
 import zmq
 from zmq.eventloop import ioloop
 from zmq.eventloop.zmqstream import ZMQStream
 
 # Local imports
 from IPython.config.configurable import Configurable
 from IPython.core.error import StdinNotImplementedError
 from IPython.core import release
 from IPython.utils import py3compat
 from IPython.utils.jsonutil import json_clean
 from IPython.utils.traitlets import (
     Any, Instance, Float, Dict, List, Set, Integer, Unicode,
     Type
 )
 
 from serialize import serialize_object, unpack_apply_message
 from session import Session
 from zmqshell import ZMQInteractiveShell
 
 
 #-----------------------------------------------------------------------------
 # Main kernel class
 #-----------------------------------------------------------------------------
 
 protocol_version = list(release.kernel_protocol_version_info)
 ipython_version = list(release.version_info)
 language_version = list(sys.version_info[:3])
 
 
 class Kernel(Configurable):
 
     #---------------------------------------------------------------------------
     # Kernel interface
     #---------------------------------------------------------------------------
 
     # attribute to override with a GUI
     eventloop = Any(None)
     def _eventloop_changed(self, name, old, new):
         """schedule call to eventloop from IOLoop"""
         loop = ioloop.IOLoop.instance()
         loop.add_timeout(time.time()+0.1, self.enter_eventloop)
 
     shell = Instance('IPython.core.interactiveshell.InteractiveShellABC')
     shell_class = Type(ZMQInteractiveShell)
 
     session = Instance(Session)
     profile_dir = Instance('IPython.core.profiledir.ProfileDir')
     shell_streams = List()
     control_stream = Instance(ZMQStream)
     iopub_socket = Instance(zmq.Socket)
     stdin_socket = Instance(zmq.Socket)
     log = Instance(logging.Logger)
     
     user_module = Any()
     def _user_module_changed(self, name, old, new):
         if self.shell is not None:
             self.shell.user_module = new
     
     user_ns = Instance(dict, args=None, allow_none=True)
     def _user_ns_changed(self, name, old, new):
         if self.shell is not None:
             self.shell.user_ns = new
             self.shell.init_user_ns()
 
     # identities:
     int_id = Integer(-1)
     ident = Unicode()
 
     def _ident_default(self):
         return unicode(uuid.uuid4())
 
 
     # Private interface
     
     # 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)
 
     # 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
 
     # This is a dict of port number that the kernel is listening on. It is set
     # by record_ports and used by connect_request.
     _recorded_ports = Dict()
 
     # A reference to the Python builtin 'raw_input' function.
     # (i.e., __builtin__.raw_input for Python 2.7, builtins.input for Python 3)
     _sys_raw_input = Any()
     _sys_eval_input = Any()
 
     # set of aborted msg_ids
     aborted = Set()
 
 
     def __init__(self, **kwargs):
         super(Kernel, self).__init__(**kwargs)
 
         # Initialize the InteractiveShell subclass
         self.shell = self.shell_class.instance(parent=self,
             profile_dir = self.profile_dir,
             user_module = self.user_module,
             user_ns     = self.user_ns,
             kernel      = self,
         )
         self.shell.displayhook.session = self.session
         self.shell.displayhook.pub_socket = self.iopub_socket
         self.shell.displayhook.topic = self._topic('pyout')
         self.shell.display_pub.session = self.session
         self.shell.display_pub.pub_socket = self.iopub_socket
         self.shell.data_pub.session = self.session
         self.shell.data_pub.pub_socket = self.iopub_socket
 
         # TMP - hack while developing
         self.shell._reply_content = None
 
         # Build dict of handlers for message types
         msg_types = [ 'execute_request', 'complete_request',
                       'object_info_request', 'history_request',
                       'kernel_info_request',
                       'connect_request', 'shutdown_request',
                       'apply_request',
                     ]
         self.shell_handlers = {}
         for msg_type in msg_types:
             self.shell_handlers[msg_type] = getattr(self, msg_type)
         
         comm_msg_types = [ 'comm_open', 'comm_msg', 'comm_close' ]
         comm_manager = self.shell.comm_manager
         for msg_type in comm_msg_types:
             self.shell_handlers[msg_type] = getattr(comm_manager, msg_type)
         
         control_msg_types = msg_types + [ 'clear_request', 'abort_request' ]
         self.control_handlers = {}
         for msg_type in control_msg_types:
             self.control_handlers[msg_type] = getattr(self, msg_type)
 
 
     def dispatch_control(self, msg):
         """dispatch control requests"""
         idents,msg = self.session.feed_identities(msg, copy=False)
         try:
             msg = self.session.unserialize(msg, content=True, copy=False)
         except:
             self.log.error("Invalid Control Message", exc_info=True)
             return
 
         self.log.debug("Control received: %s", msg)
 
         header = msg['header']
         msg_id = header['msg_id']
         msg_type = header['msg_type']
 
         handler = self.control_handlers.get(msg_type, None)
         if handler is None:
             self.log.error("UNKNOWN CONTROL MESSAGE TYPE: %r", msg_type)
         else:
             try:
                 handler(self.control_stream, idents, msg)
             except Exception:
                 self.log.error("Exception in control handler:", exc_info=True)
     
     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:
             msg = self.session.unserialize(msg, content=True, copy=False)
         except:
             self.log.error("Invalid Message", exc_info=True)
             return
 
         header = msg['header']
         msg_id = header['msg_id']
         msg_type = msg['header']['msg_type']
         
         # 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.
         self.log.debug('\n*** MESSAGE TYPE:%s***', msg_type)
         self.log.debug('   Content: %s\n   --->\n   ', msg['content'])
 
         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'}
             md = {'engine' : self.ident}
             md.update(status)
             reply_msg = self.session.send(stream, reply_type, metadata=md,
                         content=status, parent=msg, ident=idents)
             return
         
         handler = self.shell_handlers.get(msg_type, None)
         if handler is None:
             self.log.error("UNKNOWN MESSAGE TYPE: %r", msg_type)
         else:
             # ensure default_int_handler during handler call
             sig = signal(SIGINT, default_int_handler)
             try:
                 handler(stream, idents, msg)
             except Exception:
                 self.log.error("Exception in message handler:", exc_info=True)
             finally:
                 signal(SIGINT, sig)
     
     def enter_eventloop(self):
         """enter eventloop"""
         self.log.info("entering eventloop")
         # restore default_int_handler
         signal(SIGINT, default_int_handler)
         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
         self.log.info("exiting eventloop")
 
     def start(self):
         """register dispatchers for streams"""
         self.shell.exit_now = False
         if self.control_stream:
             self.control_stream.on_recv(self.dispatch_control, copy=False)
 
         def make_dispatcher(stream):
             def dispatcher(msg):
                 return self.dispatch_shell(stream, msg)
             return dispatcher
 
         for s in self.shell_streams:
             s.on_recv(make_dispatcher(s), copy=False)
 
         # publish idle status
         self._publish_status('starting')
     
     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)
 
 
     def record_ports(self, ports):
         """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.
         """
         self._recorded_ports = ports
 
     #---------------------------------------------------------------------------
     # Kernel request handlers
     #---------------------------------------------------------------------------
     
     def _make_metadata(self, other=None):
         """init metadata dict, for execute/apply_reply"""
         new_md = {
             'dependencies_met' : True,
             'engine' : self.ident,
             'started': datetime.now(),
         }
         if other:
             new_md.update(other)
         return new_md
     
     def _publish_pyin(self, code, parent, execution_count):
         """Publish the code request on the pyin stream."""
 
         self.session.send(self.iopub_socket, u'pyin',
                             {u'code':code, u'execution_count': execution_count},
                             parent=parent, ident=self._topic('pyin')
         )
     
     def _publish_status(self, status, parent=None):
         """send status (busy/idle) on IOPub"""
         self.session.send(self.iopub_socket,
                           u'status',
                           {u'execution_state': status},
                           parent=parent,
                           ident=self._topic('status'),
                           )
         
 
     def execute_request(self, stream, ident, parent):
         """handle an execute_request"""
         
         self._publish_status(u'busy', parent)
         
         try:
             content = parent[u'content']
             code = content[u'code']
             silent = content[u'silent']
             store_history = content.get(u'store_history', not silent)
         except:
             self.log.error("Got bad msg: ")
             self.log.error("%s", parent)
             return
         
         md = self._make_metadata(parent['metadata'])
 
         shell = self.shell # we'll need this a lot here
 
         # Replace raw_input. Note that is not sufficient to replace
         # raw_input in the user namespace.
         if content.get('allow_stdin', False):
             raw_input = lambda prompt='': self._raw_input(prompt, ident, parent)
             input = lambda prompt='': eval(raw_input(prompt))
         else:
             raw_input = input = lambda prompt='' : self._no_raw_input()
 
         if py3compat.PY3:
             self._sys_raw_input = __builtin__.input
             __builtin__.input = raw_input
         else:
             self._sys_raw_input = __builtin__.raw_input
             self._sys_eval_input = __builtin__.input
             __builtin__.raw_input = raw_input
             __builtin__.input = input
 
         # Set the parent message of the display hook and out streams.
         shell.set_parent(parent)
 
         # Re-broadcast our input for the benefit of listening clients, and
         # start computing output
         if not silent:
             self._publish_pyin(code, parent, shell.execution_count)
 
         reply_content = {}
         try:
             # FIXME: the shell calls the exception handler itself.
             shell.run_cell(code, store_history=store_history, silent=silent)
         except:
             status = u'error'
             # FIXME: this code right now isn't being used yet by default,
             # because the run_cell() call above directly fires off exception
             # reporting.  This code, therefore, is only active in the scenario
             # where runlines itself has an unhandled exception.  We need to
             # uniformize this, for all exception construction to come from a
             # single location in the codbase.
             etype, evalue, tb = sys.exc_info()
             tb_list = traceback.format_exception(etype, evalue, tb)
             reply_content.update(shell._showtraceback(etype, evalue, tb_list))
         else:
             status = u'ok'
         finally:
             # Restore raw_input.
              if py3compat.PY3:
                  __builtin__.input = self._sys_raw_input
              else:
                  __builtin__.raw_input = self._sys_raw_input
                  __builtin__.input = self._sys_eval_input
 
         reply_content[u'status'] = status
 
         # Return the execution counter so clients can display prompts
         reply_content['execution_count'] = shell.execution_count - 1
 
         # FIXME - fish exception info out of shell, possibly left there by
         # runlines.  We'll need to clean up this logic later.
         if shell._reply_content is not None:
             reply_content.update(shell._reply_content)
             e_info = dict(engine_uuid=self.ident, engine_id=self.int_id, method='execute')
             reply_content['engine_info'] = e_info
             # reset after use
             shell._reply_content = None
         
         if 'traceback' in reply_content:
             self.log.info("Exception in execute request:\n%s", '\n'.join(reply_content['traceback']))
         
 
         # At this point, we can tell whether the main code execution succeeded
         # or not.  If it did, we proceed to evaluate user_variables/expressions
         if reply_content['status'] == 'ok':
             reply_content[u'user_variables'] = \
                          shell.user_variables(content.get(u'user_variables', []))
             reply_content[u'user_expressions'] = \
                          shell.user_expressions(content.get(u'user_expressions', {}))
         else:
             # If there was an error, don't even try to compute variables or
             # expressions
             reply_content[u'user_variables'] = {}
             reply_content[u'user_expressions'] = {}
 
         # Payloads should be retrieved regardless of outcome, so we can both
         # recover partial output (that could have been generated early in a
         # block, before an error) and clear the payload system always.
         reply_content[u'payload'] = shell.payload_manager.read_payload()
         # Be agressive about clearing the payload because we don't want
         # it to sit in memory until the next execute_request comes in.
         shell.payload_manager.clear_payload()
 
         # 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.
         if self._execute_sleep:
             time.sleep(self._execute_sleep)
 
         # Send the reply.
         reply_content = json_clean(reply_content)
         
         md['status'] = reply_content['status']
         if reply_content['status'] == 'error' and \
                         reply_content['ename'] == 'UnmetDependency':
                 md['dependencies_met'] = False
 
         reply_msg = self.session.send(stream, u'execute_reply',
                                       reply_content, parent, metadata=md,
                                       ident=ident)
         
         self.log.debug("%s", reply_msg)
 
         if not silent and reply_msg['content']['status'] == u'error':
             self._abort_queues()
 
         self._publish_status(u'idle', parent)
 
     def complete_request(self, stream, ident, parent):
         txt, matches = self._complete(parent)
         matches = {'matches' : matches,
                    'matched_text' : txt,
                    'status' : 'ok'}
         matches = json_clean(matches)
         completion_msg = self.session.send(stream, 'complete_reply',
                                            matches, parent, ident)
         self.log.debug("%s", completion_msg)
 
     def object_info_request(self, stream, ident, parent):
         content = parent['content']
         object_info = self.shell.object_inspect(content['oname'],
                         detail_level = content.get('detail_level', 0)
         )
         # Before we send this object over, we scrub it for JSON usage
         oinfo = json_clean(object_info)
         msg = self.session.send(stream, 'object_info_reply',
                                 oinfo, parent, ident)
         self.log.debug("%s", msg)
 
     def history_request(self, stream, ident, parent):
         # We need to pull these out, as passing **kwargs doesn't work with
         # unicode keys before Python 2.6.5.
         hist_access_type = parent['content']['hist_access_type']
         raw = parent['content']['raw']
         output = parent['content']['output']
         if hist_access_type == 'tail':
             n = parent['content']['n']
             hist = self.shell.history_manager.get_tail(n, raw=raw, output=output,
                                                             include_latest=True)
 
         elif hist_access_type == 'range':
             session = parent['content']['session']
             start = parent['content']['start']
             stop = parent['content']['stop']
             hist = self.shell.history_manager.get_range(session, start, stop,
                                                         raw=raw, output=output)
 
         elif hist_access_type == 'search':
             n = parent['content'].get('n')
             unique = parent['content'].get('unique', False)
             pattern = parent['content']['pattern']
             hist = self.shell.history_manager.search(
                 pattern, raw=raw, output=output, n=n, unique=unique)
 
         else:
             hist = []
         hist = list(hist)
         content = {'history' : hist}
         content = json_clean(content)
         msg = self.session.send(stream, 'history_reply',
                                 content, parent, ident)
         self.log.debug("Sending history reply with %i entries", len(hist))
 
     def connect_request(self, stream, ident, parent):
         if self._recorded_ports is not None:
             content = self._recorded_ports.copy()
         else:
             content = {}
         msg = self.session.send(stream, 'connect_reply',
                                 content, parent, ident)
         self.log.debug("%s", msg)
 
     def kernel_info_request(self, stream, ident, parent):
         vinfo = {
             'protocol_version': protocol_version,
             'ipython_version': ipython_version,
             'language_version': language_version,
             'language': 'python',
         }
         msg = self.session.send(stream, 'kernel_info_reply',
                                 vinfo, parent, ident)
         self.log.debug("%s", msg)
 
     def shutdown_request(self, stream, ident, parent):
         self.shell.exit_now = True
         content = dict(status='ok')
         content.update(parent['content'])
         self.session.send(stream, u'shutdown_reply', content, parent, ident=ident)
         # same content, but different msg_id for broadcasting on IOPub
         self._shutdown_message = self.session.msg(u'shutdown_reply',
                                                   content, parent
         )
 
         self._at_shutdown()
         # call sys.exit after a short delay
         loop = ioloop.IOLoop.instance()
         loop.add_timeout(time.time()+0.1, loop.stop)
 
     #---------------------------------------------------------------------------
     # Engine methods
     #---------------------------------------------------------------------------
 
     def apply_request(self, stream, ident, parent):
         try:
             content = parent[u'content']
             bufs = parent[u'buffers']
             msg_id = parent['header']['msg_id']
         except:
             self.log.error("Got bad msg: %s", parent, exc_info=True)
             return
 
         self._publish_status(u'busy', parent)
 
         # Set the parent message of the display hook and out streams.
         shell = self.shell
         shell.set_parent(parent)
 
         # pyin_msg = self.session.msg(u'pyin',{u'code':code}, parent=parent)
         # self.iopub_socket.send(pyin_msg)
         # self.session.send(self.iopub_socket, u'pyin', {u'code':code},parent=parent)
         md = self._make_metadata(parent['metadata'])
         try:
             working = shell.user_ns
 
             prefix = "_"+str(msg_id).replace("-","")+"_"
 
             f,args,kwargs = unpack_apply_message(bufs, working, copy=False)
 
             fname = getattr(f, '__name__', 'f')
 
             fname = prefix+"f"
             argname = prefix+"args"
             kwargname = prefix+"kwargs"
             resultname = prefix+"result"
 
             ns = { fname : f, argname : args, kwargname : kwargs , resultname : None }
             # print ns
             working.update(ns)
             code = "%s = %s(*%s,**%s)" % (resultname, fname, argname, kwargname)
             try:
                 exec code in shell.user_global_ns, shell.user_ns
                 result = working.get(resultname)
             finally:
                 for key in ns.iterkeys():
                     working.pop(key)
 
             result_buf = serialize_object(result,
                 buffer_threshold=self.session.buffer_threshold,
                 item_threshold=self.session.item_threshold,
             )
         
         except:
             # invoke IPython traceback formatting
             shell.showtraceback()
             # FIXME - fish exception info out of shell, possibly left there by
             # run_code.  We'll need to clean up this logic later.
             reply_content = {}
             if shell._reply_content is not None:
                 reply_content.update(shell._reply_content)
                 e_info = dict(engine_uuid=self.ident, engine_id=self.int_id, method='apply')
                 reply_content['engine_info'] = e_info
                 # reset after use
                 shell._reply_content = None
             
             self.session.send(self.iopub_socket, u'pyerr', reply_content, parent=parent,
                                 ident=self._topic('pyerr'))
             self.log.info("Exception in apply request:\n%s", '\n'.join(reply_content['traceback']))
             result_buf = []
 
             if reply_content['ename'] == 'UnmetDependency':
                 md['dependencies_met'] = False
         else:
             reply_content = {'status' : 'ok'}
 
         # put 'ok'/'error' status in header, for scheduler introspection:
         md['status'] = reply_content['status']
 
         # flush i/o
         sys.stdout.flush()
         sys.stderr.flush()
         
         reply_msg = self.session.send(stream, u'apply_reply', reply_content,
                     parent=parent, ident=ident,buffers=result_buf, metadata=md)
 
         self._publish_status(u'idle', parent)
 
     #---------------------------------------------------------------------------
     # Control messages
     #---------------------------------------------------------------------------
 
     def abort_request(self, stream, ident, parent):
         """abort a specifig msg by id"""
         msg_ids = parent['content'].get('msg_ids', None)
         if isinstance(msg_ids, basestring):
             msg_ids = [msg_ids]
         if not msg_ids:
             self.abort_queues()
         for mid in msg_ids:
             self.aborted.add(str(mid))
 
         content = dict(status='ok')
         reply_msg = self.session.send(stream, 'abort_reply', content=content,
                 parent=parent, ident=ident)
         self.log.debug("%s", reply_msg)
 
     def clear_request(self, stream, idents, parent):
         """Clear our namespace."""
         self.shell.reset(False)
         msg = self.session.send(stream, 'clear_reply', ident=idents, parent=parent,
                 content = dict(status='ok'))
 
 
     #---------------------------------------------------------------------------
     # Protected interface
     #---------------------------------------------------------------------------
 
     def _wrap_exception(self, method=None):
         # import here, because _wrap_exception is only used in parallel,
         # and parallel has higher min pyzmq version
         from IPython.parallel.error import wrap_exception
         e_info = dict(engine_uuid=self.ident, engine_id=self.int_id, method=method)
         content = wrap_exception(e_info)
         return content
 
     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))
     
     def _abort_queues(self):
         for stream in self.shell_streams:
             if stream:
                 self._abort_queue(stream)
 
     def _abort_queue(self, stream):
         poller = zmq.Poller()
         poller.register(stream.socket, zmq.POLLIN)
         while True:
             idents,msg = self.session.recv(stream, zmq.NOBLOCK, content=True)
             if msg is None:
                 return
 
             self.log.info("Aborting:")
             self.log.info("%s", msg)
             msg_type = msg['header']['msg_type']
             reply_type = msg_type.split('_')[0] + '_reply'
 
             status = {'status' : 'aborted'}
             md = {'engine' : self.ident}
             md.update(status)
             reply_msg = self.session.send(stream, reply_type, metadata=md,
                         content=status, parent=msg, ident=idents)
             self.log.debug("%s", reply_msg)
             # We need to wait a bit for requests to come in. This can probably
             # be set shorter for true asynchronous clients.
             poller.poll(50)
 
 
     def _no_raw_input(self):
         """Raise StdinNotImplentedError if active frontend doesn't support
         stdin."""
         raise StdinNotImplementedError("raw_input was called, but this "
                                        "frontend does not support stdin.") 
         
     def _raw_input(self, prompt, ident, parent):
         # Flush output before making the request.
         sys.stderr.flush()
         sys.stdout.flush()
         # 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
         
         # Send the input request.
         content = json_clean(dict(prompt=prompt))
         self.session.send(self.stdin_socket, u'input_request', content, parent,
                           ident=ident)
 
         # Await a response.
         while True:
             try:
                 ident, reply = self.session.recv(self.stdin_socket, 0)
             except Exception:
                 self.log.warn("Invalid Message:", exc_info=True)
             except KeyboardInterrupt:
                 # re-raise KeyboardInterrupt, to truncate traceback
                 raise KeyboardInterrupt
             else:
                 break
         try:
             value = py3compat.unicode_to_str(reply['content']['value'])
         except:
             self.log.error("Got bad raw_input reply: ")
             self.log.error("%s", parent)
             value = ''
         if value == '\x04':
             # EOF
             raise EOFError
         return value
 
     def _complete(self, msg):
         c = msg['content']
         try:
             cpos = int(c['cursor_pos'])
         except:
             # If we don't get something that we can convert to an integer, at
             # least attempt the completion guessing the cursor is at the end of
             # the text, if there's any, and otherwise of the line
             cpos = len(c['text'])
             if cpos==0:
                 cpos = len(c['line'])
         return self.shell.complete(c['text'], c['line'], cpos)
 
     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:
             self.session.send(self.iopub_socket, self._shutdown_message, ident=self._topic('shutdown'))
             self.log.debug("%s", self._shutdown_message)
         [ s.flush(zmq.POLLOUT) for s in self.shell_streams ]
 

IPython/kernel/zmq/zmqshell.py

@@ -1,618 +1,623 b''
 """A ZMQ-based subclass of InteractiveShell.
 
 This code is meant to ease the refactoring of the base InteractiveShell into
 something with a cleaner architecture for 2-process use, without actually
 breaking InteractiveShell itself.  So we're doing something a bit ugly, where
 we subclass and override what we want to fix.  Once this is working well, we
 can go back to the base class and refactor the code for a cleaner inheritance
 implementation that doesn't rely on so much monkeypatching.
 
 But this lets us maintain a fully working IPython as we develop the new
 machinery.  This should thus be thought of as scaffolding.
 """
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 from __future__ import print_function
 
 # Stdlib
 import os
 import sys
 import time
 
 # System library imports
 from zmq.eventloop import ioloop
 
 # Our own
 from IPython.core.interactiveshell import (
     InteractiveShell, InteractiveShellABC
 )
 from IPython.core import page
 from IPython.core.autocall import ZMQExitAutocall
 from IPython.core.displaypub import DisplayPublisher
 from IPython.core.error import UsageError
 from IPython.core.magics import MacroToEdit, CodeMagics
 from IPython.core.magic import magics_class, line_magic, Magics
 from IPython.core.payloadpage import install_payload_page
 from IPython.display import display, Javascript
 from IPython.kernel.inprocess.socket import SocketABC
 from IPython.kernel import (
     get_connection_file, get_connection_info, connect_qtconsole
 )
 from IPython.testing.skipdoctest import skip_doctest
 from IPython.utils import openpy
 from IPython.utils.jsonutil import json_clean, encode_images
 from IPython.utils.process import arg_split
 from IPython.utils import py3compat
 from IPython.utils.traitlets import Instance, Type, Dict, CBool, CBytes, Any
 from IPython.utils.warn import error
 from IPython.kernel.zmq.displayhook import ZMQShellDisplayHook
 from IPython.kernel.zmq.datapub import ZMQDataPublisher
 from IPython.kernel.zmq.session import extract_header
 from IPython.kernel.comm import CommManager
 from session import Session
 
 #-----------------------------------------------------------------------------
 # Functions and classes
 #-----------------------------------------------------------------------------
 
 class ZMQDisplayPublisher(DisplayPublisher):
     """A display publisher that publishes data using a ZeroMQ PUB socket."""
 
     session = Instance(Session)
     pub_socket = Instance(SocketABC)
     parent_header = Dict({})
     topic = CBytes(b'display_data')
 
     def set_parent(self, parent):
         """Set the parent for outbound messages."""
         self.parent_header = extract_header(parent)
     
     def _flush_streams(self):
         """flush IO Streams prior to display"""
         sys.stdout.flush()
         sys.stderr.flush()
 
     def publish(self, source, data, metadata=None):
         self._flush_streams()
         if metadata is None:
             metadata = {}
         self._validate_data(source, data, metadata)
         content = {}
         content['source'] = source
         content['data'] = encode_images(data)
         content['metadata'] = metadata
         self.session.send(
             self.pub_socket, u'display_data', json_clean(content),
             parent=self.parent_header, ident=self.topic,
         )
 
     def clear_output(self, wait=False):
         content = dict(wait=wait)
 
         print('\r', file=sys.stdout, end='')
         print('\r', file=sys.stderr, end='')
         self._flush_streams()
         
         self.session.send(
             self.pub_socket, u'clear_output', content,
             parent=self.parent_header, ident=self.topic,
         )
 
 @magics_class
 class KernelMagics(Magics):
     #------------------------------------------------------------------------
     # Magic overrides
     #------------------------------------------------------------------------
     # Once the base class stops inheriting from magic, this code needs to be
     # moved into a separate machinery as well.  For now, at least isolate here
     # the magics which this class needs to implement differently from the base
     # class, or that are unique to it.
 
     @line_magic
     def doctest_mode(self, parameter_s=''):
         """Toggle doctest mode on and off.
 
         This mode is intended to make IPython behave as much as possible like a
         plain Python shell, from the perspective of how its prompts, exceptions
         and output look.  This makes it easy to copy and paste parts of a
         session into doctests.  It does so by:
 
         - Changing the prompts to the classic ``>>>`` ones.
         - Changing the exception reporting mode to 'Plain'.
         - Disabling pretty-printing of output.
 
         Note that IPython also supports the pasting of code snippets that have
         leading '>>>' and '...' prompts in them.  This means that you can paste
         doctests from files or docstrings (even if they have leading
         whitespace), and the code will execute correctly.  You can then use
         '%history -t' to see the translated history; this will give you the
         input after removal of all the leading prompts and whitespace, which
         can be pasted back into an editor.
 
         With these features, you can switch into this mode easily whenever you
         need to do testing and changes to doctests, without having to leave
         your existing IPython session.
         """
 
         from IPython.utils.ipstruct import Struct
 
         # Shorthands
         shell = self.shell
         disp_formatter = self.shell.display_formatter
         ptformatter = disp_formatter.formatters['text/plain']
         # dstore is a data store kept in the instance metadata bag to track any
         # changes we make, so we can undo them later.
         dstore = shell.meta.setdefault('doctest_mode', Struct())
         save_dstore = dstore.setdefault
 
         # save a few values we'll need to recover later
         mode = save_dstore('mode', False)
         save_dstore('rc_pprint', ptformatter.pprint)
         save_dstore('rc_active_types',disp_formatter.active_types)
         save_dstore('xmode', shell.InteractiveTB.mode)
 
         if mode == False:
             # turn on
             ptformatter.pprint = False
             disp_formatter.active_types = ['text/plain']
             shell.magic('xmode Plain')
         else:
             # turn off
             ptformatter.pprint = dstore.rc_pprint
             disp_formatter.active_types = dstore.rc_active_types
             shell.magic("xmode " + dstore.xmode)
 
         # Store new mode and inform on console
         dstore.mode = bool(1-int(mode))
         mode_label = ['OFF','ON'][dstore.mode]
         print('Doctest mode is:', mode_label)
 
         # Send the payload back so that clients can modify their prompt display
         payload = dict(
             source='doctest_mode',
             mode=dstore.mode)
         shell.payload_manager.write_payload(payload)
         
     
     _find_edit_target = CodeMagics._find_edit_target
 
     @skip_doctest
     @line_magic
     def edit(self, parameter_s='', last_call=['','']):
         """Bring up an editor and execute the resulting code.
 
         Usage:
           %edit [options] [args]
 
         %edit runs an external text editor. You will need to set the command for
         this editor via the ``TerminalInteractiveShell.editor`` option in your
         configuration file before it will work.
 
         This command allows you to conveniently edit multi-line code right in
         your IPython session.
 
         If called without arguments, %edit opens up an empty editor with a
         temporary file and will execute the contents of this file when you
         close it (don't forget to save it!).
 
 
         Options:
 
         -n <number>: open the editor at a specified line number.  By default,
         the IPython editor hook uses the unix syntax 'editor +N filename', but
         you can configure this by providing your own modified hook if your
         favorite editor supports line-number specifications with a different
         syntax.
 
         -p: this will call the editor with the same data as the previous time
         it was used, regardless of how long ago (in your current session) it
         was.
 
         -r: use 'raw' input.  This option only applies to input taken from the
         user's history.  By default, the 'processed' history is used, so that
         magics are loaded in their transformed version to valid Python.  If
         this option is given, the raw input as typed as the command line is
         used instead.  When you exit the editor, it will be executed by
         IPython's own processor.
 
         -x: do not execute the edited code immediately upon exit. This is
         mainly useful if you are editing programs which need to be called with
         command line arguments, which you can then do using %run.
 
 
         Arguments:
 
         If arguments are given, the following possibilites exist:
 
         - The arguments are numbers or pairs of colon-separated numbers (like
         1 4:8 9). These are interpreted as lines of previous input to be
         loaded into the editor. The syntax is the same of the %macro command.
 
         - If the argument doesn't start with a number, it is evaluated as a
         variable and its contents loaded into the editor. You can thus edit
         any string which contains python code (including the result of
         previous edits).
 
         - If the argument is the name of an object (other than a string),
         IPython will try to locate the file where it was defined and open the
         editor at the point where it is defined. You can use `%edit function`
         to load an editor exactly at the point where 'function' is defined,
         edit it and have the file be executed automatically.
 
         If the object is a macro (see %macro for details), this opens up your
         specified editor with a temporary file containing the macro's data.
         Upon exit, the macro is reloaded with the contents of the file.
 
         Note: opening at an exact line is only supported under Unix, and some
         editors (like kedit and gedit up to Gnome 2.8) do not understand the
         '+NUMBER' parameter necessary for this feature. Good editors like
         (X)Emacs, vi, jed, pico and joe all do.
 
         - If the argument is not found as a variable, IPython will look for a
         file with that name (adding .py if necessary) and load it into the
         editor. It will execute its contents with execfile() when you exit,
         loading any code in the file into your interactive namespace.
 
         After executing your code, %edit will return as output the code you
         typed in the editor (except when it was an existing file). This way
         you can reload the code in further invocations of %edit as a variable,
         via _<NUMBER> or Out[<NUMBER>], where <NUMBER> is the prompt number of
         the output.
 
         Note that %edit is also available through the alias %ed.
 
         This is an example of creating a simple function inside the editor and
         then modifying it. First, start up the editor:
 
         In [1]: ed
         Editing... done. Executing edited code...
         Out[1]: 'def foo():n    print "foo() was defined in an editing session"n'
 
         We can then call the function foo():
 
         In [2]: foo()
         foo() was defined in an editing session
 
         Now we edit foo.  IPython automatically loads the editor with the
         (temporary) file where foo() was previously defined:
 
         In [3]: ed foo
         Editing... done. Executing edited code...
 
         And if we call foo() again we get the modified version:
 
         In [4]: foo()
         foo() has now been changed!
 
         Here is an example of how to edit a code snippet successive
         times. First we call the editor:
 
         In [5]: ed
         Editing... done. Executing edited code...
         hello
         Out[5]: "print 'hello'n"
 
         Now we call it again with the previous output (stored in _):
 
         In [6]: ed _
         Editing... done. Executing edited code...
         hello world
         Out[6]: "print 'hello world'n"
 
         Now we call it with the output #8 (stored in _8, also as Out[8]):
 
         In [7]: ed _8
         Editing... done. Executing edited code...
         hello again
         Out[7]: "print 'hello again'n"
         """
 
         opts,args = self.parse_options(parameter_s,'prn:')
 
         try:
             filename, lineno, _ = CodeMagics._find_edit_target(self.shell, args, opts, last_call)
         except MacroToEdit as e:
             # TODO: Implement macro editing over 2 processes.
             print("Macro editing not yet implemented in 2-process model.")
             return
 
         # Make sure we send to the client an absolute path, in case the working
         # directory of client and kernel don't match
         filename = os.path.abspath(filename)
 
         payload = {
             'source' : 'edit_magic',
             'filename' : filename,
             'line_number' : lineno
         }
         self.shell.payload_manager.write_payload(payload)
 
     # A few magics that are adapted to the specifics of using pexpect and a
     # remote terminal
 
     @line_magic
     def clear(self, arg_s):
         """Clear the terminal."""
         if os.name == 'posix':
             self.shell.system("clear")
         else:
             self.shell.system("cls")
 
     if os.name == 'nt':
         # This is the usual name in windows
         cls = line_magic('cls')(clear)
 
     # Terminal pagers won't work over pexpect, but we do have our own pager
 
     @line_magic
     def less(self, arg_s):
         """Show a file through the pager.
 
         Files ending in .py are syntax-highlighted."""
         if not arg_s:
             raise UsageError('Missing filename.')
 
         cont = open(arg_s).read()
         if arg_s.endswith('.py'):
             cont = self.shell.pycolorize(openpy.read_py_file(arg_s, skip_encoding_cookie=False))
         else:
             cont = open(arg_s).read()
         page.page(cont)
 
     more = line_magic('more')(less)
 
     # Man calls a pager, so we also need to redefine it
     if os.name == 'posix':
         @line_magic
         def man(self, arg_s):
             """Find the man page for the given command and display in pager."""
             page.page(self.shell.getoutput('man %s | col -b' % arg_s,
                                            split=False))
 
     @line_magic
     def connect_info(self, arg_s):
         """Print information for connecting other clients to this kernel
         
         It will print the contents of this session's connection file, as well as
         shortcuts for local clients.
         
         In the simplest case, when called from the most recently launched kernel,
         secondary clients can be connected, simply with:
         
         $> ipython <app> --existing
         
         """
         
         from IPython.core.application import BaseIPythonApplication as BaseIPApp
         
         if BaseIPApp.initialized():
             app = BaseIPApp.instance()
             security_dir = app.profile_dir.security_dir
             profile = app.profile
         else:
             profile = 'default'
             security_dir = ''
         
         try:
             connection_file = get_connection_file()
             info = get_connection_info(unpack=False)
         except Exception as e:
             error("Could not get connection info: %r" % e)
             return
         
         # add profile flag for non-default profile
         profile_flag = "--profile %s" % profile if profile != 'default' else ""
         
         # if it's in the security dir, truncate to basename
         if security_dir == os.path.dirname(connection_file):
             connection_file = os.path.basename(connection_file)
         
         
         print (info + '\n')
         print ("Paste the above JSON into a file, and connect with:\n"
             "    $> ipython <app> --existing <file>\n"
             "or, if you are local, you can connect with just:\n"
             "    $> ipython <app> --existing {0} {1}\n"
             "or even just:\n"
             "    $> ipython <app> --existing {1}\n"
             "if this is the most recent IPython session you have started.".format(
             connection_file, profile_flag
             )
         )
 
     @line_magic
     def qtconsole(self, arg_s):
         """Open a qtconsole connected to this kernel.
         
         Useful for connecting a qtconsole to running notebooks, for better
         debugging.
         """
         
         # %qtconsole should imply bind_kernel for engines:
         try:
             from IPython.parallel import bind_kernel
         except ImportError:
             # technically possible, because parallel has higher pyzmq min-version
             pass
         else:
             bind_kernel()
         
         try:
             p = connect_qtconsole(argv=arg_split(arg_s, os.name=='posix'))
         except Exception as e:
             error("Could not start qtconsole: %r" % e)
             return
     
     @line_magic
     def autosave(self, arg_s):
         """Set the autosave interval in the notebook (in seconds).
         
         The default value is 120, or two minutes.
         ``%autosave 0`` will disable autosave.
         
         This magic only has an effect when called from the notebook interface.
         It has no effect when called in a startup file.
         """
         
         try:
             interval = int(arg_s)
         except ValueError:
             raise UsageError("%%autosave requires an integer, got %r" % arg_s)
         
         # javascript wants milliseconds
         milliseconds = 1000 * interval
         display(Javascript("IPython.notebook.set_autosave_interval(%i)" % milliseconds),
             include=['application/javascript']
         )
         if interval:
             print("Autosaving every %i seconds" % interval)
         else:
             print("Autosave disabled")
 
 
 class ZMQInteractiveShell(InteractiveShell):
     """A subclass of InteractiveShell for ZMQ."""
 
     displayhook_class = Type(ZMQShellDisplayHook)
     display_pub_class = Type(ZMQDisplayPublisher)
     data_pub_class = Type(ZMQDataPublisher)
     kernel = Any()
+    parent_header = Any()
 
     # Override the traitlet in the parent class, because there's no point using
     # readline for the kernel. Can be removed when the readline code is moved
     # to the terminal frontend.
     colors_force = CBool(True)
     readline_use = CBool(False)
     # autoindent has no meaning in a zmqshell, and attempting to enable it
     # will print a warning in the absence of readline.
     autoindent = CBool(False)
 
     exiter = Instance(ZMQExitAutocall)
     def _exiter_default(self):
         return ZMQExitAutocall(self)
     
     def _exit_now_changed(self, name, old, new):
         """stop eventloop when exit_now fires"""
         if new:
             loop = ioloop.IOLoop.instance()
             loop.add_timeout(time.time()+0.1, loop.stop)
 
     keepkernel_on_exit = None
 
     # Over ZeroMQ, GUI control isn't done with PyOS_InputHook as there is no
     # interactive input being read; we provide event loop support in ipkernel
     @staticmethod
     def enable_gui(gui):
         from .eventloops import enable_gui as real_enable_gui
         try:
             real_enable_gui(gui)
         except ValueError as e:
             raise UsageError("%s" % e)
 
     def init_environment(self):
         """Configure the user's environment.
 
         """
         env = os.environ
         # These two ensure 'ls' produces nice coloring on BSD-derived systems
         env['TERM'] = 'xterm-color'
         env['CLICOLOR'] = '1'
         # Since normal pagers don't work at all (over pexpect we don't have
         # single-key control of the subprocess), try to disable paging in
         # subprocesses as much as possible.
         env['PAGER'] = 'cat'
         env['GIT_PAGER'] = 'cat'
         
         # And install the payload version of page.
         install_payload_page()
 
     def auto_rewrite_input(self, cmd):
         """Called to show the auto-rewritten input for autocall and friends.
 
         FIXME: this payload is currently not correctly processed by the
         frontend.
         """
         new = self.prompt_manager.render('rewrite') + cmd
         payload = dict(
             source='auto_rewrite_input',
             transformed_input=new,
             )
         self.payload_manager.write_payload(payload)
 
     def ask_exit(self):
         """Engage the exit actions."""
         self.exit_now = True
         payload = dict(
             source='ask_exit',
             exit=True,
             keepkernel=self.keepkernel_on_exit,
             )
         self.payload_manager.write_payload(payload)
 
     def _showtraceback(self, etype, evalue, stb):
 
         exc_content = {
             u'traceback' : stb,
             u'ename' : unicode(etype.__name__),
             u'evalue' : py3compat.safe_unicode(evalue),
         }
 
         dh = self.displayhook
         # Send exception info over pub socket for other clients than the caller
         # to pick up
         topic = None
         if dh.topic:
             topic = dh.topic.replace(b'pyout', b'pyerr')
         
         exc_msg = dh.session.send(dh.pub_socket, u'pyerr', json_clean(exc_content), dh.parent_header, ident=topic)
 
         # FIXME - Hack: store exception info in shell object.  Right now, the
         # caller is reading this info after the fact, we need to fix this logic
         # to remove this hack.  Even uglier, we need to store the error status
         # here, because in the main loop, the logic that sets it is being
         # skipped because runlines swallows the exceptions.
         exc_content[u'status'] = u'error'
         self._reply_content = exc_content
         # /FIXME
 
         return exc_content
 
     def set_next_input(self, text):
         """Send the specified text to the frontend to be presented at the next
         input cell."""
         payload = dict(
             source='set_next_input',
             text=text
         )
         self.payload_manager.write_payload(payload)
     
     def set_parent(self, parent):
         """Set the parent header for associating output with its triggering input"""
+        self.parent_header = parent
         self.displayhook.set_parent(parent)
         self.display_pub.set_parent(parent)
         self.data_pub.set_parent(parent)
         try:
             sys.stdout.set_parent(parent)
         except AttributeError:
             pass
         try:
             sys.stderr.set_parent(parent)
         except AttributeError:
             pass
     
+    def get_parent(self):
+        return self.parent_header
+    
     #-------------------------------------------------------------------------
     # Things related to magics
     #-------------------------------------------------------------------------
 
     def init_magics(self):
         super(ZMQInteractiveShell, self).init_magics()
         self.register_magics(KernelMagics)
         self.magics_manager.register_alias('ed', 'edit')
     
     def init_comms(self):
         self.comm_manager = CommManager(shell=self, parent=self)
         self.configurables.append(self.comm_manager)
 
 
 InteractiveShellABC.register(ZMQInteractiveShell)