move apply serialization into zmq.serialize

IPython/zmq/serialize.py

@@ -0,0 +1,179 b''
+"""serialization utilities for apply messages
+
+Authors:
+
+* Min RK
+"""
+#-----------------------------------------------------------------------------
+#  Copyright (C) 2010-2011  The IPython Development Team
+#
+#  Distributed under the terms of the BSD License.  The full license is in
+#  the file COPYING, distributed as part of this software.
+#-----------------------------------------------------------------------------
+
+#-----------------------------------------------------------------------------
+# Imports
+#-----------------------------------------------------------------------------
+
+# Standard library imports
+import logging
+import os
+import re
+import socket
+import sys
+
+try:
+    import cPickle
+    pickle = cPickle
+except:
+    cPickle = None
+    import pickle
+
+
+# IPython imports
+from IPython.utils import py3compat
+from IPython.utils.pickleutil import can, uncan, canSequence, uncanSequence
+from IPython.utils.newserialized import serialize, unserialize
+
+if py3compat.PY3:
+    buffer = memoryview
+
+#-----------------------------------------------------------------------------
+# Serialization Functions
+#-----------------------------------------------------------------------------
+
+def serialize_object(obj, threshold=64e-6):
+    """Serialize an object into a list of sendable buffers.
+    
+    Parameters
+    ----------
+    
+    obj : object
+        The object to be serialized
+    threshold : float
+        The threshold for not double-pickling the content.
+        
+    
+    Returns
+    -------
+    ('pmd', [bufs]) :
+        where pmd is the pickled metadata wrapper,
+        bufs is a list of data buffers
+    """
+    databuffers = []
+    if isinstance(obj, (list, tuple)):
+        clist = canSequence(obj)
+        slist = map(serialize, clist)
+        for s in slist:
+            if s.typeDescriptor in ('buffer', 'ndarray') or s.getDataSize() > threshold:
+                databuffers.append(s.getData())
+                s.data = None
+        return pickle.dumps(slist,-1), databuffers
+    elif isinstance(obj, dict):
+        sobj = {}
+        for k in sorted(obj.iterkeys()):
+            s = serialize(can(obj[k]))
+            if s.typeDescriptor in ('buffer', 'ndarray') or s.getDataSize() > threshold:
+                databuffers.append(s.getData())
+                s.data = None
+            sobj[k] = s
+        return pickle.dumps(sobj,-1),databuffers
+    else:
+        s = serialize(can(obj))
+        if s.typeDescriptor in ('buffer', 'ndarray') or s.getDataSize() > threshold:
+            databuffers.append(s.getData())
+            s.data = None
+        return pickle.dumps(s,-1),databuffers
+            
+        
+def unserialize_object(bufs):
+    """reconstruct an object serialized by serialize_object from data buffers."""
+    bufs = list(bufs)
+    sobj = pickle.loads(bufs.pop(0))
+    if isinstance(sobj, (list, tuple)):
+        for s in sobj:
+            if s.data is None:
+                s.data = bufs.pop(0)
+        return uncanSequence(map(unserialize, sobj)), bufs
+    elif isinstance(sobj, dict):
+        newobj = {}
+        for k in sorted(sobj.iterkeys()):
+            s = sobj[k]
+            if s.data is None:
+                s.data = bufs.pop(0)
+            newobj[k] = uncan(unserialize(s))
+        return newobj, bufs
+    else:
+        if sobj.data is None:
+            sobj.data = bufs.pop(0)
+        return uncan(unserialize(sobj)), bufs
+
+def pack_apply_message(f, args, kwargs, threshold=64e-6):
+    """pack up a function, args, and kwargs to be sent over the wire
+    as a series of buffers. Any object whose data is larger than `threshold`
+    will not have their data copied (currently only numpy arrays support zero-copy)"""
+    msg = [pickle.dumps(can(f),-1)]
+    databuffers = [] # for large objects
+    sargs, bufs = serialize_object(args,threshold)
+    msg.append(sargs)
+    databuffers.extend(bufs)
+    skwargs, bufs = serialize_object(kwargs,threshold)
+    msg.append(skwargs)
+    databuffers.extend(bufs)
+    msg.extend(databuffers)
+    return msg
+
+def unpack_apply_message(bufs, g=None, copy=True):
+    """unpack f,args,kwargs from buffers packed by pack_apply_message()
+    Returns: original f,args,kwargs"""
+    bufs = list(bufs) # allow us to pop
+    assert len(bufs) >= 3, "not enough buffers!"
+    if not copy:
+        for i in range(3):
+            bufs[i] = bufs[i].bytes
+    cf = pickle.loads(bufs.pop(0))
+    sargs = list(pickle.loads(bufs.pop(0)))
+    skwargs = dict(pickle.loads(bufs.pop(0)))
+    # print sargs, skwargs
+    f = uncan(cf, g)
+    for sa in sargs:
+        if sa.data is None:
+            m = bufs.pop(0)
+            if sa.getTypeDescriptor() in ('buffer', 'ndarray'):
+                # always use a buffer, until memoryviews get sorted out
+                sa.data = buffer(m)
+                # disable memoryview support
+                # if copy:
+                #     sa.data = buffer(m)
+                # else:
+                #     sa.data = m.buffer
+            else:
+                if copy:
+                    sa.data = m
+                else:
+                    sa.data = m.bytes
+    
+    args = uncanSequence(map(unserialize, sargs), g)
+    kwargs = {}
+    for k in sorted(skwargs.iterkeys()):
+        sa = skwargs[k]
+        if sa.data is None:
+            m = bufs.pop(0)
+            if sa.getTypeDescriptor() in ('buffer', 'ndarray'):
+                # always use a buffer, until memoryviews get sorted out
+                sa.data = buffer(m)
+                # disable memoryview support
+                # if copy:
+                #     sa.data = buffer(m)
+                # else:
+                #     sa.data = m.buffer
+            else:
+                if copy:
+                    sa.data = m
+                else:
+                    sa.data = m.bytes
+
+        kwargs[k] = uncan(unserialize(sa), g)
+    
+    return f,args,kwargs
+

IPython/parallel/util.py

@@ -1,495 +1,364 b''
 """some generic utilities for dealing with classes, urls, and serialization
 
 Authors:
 
 * Min RK
 """
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2010-2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 # Standard library imports.
 import logging
 import os
 import re
 import stat
 import socket
 import sys
 from signal import signal, SIGINT, SIGABRT, SIGTERM
 try:
     from signal import SIGKILL
 except ImportError:
     SIGKILL=None
 
 try:
     import cPickle
     pickle = cPickle
 except:
     cPickle = None
     import pickle
 
 # System library imports
 import zmq
 from zmq.log import handlers
 
 from IPython.external.decorator import decorator
 
 # IPython imports
 from IPython.config.application import Application
 from IPython.utils import py3compat
 from IPython.utils.pickleutil import can, uncan, canSequence, uncanSequence
 from IPython.utils.newserialized import serialize, unserialize
 from IPython.zmq.log import EnginePUBHandler
+from IPython.zmq.serialize import (
+    unserialize_object, serialize_object, pack_apply_message, unpack_apply_message
+)
 
 if py3compat.PY3:
     buffer = memoryview
 
 #-----------------------------------------------------------------------------
 # Classes
 #-----------------------------------------------------------------------------
 
 class Namespace(dict):
     """Subclass of dict for attribute access to keys."""
     
     def __getattr__(self, key):
         """getattr aliased to getitem"""
         if key in self.iterkeys():
             return self[key]
         else:
             raise NameError(key)
 
     def __setattr__(self, key, value):
         """setattr aliased to setitem, with strict"""
         if hasattr(dict, key):
             raise KeyError("Cannot override dict keys %r"%key)
         self[key] = value
     
 
 class ReverseDict(dict):
     """simple double-keyed subset of dict methods."""
     
     def __init__(self, *args, **kwargs):
         dict.__init__(self, *args, **kwargs)
         self._reverse = dict()
         for key, value in self.iteritems():
             self._reverse[value] = key
     
     def __getitem__(self, key):
         try:
             return dict.__getitem__(self, key)
         except KeyError:
             return self._reverse[key]
     
     def __setitem__(self, key, value):
         if key in self._reverse:
             raise KeyError("Can't have key %r on both sides!"%key)
         dict.__setitem__(self, key, value)
         self._reverse[value] = key
     
     def pop(self, key):
         value = dict.pop(self, key)
         self._reverse.pop(value)
         return value
     
     def get(self, key, default=None):
         try:
             return self[key]
         except KeyError:
             return default
 
 #-----------------------------------------------------------------------------
 # Functions
 #-----------------------------------------------------------------------------
 
 @decorator
 def log_errors(f, self, *args, **kwargs):
     """decorator to log unhandled exceptions raised in a method.
     
     For use wrapping on_recv callbacks, so that exceptions
     do not cause the stream to be closed.
     """
     try:
         return f(self, *args, **kwargs)
     except Exception:
         self.log.error("Uncaught exception in %r" % f, exc_info=True)
     
 
 def asbytes(s):
     """ensure that an object is ascii bytes"""
     if isinstance(s, unicode):
         s = s.encode('ascii')
     return s
 
 def is_url(url):
     """boolean check for whether a string is a zmq url"""
     if '://' not in url:
         return False
     proto, addr = url.split('://', 1)
     if proto.lower() not in ['tcp','pgm','epgm','ipc','inproc']:
         return False
     return True
 
 def validate_url(url):
     """validate a url for zeromq"""
     if not isinstance(url, basestring):
         raise TypeError("url must be a string, not %r"%type(url))
     url = url.lower()
     
     proto_addr = url.split('://')
     assert len(proto_addr) == 2, 'Invalid url: %r'%url
     proto, addr = proto_addr
     assert proto in ['tcp','pgm','epgm','ipc','inproc'], "Invalid protocol: %r"%proto
     
     # domain pattern adapted from http://www.regexlib.com/REDetails.aspx?regexp_id=391
     # author: Remi Sabourin
     pat = re.compile(r'^([\w\d]([\w\d\-]{0,61}[\w\d])?\.)*[\w\d]([\w\d\-]{0,61}[\w\d])?$')
     
     if proto == 'tcp':
         lis = addr.split(':')
         assert len(lis) == 2, 'Invalid url: %r'%url
         addr,s_port = lis
         try:
             port = int(s_port)
         except ValueError:
             raise AssertionError("Invalid port %r in url: %r"%(port, url))
         
         assert addr == '*' or pat.match(addr) is not None, 'Invalid url: %r'%url
         
     else:
         # only validate tcp urls currently
         pass
     
     return True
 
 
 def validate_url_container(container):
     """validate a potentially nested collection of urls."""
     if isinstance(container, basestring):
         url = container
         return validate_url(url)
     elif isinstance(container, dict):
         container = container.itervalues()
     
     for element in container:
         validate_url_container(element)
 
 
 def split_url(url):
     """split a zmq url (tcp://ip:port) into ('tcp','ip','port')."""
     proto_addr = url.split('://')
     assert len(proto_addr) == 2, 'Invalid url: %r'%url
     proto, addr = proto_addr
     lis = addr.split(':')
     assert len(lis) == 2, 'Invalid url: %r'%url
     addr,s_port = lis
     return proto,addr,s_port
     
 def disambiguate_ip_address(ip, location=None):
     """turn multi-ip interfaces '0.0.0.0' and '*' into connectable
     ones, based on the location (default interpretation of location is localhost)."""
     if ip in ('0.0.0.0', '*'):
         try:
             external_ips = socket.gethostbyname_ex(socket.gethostname())[2]
         except (socket.gaierror, IndexError):
             # couldn't identify this machine, assume localhost
             external_ips = []
         if location is None or location in external_ips or not external_ips:
             # If location is unspecified or cannot be determined, assume local
             ip='127.0.0.1'
         elif location:
             return location
     return ip
 
 def disambiguate_url(url, location=None):
     """turn multi-ip interfaces '0.0.0.0' and '*' into connectable
     ones, based on the location (default interpretation is localhost).
     
     This is for zeromq urls, such as tcp://*:10101."""
     try:
         proto,ip,port = split_url(url)
     except AssertionError:
         # probably not tcp url; could be ipc, etc.
         return url
     
     ip = disambiguate_ip_address(ip,location)
     
     return "%s://%s:%s"%(proto,ip,port)
 
-def serialize_object(obj, threshold=64e-6):
-    """Serialize an object into a list of sendable buffers.
-    
-    Parameters
-    ----------
-    
-    obj : object
-        The object to be serialized
-    threshold : float
-        The threshold for not double-pickling the content.
-        
-    
-    Returns
-    -------
-    ('pmd', [bufs]) :
-        where pmd is the pickled metadata wrapper,
-        bufs is a list of data buffers
-    """
-    databuffers = []
-    if isinstance(obj, (list, tuple)):
-        clist = canSequence(obj)
-        slist = map(serialize, clist)
-        for s in slist:
-            if s.typeDescriptor in ('buffer', 'ndarray') or s.getDataSize() > threshold:
-                databuffers.append(s.getData())
-                s.data = None
-        return pickle.dumps(slist,-1), databuffers
-    elif isinstance(obj, dict):
-        sobj = {}
-        for k in sorted(obj.iterkeys()):
-            s = serialize(can(obj[k]))
-            if s.typeDescriptor in ('buffer', 'ndarray') or s.getDataSize() > threshold:
-                databuffers.append(s.getData())
-                s.data = None
-            sobj[k] = s
-        return pickle.dumps(sobj,-1),databuffers
-    else:
-        s = serialize(can(obj))
-        if s.typeDescriptor in ('buffer', 'ndarray') or s.getDataSize() > threshold:
-            databuffers.append(s.getData())
-            s.data = None
-        return pickle.dumps(s,-1),databuffers
-            
-        
-def unserialize_object(bufs):
-    """reconstruct an object serialized by serialize_object from data buffers."""
-    bufs = list(bufs)
-    sobj = pickle.loads(bufs.pop(0))
-    if isinstance(sobj, (list, tuple)):
-        for s in sobj:
-            if s.data is None:
-                s.data = bufs.pop(0)
-        return uncanSequence(map(unserialize, sobj)), bufs
-    elif isinstance(sobj, dict):
-        newobj = {}
-        for k in sorted(sobj.iterkeys()):
-            s = sobj[k]
-            if s.data is None:
-                s.data = bufs.pop(0)
-            newobj[k] = uncan(unserialize(s))
-        return newobj, bufs
-    else:
-        if sobj.data is None:
-            sobj.data = bufs.pop(0)
-        return uncan(unserialize(sobj)), bufs
-
-def pack_apply_message(f, args, kwargs, threshold=64e-6):
-    """pack up a function, args, and kwargs to be sent over the wire
-    as a series of buffers. Any object whose data is larger than `threshold`
-    will not have their data copied (currently only numpy arrays support zero-copy)"""
-    msg = [pickle.dumps(can(f),-1)]
-    databuffers = [] # for large objects
-    sargs, bufs = serialize_object(args,threshold)
-    msg.append(sargs)
-    databuffers.extend(bufs)
-    skwargs, bufs = serialize_object(kwargs,threshold)
-    msg.append(skwargs)
-    databuffers.extend(bufs)
-    msg.extend(databuffers)
-    return msg
-
-def unpack_apply_message(bufs, g=None, copy=True):
-    """unpack f,args,kwargs from buffers packed by pack_apply_message()
-    Returns: original f,args,kwargs"""
-    bufs = list(bufs) # allow us to pop
-    assert len(bufs) >= 3, "not enough buffers!"
-    if not copy:
-        for i in range(3):
-            bufs[i] = bufs[i].bytes
-    cf = pickle.loads(bufs.pop(0))
-    sargs = list(pickle.loads(bufs.pop(0)))
-    skwargs = dict(pickle.loads(bufs.pop(0)))
-    # print sargs, skwargs
-    f = uncan(cf, g)
-    for sa in sargs:
-        if sa.data is None:
-            m = bufs.pop(0)
-            if sa.getTypeDescriptor() in ('buffer', 'ndarray'):
-                # always use a buffer, until memoryviews get sorted out
-                sa.data = buffer(m)
-                # disable memoryview support
-                # if copy:
-                #     sa.data = buffer(m)
-                # else:
-                #     sa.data = m.buffer
-            else:
-                if copy:
-                    sa.data = m
-                else:
-                    sa.data = m.bytes
-    
-    args = uncanSequence(map(unserialize, sargs), g)
-    kwargs = {}
-    for k in sorted(skwargs.iterkeys()):
-        sa = skwargs[k]
-        if sa.data is None:
-            m = bufs.pop(0)
-            if sa.getTypeDescriptor() in ('buffer', 'ndarray'):
-                # always use a buffer, until memoryviews get sorted out
-                sa.data = buffer(m)
-                # disable memoryview support
-                # if copy:
-                #     sa.data = buffer(m)
-                # else:
-                #     sa.data = m.buffer
-            else:
-                if copy:
-                    sa.data = m
-                else:
-                    sa.data = m.bytes
-
-        kwargs[k] = uncan(unserialize(sa), g)
-    
-    return f,args,kwargs
 
 #--------------------------------------------------------------------------
 # helpers for implementing old MEC API via view.apply
 #--------------------------------------------------------------------------
 
 def interactive(f):
     """decorator for making functions appear as interactively defined.
     This results in the function being linked to the user_ns as globals()
     instead of the module globals().
     """
     f.__module__ = '__main__'
     return f
 
 @interactive
 def _push(**ns):
     """helper method for implementing `client.push` via `client.apply`"""
     globals().update(ns)
 
 @interactive
 def _pull(keys):
     """helper method for implementing `client.pull` via `client.apply`"""
     user_ns = globals()
     if isinstance(keys, (list,tuple, set)):
         for key in keys:
             if not user_ns.has_key(key):
                 raise NameError("name '%s' is not defined"%key)
         return map(user_ns.get, keys)
     else:
         if not user_ns.has_key(keys):
             raise NameError("name '%s' is not defined"%keys)
         return user_ns.get(keys)
 
 @interactive
 def _execute(code):
     """helper method for implementing `client.execute` via `client.apply`"""
     exec code in globals()
 
 #--------------------------------------------------------------------------
 # extra process management utilities
 #--------------------------------------------------------------------------
 
 _random_ports = set()
 
 def select_random_ports(n):
     """Selects and return n random ports that are available."""
     ports = []
     for i in xrange(n):
         sock = socket.socket()
         sock.bind(('', 0))
         while sock.getsockname()[1] in _random_ports:
             sock.close()
             sock = socket.socket()
             sock.bind(('', 0))
         ports.append(sock)
     for i, sock in enumerate(ports):
         port = sock.getsockname()[1]
         sock.close()
         ports[i] = port
         _random_ports.add(port)
     return ports
 
 def signal_children(children):
     """Relay interupt/term signals to children, for more solid process cleanup."""
     def terminate_children(sig, frame):
         log = Application.instance().log
         log.critical("Got signal %i, terminating children..."%sig)
         for child in children:
             child.terminate()
         
         sys.exit(sig != SIGINT)
         # sys.exit(sig)
     for sig in (SIGINT, SIGABRT, SIGTERM):
         signal(sig, terminate_children)
 
 def generate_exec_key(keyfile):
     import uuid
     newkey = str(uuid.uuid4())
     with open(keyfile, 'w') as f:
         # f.write('ipython-key ')
         f.write(newkey+'\n')
     # set user-only RW permissions (0600)
     # this will have no effect on Windows
     os.chmod(keyfile, stat.S_IRUSR|stat.S_IWUSR)
 
 
 def integer_loglevel(loglevel):
     try:
         loglevel = int(loglevel)
     except ValueError:
         if isinstance(loglevel, str):
             loglevel = getattr(logging, loglevel)
     return loglevel
 
 def connect_logger(logname, context, iface, root="ip", loglevel=logging.DEBUG):
     logger = logging.getLogger(logname)
     if any([isinstance(h, handlers.PUBHandler) for h in logger.handlers]):
         # don't add a second PUBHandler
         return
     loglevel = integer_loglevel(loglevel)
     lsock = context.socket(zmq.PUB)
     lsock.connect(iface)
     handler = handlers.PUBHandler(lsock)
     handler.setLevel(loglevel)
     handler.root_topic = root
     logger.addHandler(handler)
     logger.setLevel(loglevel)
 
 def connect_engine_logger(context, iface, engine, loglevel=logging.DEBUG):
     logger = logging.getLogger()
     if any([isinstance(h, handlers.PUBHandler) for h in logger.handlers]):
         # don't add a second PUBHandler
         return
     loglevel = integer_loglevel(loglevel)
     lsock = context.socket(zmq.PUB)
     lsock.connect(iface)
     handler = EnginePUBHandler(engine, lsock)
     handler.setLevel(loglevel)
     logger.addHandler(handler)
     logger.setLevel(loglevel)
     return logger
 
 def local_logger(logname, loglevel=logging.DEBUG):
     loglevel = integer_loglevel(loglevel)
     logger = logging.getLogger(logname)
     if any([isinstance(h, logging.StreamHandler) for h in logger.handlers]):
         # don't add a second StreamHandler
         return
     handler = logging.StreamHandler()
     handler.setLevel(loglevel)
     formatter = logging.Formatter("%(asctime)s.%(msecs).03d [%(name)s] %(message)s",
                 datefmt="%Y-%m-%d %H:%M:%S")
     handler.setFormatter(formatter)
 
     logger.addHandler(handler)
     logger.setLevel(loglevel)
     return logger
 

IPython/zmq/ipkernel.py

@@ -1,864 +1,868 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.
 """
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 from __future__ import print_function
 
 # Standard library imports.
 import __builtin__
 import atexit
 import sys
 import time
 import traceback
 import logging
+import uuid
+
+from datetime import datetime
 from signal import (
         signal, default_int_handler, SIGINT, SIG_IGN
 )
 # System library imports.
 import zmq
 
 # Local imports.
 from IPython.core import pylabtools
 from IPython.config.configurable import Configurable
 from IPython.config.application import boolean_flag, catch_config_error
 from IPython.core.application import ProfileDir
 from IPython.core.error import StdinNotImplementedError
 from IPython.core.shellapp import (
     InteractiveShellApp, shell_flags, shell_aliases
 )
 from IPython.utils import io
 from IPython.utils import py3compat
 from IPython.utils.frame import extract_module_locals
 from IPython.utils.jsonutil import json_clean
 from IPython.utils.traitlets import (
     Any, Instance, Float, Dict, CaselessStrEnum, List, Set
 )
 
 from entry_point import base_launch_kernel
 from kernelapp import KernelApp, kernel_flags, kernel_aliases
+from serialize import serialize_object, unpack_apply_message
 from session import Session, Message
 from zmqshell import ZMQInteractiveShell
 
 
 #-----------------------------------------------------------------------------
 # Main kernel class
 #-----------------------------------------------------------------------------
 
 class Kernel(Configurable):
 
     #---------------------------------------------------------------------------
     # Kernel interface
     #---------------------------------------------------------------------------
 
     # attribute to override with a GUI
     eventloop = Any(None)
 
     shell = Instance('IPython.core.interactiveshell.InteractiveShellABC')
     session = Instance(Session)
     profile_dir = Instance('IPython.core.profiledir.ProfileDir')
     shell_sockets = List()
     control_socket = Instance('zmq.Socket')
     iopub_socket = Instance('zmq.Socket')
     log = Instance(logging.Logger)
     
     user_module = Instance('types.ModuleType')
     def _user_module_changed(self, name, old, new):
         if self.shell is not None:
             self.shell.user_module = new
     
     user_ns = Dict(default_value=None)
     def _user_ns_changed(self, name, old, new):
         if self.shell is not None:
             self.shell.user_ns = new
             self.shell.init_user_ns()
 
     # 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()
     
     # set of aborted msg_ids
     aborted = Set()
 
 
 
     def __init__(self, **kwargs):
         super(Kernel, self).__init__(**kwargs)
 
         # Before we even start up the shell, register *first* our exit handlers
         # so they come before the shell's
         atexit.register(self._at_shutdown)
 
         # Initialize the InteractiveShell subclass
         self.shell = ZMQInteractiveShell.instance(config=self.config,
             profile_dir = self.profile_dir,
             user_module = self.user_module,
             user_ns     = self.user_ns,
         )
         self.shell.displayhook.session = self.session
         self.shell.displayhook.pub_socket = self.iopub_socket
         self.shell.display_pub.session = self.session
         self.shell.display_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',
                       'connect_request', 'shutdown_request',
                       'apply_request',
                     ]
         self.handlers = {}
         for msg_type in msg_types:
             self.handlers[msg_type] = getattr(self, msg_type)
         
         control_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 do_one_iteration(self):
         """Do one iteration of the kernel's evaluation loop.
         """
         
         # always flush control socket first
         while True:
             if self.control_socket is None:
                 break
             try:
                 idents,msg = self.session.recv(self.control_socket, zmq.NOBLOCK)
             except Exception:
                 self.log.warn("Invalid Control Message:", exc_info=True)
                 continue
             if msg is None:
                 break
             self.dispatch_message(self.control_socket, idents, msg, self.control_handlers)
         
         for socket in self.shell_sockets:
             try:
                 idents,msg = self.session.recv(socket, zmq.NOBLOCK, copy=False)
             except Exception:
                 self.log.warn("Invalid Message:", exc_info=True)
                 continue
             
             if msg is None:
                 continue
             
             self.dispatch_message(socket, idents, msg, self.handlers)
 
     def dispatch_message(self, socket, idents, msg, handlers):
         msg_type = msg['header']['msg_type']
         msg_id = msg['header']['msg_id']
 
         # This assert will raise in versions of zeromq 2.0.7 and lesser.
         # We now require 2.0.8 or above, so we can uncomment for safety.
         # print(ident,msg, file=sys.__stdout__)
         assert idents is not None, "Missing message part."
 
         # 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'])
 
         # check if request has been aborted
         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'}
             reply_msg = self.session.send(socket, reply_type, subheader=status,
                         content=status, parent=msg, ident=idents)
             return
         
 
         # Find and call actual handler for message
         handler = handlers.get(msg_type, None)
         if handler is None:
             self.log.error("UNKNOWN MESSAGE TYPE: %s", msg)
         else:
             handler(socket, idents, msg)
 
         # Check whether we should exit, in case the incoming message set the
         # exit flag on
         if self.shell.exit_now:
             self.log.debug('\nExiting IPython kernel...')
             # We do a normal, clean exit, which allows any actions registered
             # via atexit (such as history saving) to take place.
             sys.exit(0)
 
 
     def start(self):
         """ Start the kernel main loop.
         """
         # a KeyboardInterrupt (SIGINT) can occur on any python statement, so
         # let's ignore (SIG_IGN) them until we're in a place to handle them properly
         signal(SIGINT,SIG_IGN)
         poller = zmq.Poller()
         for socket in self.shell_sockets:
             poller.register(socket, zmq.POLLIN)
         if self.control_socket:
             poller.register(self.control_socket, zmq.POLLIN)
         
         # loop while self.eventloop has not been overridden
         while self.eventloop is None:
             try:
                 # scale by extra factor of 10, because there is no
                 # reason for this to be anything less than ~ 0.1s
                 # since it is a real poller and will respond
                 # to events immediately
 
                 # double nested try/except, to properly catch KeyboardInterrupt
                 # due to pyzmq Issue #130
                 try:
                     poller.poll(10*1000*self._poll_interval)
                     # restore raising of KeyboardInterrupt
                     signal(SIGINT, default_int_handler)
                     self.do_one_iteration()
                 except:
                     raise
                 finally:
                     # prevent raising of KeyboardInterrupt
                     signal(SIGINT,SIG_IGN)
             except KeyboardInterrupt:
                 # Ctrl-C shouldn't crash the kernel
                 io.raw_print("KeyboardInterrupt caught in kernel")
         # stop ignoring sigint, now that we are out of our own loop,
         # we don't want to prevent future code from handling it
         signal(SIGINT, default_int_handler)
         while self.eventloop is not None:
             try:
                 self.eventloop(self)
             except KeyboardInterrupt:
                 # Ctrl-C shouldn't crash the kernel
                 io.raw_print("KeyboardInterrupt caught in kernel")
                 continue
             else:
                 # eventloop exited cleanly, this means we should stop (right?)
                 self.eventloop = None
                 break
 
 
     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 _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)
 
     def execute_request(self, socket, ident, parent):
 
         self.session.send(self.iopub_socket,
                           u'status',
                           {u'execution_state':u'busy'},
                           parent=parent )
         
         try:
             content = parent[u'content']
             code = content[u'code']
             silent = content[u'silent']
         except:
             self.log.error("Got bad msg: ")
             self.log.error("%s", parent)
             return
 
         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)
         else:
             raw_input = lambda prompt='' : self._no_raw_input()
 
         if py3compat.PY3:
             __builtin__.input = raw_input
         else:
             __builtin__.raw_input = raw_input
 
         # Set the parent message of the display hook and out streams.
         shell.displayhook.set_parent(parent)
         shell.display_pub.set_parent(parent)
         sys.stdout.set_parent(parent)
         sys.stderr.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:
             if silent:
                 # run_code uses 'exec' mode, so no displayhook will fire, and it
                 # doesn't call logging or history manipulations.  Print
                 # statements in that code will obviously still execute.
                 shell.run_code(code)
             else:
                 # FIXME: the shell calls the exception handler itself.
                 shell.run_cell(code, store_history=True)
         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'
 
         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)
             # reset after use
             shell._reply_content = None
 
         # 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[u'user_variables'])
             reply_content[u'user_expressions'] = \
                          shell.user_expressions(content[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)
         reply_msg = self.session.send(socket, u'execute_reply',
                                       reply_content, parent, ident=ident)
         self.log.debug("%s", reply_msg)
 
         if reply_msg['content']['status'] == u'error':
             self._abort_queues()
 
         self.session.send(self.iopub_socket,
                           u'status',
                           {u'execution_state':u'idle'},
                           parent=parent )
 
     def complete_request(self, socket, ident, parent):
         txt, matches = self._complete(parent)
         matches = {'matches' : matches,
                    'matched_text' : txt,
                    'status' : 'ok'}
         matches = json_clean(matches)
         completion_msg = self.session.send(socket, 'complete_reply',
                                            matches, parent, ident)
         self.log.debug("%s", completion_msg)
 
     def object_info_request(self, socket, 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(socket, 'object_info_reply',
                                 oinfo, parent, ident)
         self.log.debug("%s", msg)
 
     def history_request(self, socket, 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':
             pattern = parent['content']['pattern']
             hist = self.shell.history_manager.search(pattern, raw=raw,
                                                      output=output) 
 
         else:
             hist = []
         hist = list(hist)
         content = {'history' : hist}
         content = json_clean(content)
         msg = self.session.send(socket, 'history_reply',
                                 content, parent, ident)
         self.log.debug("Sending history reply with %i entries", len(hist))
 
     def connect_request(self, socket, ident, parent):
         if self._recorded_ports is not None:
             content = self._recorded_ports.copy()
         else:
             content = {}
         msg = self.session.send(socket, 'connect_reply',
                                 content, parent, ident)
         self.log.debug("%s", msg)
 
     def shutdown_request(self, ident, parent):
         self.shell.exit_now = True
         self._shutdown_message = self.session.msg(u'shutdown_reply',
                                                   parent['content'], parent)
         sys.exit(0)
 
     #---------------------------------------------------------------------------
     # Engine methods
     #---------------------------------------------------------------------------
 
     def apply_request(self, socket, ident, parent):
         try:
             content = parent[u'content']
             bufs = parent[u'buffers']
             msg_id = parent['header']['msg_id']
             # bound = parent['header'].get('bound', False)
         except:
             self.log.error("Got bad msg: %s", parent, exc_info=True)
             return
         # pyin_msg = self.session.msg(u'pyin',{u'code':code}, parent=parent)
         # self.iopub_stream.send(pyin_msg)
         # self.session.send(self.iopub_stream, u'pyin', {u'code':code},parent=parent)
         sub = {'dependencies_met' : True, 'engine' : self.ident,
                 'started': datetime.now()}
         try:
             # allow for not overriding displayhook
             if hasattr(sys.displayhook, 'set_parent'):
                 sys.displayhook.set_parent(parent)
                 sys.stdout.set_parent(parent)
                 sys.stderr.set_parent(parent)
             # exec "f(*args,**kwargs)" in self.user_ns, self.user_ns
             working = self.user_ns
             # suffix =
             prefix = "_"+str(msg_id).replace("-","")+"_"
 
             f,args,kwargs = unpack_apply_message(bufs, working, copy=False)
             # if bound:
             #     bound_ns = Namespace(working)
             #     args = [bound_ns]+list(args)
 
             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 working,working
                 result = working.get(resultname)
             finally:
                 for key in ns.iterkeys():
                     working.pop(key)
             # if bound:
             #     working.update(bound_ns)
 
             packed_result,buf = serialize_object(result)
             result_buf = [packed_result]+buf
         except:
             exc_content = self._wrap_exception('apply')
             # exc_msg = self.session.msg(u'pyerr', exc_content, parent)
             self.session.send(self.iopub_socket, u'pyerr', exc_content, parent=parent,
-                                ident=asbytes('%s.pyerr'%self.prefix))
+                                ident=py3compat.str_to_bytes('%s.pyerr'%self.prefix))
             reply_content = exc_content
             result_buf = []
 
             if exc_content['ename'] == 'UnmetDependency':
                 sub['dependencies_met'] = False
         else:
             reply_content = {'status' : 'ok'}
 
         # put 'ok'/'error' status in header, for scheduler introspection:
         sub['status'] = reply_content['status']
 
         # flush i/o
         sys.stdout.flush()
         sys.stderr.flush()
         
         reply_msg = self.session.send(socket, u'apply_reply', reply_content,
                     parent=parent, ident=ident,buffers=result_buf, subheader=sub)
 
     #---------------------------------------------------------------------------
     # Control messages
     #---------------------------------------------------------------------------
 
     def abort_request(self, socket, 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(socket, 'abort_reply', content=content,
                 parent=parent, ident=ident)
         self.log.debug("%s", reply_msg)
 
     def clear_request(self, socket, idents, parent):
         """Clear our namespace."""
         self.user_ns = {}
         msg = self.session.send(socket, 'clear_reply', ident=idents, parent=parent,
                 content = dict(status='ok'))
         self._initial_exec_lines()
 
 
     #---------------------------------------------------------------------------
     # Protected interface
     #---------------------------------------------------------------------------
 
     def _abort_queues(self):
         for socket in self.shell_sockets:
             if socket:
                 self._abort_queue(socket)
 
     def _abort_queue(self, socket):
         while True:
             idents,msg = self.session.recv(socket, 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'
             # reply_msg = self.session.msg(reply_type, {'status' : 'aborted'}, msg)
             # self.reply_socket.send(ident,zmq.SNDMORE)
             # self.reply_socket.send_json(reply_msg)
             reply_msg = self.session.send(socket, reply_type,
                         content={'status' : 'aborted'}, 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.
             time.sleep(0.05)
 
 
     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()
 
         # 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)
             else:
                 break
         try:
             value = 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 _object_info(self, context):
         symbol, leftover = self._symbol_from_context(context)
         if symbol is not None and not leftover:
             doc = getattr(symbol, '__doc__', '')
         else:
             doc = ''
         object_info = dict(docstring = doc)
         return object_info
 
     def _symbol_from_context(self, context):
         if not context:
             return None, context
 
         base_symbol_string = context[0]
         symbol = self.shell.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, []
 
     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.shell_socket, self._shutdown_message)
             self.session.send(self.iopub_socket, self._shutdown_message)
             self.log.debug("%s", self._shutdown_message)
             # A very short sleep to give zmq time to flush its message buffers
             # before Python truly shuts down.
             time.sleep(0.01)
 
 #-----------------------------------------------------------------------------
 # Aliases and Flags for the IPKernelApp
 #-----------------------------------------------------------------------------
 
 flags = dict(kernel_flags)
 flags.update(shell_flags)
 
 addflag = lambda *args: flags.update(boolean_flag(*args))
 
 flags['pylab'] = (
     {'IPKernelApp' : {'pylab' : 'auto'}},
     """Pre-load matplotlib and numpy for interactive use with
     the default matplotlib backend."""
 )
 
 aliases = dict(kernel_aliases)
 aliases.update(shell_aliases)
 
 # it's possible we don't want short aliases for *all* of these:
 aliases.update(dict(
     pylab='IPKernelApp.pylab',
 ))
 
 #-----------------------------------------------------------------------------
 # The IPKernelApp class
 #-----------------------------------------------------------------------------
 
 class IPKernelApp(KernelApp, InteractiveShellApp):
     name = 'ipkernel'
 
     aliases = Dict(aliases)
     flags = Dict(flags)
     classes = [Kernel, ZMQInteractiveShell, ProfileDir, Session]
     
     # configurables
     pylab = CaselessStrEnum(['tk', 'qt', 'wx', 'gtk', 'osx', 'inline', 'auto'],
         config=True,
         help="""Pre-load matplotlib and numpy for interactive use,
         selecting a particular matplotlib backend and loop integration.
         """
     )
     
     @catch_config_error
     def initialize(self, argv=None):
         super(IPKernelApp, self).initialize(argv)
         self.init_path()
         self.init_shell()
         self.init_extensions()
         self.init_code()
 
     def init_kernel(self):
 
         kernel = Kernel(config=self.config, session=self.session,
                                 shell_sockets=[self.shell_socket],
                                 iopub_socket=self.iopub_socket,
                                 stdin_socket=self.stdin_socket,
                                 log=self.log,
                                 profile_dir=self.profile_dir,
         )
         self.kernel = kernel
         kernel.record_ports(self.ports)
         shell = kernel.shell
         if self.pylab:
             try:
                 gui, backend = pylabtools.find_gui_and_backend(self.pylab)
                 shell.enable_pylab(gui, import_all=self.pylab_import_all)
             except Exception:
                 self.log.error("Pylab initialization failed", exc_info=True)
                 # print exception straight to stdout, because normally 
                 # _showtraceback associates the reply with an execution, 
                 # which means frontends will never draw it, as this exception 
                 # is not associated with any execute request.
                 
                 # replace pyerr-sending traceback with stdout
                 _showtraceback = shell._showtraceback
                 def print_tb(etype, evalue, stb):
                     print ("Error initializing pylab, pylab mode will not "
                            "be active", file=io.stderr)
                     print (shell.InteractiveTB.stb2text(stb), file=io.stdout)
                 shell._showtraceback = print_tb
                 
                 # send the traceback over stdout
                 shell.showtraceback(tb_offset=0)
                 
                 # restore proper _showtraceback method
                 shell._showtraceback = _showtraceback
                 
 
     def init_shell(self):
         self.shell = self.kernel.shell
         self.shell.configurables.append(self)
 
 
 #-----------------------------------------------------------------------------
 # Kernel main and launch functions
 #-----------------------------------------------------------------------------
 
 def launch_kernel(*args, **kwargs):
     """Launches a localhost IPython kernel, binding to the specified ports.
 
     This function simply calls entry_point.base_launch_kernel with the right
     first command to start an ipkernel.  See base_launch_kernel for arguments.
 
     Returns
     -------
     A tuple of form:
         (kernel_process, shell_port, iopub_port, stdin_port, hb_port)
     where kernel_process is a Popen object and the ports are integers.
     """
     return base_launch_kernel('from IPython.zmq.ipkernel import main; main()',
                               *args, **kwargs)
 
 
 def embed_kernel(module=None, local_ns=None, **kwargs):
     """Embed and start an IPython kernel in a given scope.
     
     Parameters
     ----------
     module : ModuleType, optional
         The module to load into IPython globals (default: caller)
     local_ns : dict, optional
         The namespace to load into IPython user namespace (default: caller)
     
     kwargs : various, optional
         Further keyword args are relayed to the KernelApp constructor,
         allowing configuration of the Kernel.  Will only have an effect
         on the first embed_kernel call for a given process.
     
     """
     # get the app if it exists, or set it up if it doesn't
     if IPKernelApp.initialized():
         app = IPKernelApp.instance()
     else:
         app = IPKernelApp.instance(**kwargs)
         app.initialize([])
 
     # load the calling scope if not given
     (caller_module, caller_locals) = extract_module_locals(1)
     if module is None:
         module = caller_module
     if local_ns is None:
         local_ns = caller_locals
     
     app.kernel.user_module = module
     app.kernel.user_ns = local_ns
     app.start()
 
 def main():
     """Run an IPKernel as an application"""
     app = IPKernelApp.instance()
     app.initialize()
     app.start()
 
 
 if __name__ == '__main__':
     main()