Messing with newlines.

IPython/kernel/engineservice.py

@@ -1,903 +1,904 b''
 # encoding: utf-8
 # -*- test-case-name: IPython.kernel.tests.test_engineservice -*-
 
 """A Twisted Service Representation of the IPython core.
 
 The IPython Core exposed to the network is called the Engine.  Its
 representation in Twisted in the EngineService.  Interfaces and adapters
 are used to abstract out the details of the actual network protocol used.
 The EngineService is an Engine that knows nothing about the actual protocol
 used.
 
 The EngineService is exposed with various network protocols in modules like:
 
 enginepb.py
 enginevanilla.py
 
 As of 12/12/06 the classes in this module have been simplified greatly.  It was 
 felt that we had over-engineered things.  To improve the maintainability of the
 code we have taken out the ICompleteEngine interface and the completeEngine
 method that automatically added methods to engines.
 
 """
 
 __docformat__ = "restructuredtext en"
 
 #-------------------------------------------------------------------------------
 #  Copyright (C) 2008  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
 #-------------------------------------------------------------------------------
 
 import os, sys, copy
 import cPickle as pickle
 from new import instancemethod
 
 from twisted.application import service
 from twisted.internet import defer, reactor
 from twisted.python import log, failure, components
 import zope.interface as zi
 
 from IPython.kernel.core.interpreter import Interpreter
 from IPython.kernel import newserialized, error, util
 from IPython.kernel.util import printer
 from IPython.kernel.twistedutil import gatherBoth, DeferredList
 from IPython.kernel import codeutil
 
 
 #-------------------------------------------------------------------------------
 # Interface specification for the Engine
 #-------------------------------------------------------------------------------
 
 class IEngineCore(zi.Interface):
     """The minimal required interface for the IPython Engine.
     
     This interface provides a formal specification of the IPython core.
     All these methods should return deferreds regardless of what side of a
     network connection they are on.
     
     In general, this class simply wraps a shell class and wraps its return
     values as Deferred objects.  If the underlying shell class method raises
     an exception, this class should convert it to a twisted.failure.Failure
     that will be propagated along the Deferred's errback chain.
     
     In addition, Failures are aggressive.  By this, we mean that if a method
     is performing multiple actions (like pulling multiple object) if any
     single one fails, the entire method will fail with that Failure.  It is
     all or nothing. 
     """
     
     id = zi.interface.Attribute("the id of the Engine object")
     properties = zi.interface.Attribute("A dict of properties of the Engine")
     
     def execute(lines):
         """Execute lines of Python code.
         
         Returns a dictionary with keys (id, number, stdin, stdout, stderr)
         upon success.
         
         Returns a failure object if the execution of lines raises an exception.
         """
     
     def push(namespace):
         """Push dict namespace into the user's namespace.
         
         Returns a deferred to None or a failure.
         """
     
     def pull(keys):
         """Pulls values out of the user's namespace by keys.
         
         Returns a deferred to a tuple objects or a single object.
         
         Raises NameError if any one of objects doess not exist.
         """
     
     def push_function(namespace):
         """Push a dict of key, function pairs into the user's namespace.
         
         Returns a deferred to None or a failure."""
     
     def pull_function(keys):
         """Pulls functions out of the user's namespace by keys.
         
         Returns a deferred to a tuple of functions or a single function.
         
         Raises NameError if any one of the functions does not exist.
         """
     
     def get_result(i=None):
         """Get the stdin/stdout/stderr of command i.
         
         Returns a deferred to a dict with keys
         (id, number, stdin, stdout, stderr).
         
         Raises IndexError if command i does not exist.
         Raises TypeError if i in not an int.
         """
     
     def reset():
         """Reset the shell.
         
         This clears the users namespace.  Won't cause modules to be
         reloaded.  Should also re-initialize certain variables like id.
         """
     
     def kill():
         """Kill the engine by stopping the reactor."""
     
     def keys():
         """Return the top level variables in the users namspace.
         
         Returns a deferred to a dict."""
     
 
 class IEngineSerialized(zi.Interface):
     """Push/Pull methods that take Serialized objects.  
     
     All methods should return deferreds.
     """
     
     def push_serialized(namespace):
         """Push a dict of keys and Serialized objects into the user's namespace."""
     
     def pull_serialized(keys):
         """Pull objects by key from the user's namespace as Serialized.
         
         Returns a list of or one Serialized.
         
         Raises NameError is any one of the objects does not exist.
         """
     
 
 class IEngineProperties(zi.Interface):
     """Methods for access to the properties object of an Engine"""
     
     properties = zi.Attribute("A StrictDict object, containing the properties")
     
     def set_properties(properties):
         """set properties by key and value"""
     
     def get_properties(keys=None):
         """get a list of properties by `keys`, if no keys specified, get all"""
     
     def del_properties(keys):
         """delete properties by `keys`"""
     
     def has_properties(keys):
         """get a list of bool values for whether `properties` has `keys`"""
 
     def clear_properties():
         """clear the properties dict"""
     
 class IEngineBase(IEngineCore, IEngineSerialized, IEngineProperties):
     """The basic engine interface that EngineService will implement.
     
     This exists so it is easy to specify adapters that adapt to and from the
     API that the basic EngineService implements.
     """
     pass
 
 class IEngineQueued(IEngineBase):
     """Interface for adding a queue to an IEngineBase.  
     
     This interface extends the IEngineBase interface to add methods for managing
     the engine's queue.  The implicit details of this interface are that the 
     execution of all methods declared in IEngineBase should appropriately be
     put through a queue before execution. 
     
     All methods should return deferreds.
     """
     
     def clear_queue():
         """Clear the queue."""
     
     def queue_status():
         """Get the queued and pending commands in the queue."""
     
     def register_failure_observer(obs):
         """Register an observer of pending Failures.
         
         The observer must implement IFailureObserver.
         """
     
     def unregister_failure_observer(obs):
         """Unregister an observer of pending Failures."""
     
 
 class IEngineThreaded(zi.Interface):
     """A place holder for threaded commands.  
     
     All methods should return deferreds.
     """
     pass
 
 
 #-------------------------------------------------------------------------------
 # Functions and classes to implement the EngineService
 #-------------------------------------------------------------------------------
 
 
 class StrictDict(dict):
     """This is a strict copying dictionary for use as the interface to the 
     properties of an Engine.
 
     :IMPORTANT:
         This object copies the values you set to it, and returns copies to you
         when you request them.  The only way to change properties os explicitly
         through the setitem and getitem of the dictionary interface.
 
     Example:
         >>> e = get_engine(id)
         >>> L = [1,2,3]
         >>> e.properties['L'] = L
         >>> L == e.properties['L']
         True
         >>> L.append(99)
         >>> L == e.properties['L']
         False
         
         Note that getitem copies, so calls to methods of objects do not affect
         the properties, as seen here:
         
         >>> e.properties[1] = range(2)
         >>> print e.properties[1]
         [0, 1]
         >>> e.properties[1].append(2)
         >>> print e.properties[1]
         [0, 1]
     """
     def __init__(self, *args, **kwargs):
         dict.__init__(self, *args, **kwargs)
         self.modified = True
     
     def __getitem__(self, key):
         return copy.deepcopy(dict.__getitem__(self, key))
     
     def __setitem__(self, key, value):
         # check if this entry is valid for transport around the network
         # and copying
         try:
             pickle.dumps(key, 2)
             pickle.dumps(value, 2)
             newvalue = copy.deepcopy(value)
         except:
             raise error.InvalidProperty(value)
         dict.__setitem__(self, key, newvalue)
         self.modified = True
     
     def __delitem__(self, key):
         dict.__delitem__(self, key)
         self.modified = True
     
     def update(self, dikt):
         for k,v in dikt.iteritems():
             self[k] = v
     
     def pop(self, key):
         self.modified = True
         return dict.pop(self, key)
     
     def popitem(self):
         self.modified = True
         return dict.popitem(self)
     
     def clear(self):
         self.modified = True
         dict.clear(self)
     
     def subDict(self, *keys):
         d = {}
         for key in keys:
             d[key] = self[key]
         return d
     
 
 
 class EngineAPI(object):
     """This is the object through which the user can edit the `properties`
     attribute of an Engine.
     The Engine Properties object copies all object in and out of itself.
     See the EngineProperties object for details.
     """
     _fix=False
     def __init__(self, id):
         self.id = id
         self.properties = StrictDict()
         self._fix=True
     
     def __setattr__(self, k,v):
         if self._fix:
             raise error.KernelError("I am protected!")
         else:
             object.__setattr__(self, k, v)
     
     def __delattr__(self, key):
         raise error.KernelError("I am protected!")
     
 
 _apiDict = {}
 
 def get_engine(id):
     """Get the Engine API object, whcih currently just provides the properties 
     object, by ID"""
     global _apiDict
     if not _apiDict.get(id):
         _apiDict[id] = EngineAPI(id)
     return _apiDict[id]
 
 def drop_engine(id):
     """remove an engine"""
     global _apiDict
     if _apiDict.has_key(id):
         del _apiDict[id]
 
 class EngineService(object, service.Service):
     """Adapt a IPython shell into a IEngine implementing Twisted Service."""
     
     zi.implements(IEngineBase)
     name = 'EngineService'
     
     def __init__(self, shellClass=Interpreter, mpi=None):
         """Create an EngineService.
         
         shellClass: something that implements IInterpreter or core1
         mpi:        an mpi module that has rank and size attributes
         """
         self.shellClass = shellClass
         self.shell = self.shellClass()
         self.mpi = mpi
         self.id = None
         self.properties = get_engine(self.id).properties
         if self.mpi is not None:
             log.msg("MPI started with rank = %i and size = %i" % 
                 (self.mpi.rank, self.mpi.size))
             self.id = self.mpi.rank
         self._seedNamespace()
     
     # Make id a property so that the shell can get the updated id
         
     def _setID(self, id):
         self._id = id
         self.properties = get_engine(id).properties
         self.shell.push({'id': id})
     
     def _getID(self):
         return self._id
     
     id = property(_getID, _setID)
     
     def _seedNamespace(self):
         self.shell.push({'mpi': self.mpi, 'id' : self.id})
     
     def executeAndRaise(self, msg, callable, *args, **kwargs):
         """Call a method of self.shell and wrap any exception."""
         d = defer.Deferred()
         try:
             result = callable(*args, **kwargs)
         except:
             # This gives the following:
             # et=exception class
             # ev=exception class instance
             # tb=traceback object
             et,ev,tb = sys.exc_info()
             # This call adds attributes to the exception value
             et,ev,tb = self.shell.formatTraceback(et,ev,tb,msg)
             # Add another attribute
             ev._ipython_engine_info = msg
             f = failure.Failure(ev,et,None)
             d.errback(f)
         else:
             d.callback(result)
             
         return d
     
     
     # The IEngine methods.  See the interface for documentation.
     
+    @profile
     def execute(self, lines):
         msg = {'engineid':self.id,
                'method':'execute',
                'args':[lines]}
         d = self.executeAndRaise(msg, self.shell.execute, lines)
         d.addCallback(self.addIDToResult)
         return d
     
     def addIDToResult(self, result):
         result['id'] = self.id
         return result
     
     def push(self, namespace):
         msg = {'engineid':self.id,
                'method':'push',
                'args':[repr(namespace.keys())]}
         d = self.executeAndRaise(msg, self.shell.push, namespace)
         return d
     
     def pull(self, keys):
         msg = {'engineid':self.id,
                'method':'pull',
                'args':[repr(keys)]}
         d = self.executeAndRaise(msg, self.shell.pull, keys)
         return d
     
     def push_function(self, namespace):
         msg = {'engineid':self.id,
                'method':'push_function',
                'args':[repr(namespace.keys())]}
         d = self.executeAndRaise(msg, self.shell.push_function, namespace)
         return d
     
     def pull_function(self, keys):
         msg = {'engineid':self.id,
                'method':'pull_function',
                'args':[repr(keys)]}
         d = self.executeAndRaise(msg, self.shell.pull_function, keys)
         return d
     
     def get_result(self, i=None):
         msg = {'engineid':self.id,
                'method':'get_result',
                'args':[repr(i)]}
         d = self.executeAndRaise(msg, self.shell.getCommand, i)
         d.addCallback(self.addIDToResult)
         return d
     
     def reset(self):
         msg = {'engineid':self.id,
                'method':'reset',
                'args':[]}
         del self.shell
         self.shell = self.shellClass()
         self.properties.clear()
         d = self.executeAndRaise(msg, self._seedNamespace)
         return d
     
     def kill(self):
         drop_engine(self.id)
         try:
             reactor.stop()
         except RuntimeError:
             log.msg('The reactor was not running apparently.')
             return defer.fail()
         else:
             return defer.succeed(None)
     
     def keys(self):
         """Return a list of variables names in the users top level namespace.
         
         This used to return a dict of all the keys/repr(values) in the 
         user's namespace.  This was too much info for the ControllerService
         to handle so it is now just a list of keys.
         """
         
         remotes = []
         for k in self.shell.user_ns.iterkeys():
             if k not in ['__name__', '_ih', '_oh', '__builtins__',
                          'In', 'Out', '_', '__', '___', '__IP', 'input', 'raw_input']:
                 remotes.append(k)
         return defer.succeed(remotes)
     
     def set_properties(self, properties):
         msg = {'engineid':self.id,
                'method':'set_properties',
                'args':[repr(properties.keys())]}
         return self.executeAndRaise(msg, self.properties.update, properties)
     
     def get_properties(self, keys=None):
         msg = {'engineid':self.id,
                'method':'get_properties',
                'args':[repr(keys)]}
         if keys is None:
             keys = self.properties.keys()
         return self.executeAndRaise(msg, self.properties.subDict, *keys)
     
     def _doDel(self, keys):
         for key in keys:
             del self.properties[key]
     
     def del_properties(self, keys):
         msg = {'engineid':self.id,
                'method':'del_properties',
                'args':[repr(keys)]}
         return self.executeAndRaise(msg, self._doDel, keys)
     
     def _doHas(self, keys):
         return [self.properties.has_key(key) for key in keys]
     
     def has_properties(self, keys):
         msg = {'engineid':self.id,
                'method':'has_properties',
                'args':[repr(keys)]}
         return self.executeAndRaise(msg, self._doHas, keys)
     
     def clear_properties(self):
         msg = {'engineid':self.id,
                'method':'clear_properties',
                'args':[]}
         return self.executeAndRaise(msg, self.properties.clear)
     
     def push_serialized(self, sNamespace):
         msg = {'engineid':self.id,
                'method':'push_serialized',
                'args':[repr(sNamespace.keys())]}       
         ns = {}
         for k,v in sNamespace.iteritems():
             try:
                 unserialized = newserialized.IUnSerialized(v)
                 ns[k] = unserialized.getObject()
             except:
                 return defer.fail()
         return self.executeAndRaise(msg, self.shell.push, ns)
     
     def pull_serialized(self, keys):
         msg = {'engineid':self.id,
                'method':'pull_serialized',
                'args':[repr(keys)]}
         if isinstance(keys, str):
             keys = [keys]
         if len(keys)==1:
             d = self.executeAndRaise(msg, self.shell.pull, keys)
             d.addCallback(newserialized.serialize)
             return d
         elif len(keys)>1:
             d = self.executeAndRaise(msg, self.shell.pull, keys)         
             @d.addCallback
             def packThemUp(values):
                 serials = []
                 for v in values:
                     try:
                         serials.append(newserialized.serialize(v))
                     except:
                         return defer.fail(failure.Failure())
                 return serials
             return packThemUp
 
 
 def queue(methodToQueue):
     def queuedMethod(this, *args, **kwargs):
         name = methodToQueue.__name__
         return this.submitCommand(Command(name, *args, **kwargs))
     return queuedMethod
     
 class QueuedEngine(object):
     """Adapt an IEngineBase to an IEngineQueued by wrapping it.
     
     The resulting object will implement IEngineQueued which extends
     IEngineCore which extends (IEngineBase, IEngineSerialized). 
     
     This seems like the best way of handling it, but I am not sure.  The
     other option is to have the various base interfaces be used like
     mix-in intefaces.  The problem I have with this is adpatation is
     more difficult and complicated because there can be can multiple
     original and final Interfaces. 
     """
     
     zi.implements(IEngineQueued)
     
     def __init__(self, engine):
         """Create a QueuedEngine object from an engine
         
         engine:       An implementor of IEngineCore and IEngineSerialized
         keepUpToDate: whether to update the remote status when the 
                       queue is empty.  Defaults to False.
         """
         
         # This is the right way to do these tests rather than 
         # IEngineCore in list(zi.providedBy(engine)) which will only 
         # picks of the interfaces that are directly declared by engine.
         assert IEngineBase.providedBy(engine), \
             "engine passed to QueuedEngine doesn't provide IEngineBase"
             
         self.engine = engine
         self.id = engine.id
         self.queued = []
         self.history = {}
         self.engineStatus = {}
         self.currentCommand = None
         self.failureObservers = []
     
     def _get_properties(self):
         return self.engine.properties
     
     properties = property(_get_properties, lambda self, _: None)
     # Queue management methods.  You should not call these directly
     
     def submitCommand(self, cmd):
         """Submit command to queue."""
         
         d = defer.Deferred()
         cmd.setDeferred(d)
         if self.currentCommand is not None:
             if self.currentCommand.finished:
                 # log.msg("Running command immediately: %r" % cmd)
                 self.currentCommand = cmd
                 self.runCurrentCommand()
             else:  # command is still running  
                 # log.msg("Command is running: %r" % self.currentCommand)
                 # log.msg("Queueing: %r" % cmd)
                 self.queued.append(cmd)
         else:
             # log.msg("No current commands, running: %r" % cmd)
             self.currentCommand = cmd
             self.runCurrentCommand()
         return d
     
     def runCurrentCommand(self):
         """Run current command."""
         
         cmd = self.currentCommand
         f = getattr(self.engine, cmd.remoteMethod, None)
         if f:
             d = f(*cmd.args, **cmd.kwargs)
             if cmd.remoteMethod is 'execute':
                 d.addCallback(self.saveResult)
             d.addCallback(self.finishCommand)
             d.addErrback(self.abortCommand)
         else:
             return defer.fail(AttributeError(cmd.remoteMethod))
     
     def _flushQueue(self):
         """Pop next command in queue and run it."""
         
         if len(self.queued) > 0:
             self.currentCommand = self.queued.pop(0)
             self.runCurrentCommand()
     
     def saveResult(self, result):
         """Put the result in the history."""
         self.history[result['number']] = result
         return result
     
     def finishCommand(self, result):
         """Finish currrent command."""
         
         # The order of these commands is absolutely critical.
         self.currentCommand.handleResult(result)
         self.currentCommand.finished = True
         self._flushQueue()
         return result
     
     def abortCommand(self, reason):
         """Abort current command.
         
         This eats the Failure but first passes it onto the Deferred that the 
         user has.
         
         It also clear out the queue so subsequence commands don't run.
         """
 
         # The order of these 3 commands is absolutely critical.  The currentCommand
         # must first be marked as finished BEFORE the queue is cleared and before
         # the current command is sent the failure.
         # Also, the queue must be cleared BEFORE the current command is sent the Failure
         # otherwise the errback chain could trigger new commands to be added to the 
         # queue before we clear it.  We should clear ONLY the commands that were in
         # the queue when the error occured. 
         self.currentCommand.finished = True
         s = "%r %r %r" % (self.currentCommand.remoteMethod, self.currentCommand.args, self.currentCommand.kwargs)
         self.clear_queue(msg=s)
         self.currentCommand.handleError(reason)
         
         return None
     
     #---------------------------------------------------------------------------
     # IEngineCore methods
     #---------------------------------------------------------------------------
     
     @queue
     def execute(self, lines):
         pass
 
     @queue
     def push(self, namespace):
         pass      
     
     @queue
     def pull(self, keys):
         pass
         
     @queue
     def push_function(self, namespace):
         pass      
     
     @queue
     def pull_function(self, keys):
         pass        
 
     def get_result(self, i=None):
         if i is None:
             i = max(self.history.keys()+[None])
 
         cmd = self.history.get(i, None)
         # Uncomment this line to disable chaching of results
         #cmd = None
         if cmd is None:
             return self.submitCommand(Command('get_result', i))
         else:
             return defer.succeed(cmd)
         
     def reset(self):
         self.clear_queue()
         self.history = {}  # reset the cache - I am not sure we should do this
         return self.submitCommand(Command('reset'))
     
     def kill(self):
         self.clear_queue()
         return self.submitCommand(Command('kill'))
     
     @queue
     def keys(self):
         pass
     
     #---------------------------------------------------------------------------
     # IEngineSerialized methods
     #---------------------------------------------------------------------------
 
     @queue
     def push_serialized(self, namespace):
         pass
         
     @queue
     def pull_serialized(self, keys):
         pass
     
     #---------------------------------------------------------------------------
     # IEngineProperties methods
     #---------------------------------------------------------------------------
 
     @queue
     def set_properties(self, namespace):
         pass
         
     @queue
     def get_properties(self, keys=None):
         pass
     
     @queue
     def del_properties(self, keys):
         pass
     
     @queue
     def has_properties(self, keys):
         pass
     
     @queue
     def clear_properties(self):
         pass
     
     #---------------------------------------------------------------------------
     # IQueuedEngine methods
     #---------------------------------------------------------------------------
     
     def clear_queue(self, msg=''):
         """Clear the queue, but doesn't cancel the currently running commmand."""
         
         for cmd in self.queued:
             cmd.deferred.errback(failure.Failure(error.QueueCleared(msg)))
         self.queued = []
         return defer.succeed(None)
     
     def queue_status(self):
         if self.currentCommand is not None:
             if self.currentCommand.finished:
                 pending = repr(None)
             else:
                 pending = repr(self.currentCommand)
         else:
             pending = repr(None)
         dikt = {'queue':map(repr,self.queued), 'pending':pending}
         return defer.succeed(dikt)
         
     def register_failure_observer(self, obs):
         self.failureObservers.append(obs)
     
     def unregister_failure_observer(self, obs):
         self.failureObservers.remove(obs)
     
 
 # Now register QueuedEngine as an adpater class that makes an IEngineBase into a
 # IEngineQueued.  
 components.registerAdapter(QueuedEngine, IEngineBase, IEngineQueued)
     
 
 class Command(object):
     """A command object that encapslates queued commands.
     
     This class basically keeps track of a command that has been queued
     in a QueuedEngine.  It manages the deferreds and hold the method to be called
     and the arguments to that method.
     """
     
     
     def __init__(self, remoteMethod, *args, **kwargs):
         """Build a new Command object."""
         
         self.remoteMethod = remoteMethod
         self.args = args
         self.kwargs = kwargs
         self.finished = False
     
     def setDeferred(self, d):
         """Sets the deferred attribute of the Command."""  
           
         self.deferred = d
     
     def __repr__(self):
         if not self.args:
             args = ''
         else:
             args = str(self.args)[1:-2]  #cut off (...,)
         for k,v in self.kwargs.iteritems():
             if args:
                 args += ', '
             args += '%s=%r' %(k,v)
         return "%s(%s)" %(self.remoteMethod, args)
     
     def handleResult(self, result):
         """When the result is ready, relay it to self.deferred."""
         
         self.deferred.callback(result)
     
     def handleError(self, reason):
         """When an error has occured, relay it to self.deferred."""
         
         self.deferred.errback(reason)
 
 class ThreadedEngineService(EngineService):
     """An EngineService subclass that defers execute commands to a separate 
     thread.
     
     ThreadedEngineService uses twisted.internet.threads.deferToThread to 
     defer execute requests to a separate thread. GUI frontends may want to 
     use ThreadedEngineService as the engine in an 
     IPython.frontend.frontendbase.FrontEndBase subclass to prevent
     block execution from blocking the GUI thread.
     """
     
     zi.implements(IEngineBase)
 
     def __init__(self, shellClass=Interpreter, mpi=None):
         EngineService.__init__(self, shellClass, mpi)
     
     def wrapped_execute(self, msg, lines):
         """Wrap self.shell.execute to add extra information to tracebacks"""
         
         try:
             result = self.shell.execute(lines)
         except Exception,e:
             # This gives the following:
             # et=exception class
             # ev=exception class instance
             # tb=traceback object
             et,ev,tb = sys.exc_info()
             # This call adds attributes to the exception value
             et,ev,tb = self.shell.formatTraceback(et,ev,tb,msg)
             # Add another attribute
             
             # Create a new exception with the new attributes
             e = et(ev._ipython_traceback_text)
             e._ipython_engine_info = msg
             
             # Re-raise
             raise e
         
         return result
     
     
     def execute(self, lines):
         # Only import this if we are going to use this class
         from twisted.internet import threads
         
         msg = {'engineid':self.id,
                'method':'execute',
                'args':[lines]}
         
         d = threads.deferToThread(self.wrapped_execute, msg, lines)
         d.addCallback(self.addIDToResult)
         return d

IPython/kernel/multiengine.py

@@ -1,753 +1,754 b''
 # encoding: utf-8
 # -*- test-case-name: IPython.kernel.test.test_multiengine -*-
 
 """Adapt the IPython ControllerServer to IMultiEngine.
 
 This module provides classes that adapt a ControllerService to the 
 IMultiEngine interface.  This interface is a basic interactive interface
 for working with a set of engines where it is desired to have explicit 
 access to each registered engine.  
 
 The classes here are exposed to the network in files like:
 
 * multienginevanilla.py
 * multienginepb.py
 """
 
 __docformat__ = "restructuredtext en"
 
 #-------------------------------------------------------------------------------
 #  Copyright (C) 2008  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
 #-------------------------------------------------------------------------------
 
 from new import instancemethod
 from types import FunctionType
 
 from twisted.application import service
 from twisted.internet import defer, reactor
 from twisted.python import log, components, failure
 from zope.interface import Interface, implements, Attribute
 
 from IPython.tools import growl
 from IPython.kernel.util import printer
 from IPython.kernel.twistedutil import gatherBoth
 from IPython.kernel import map as Map
 from IPython.kernel import error
 from IPython.kernel.pendingdeferred import PendingDeferredManager, two_phase
 from IPython.kernel.controllerservice import \
     ControllerAdapterBase, \
     ControllerService, \
     IControllerBase
 
 
 #-------------------------------------------------------------------------------
 # Interfaces for the MultiEngine representation of a controller
 #-------------------------------------------------------------------------------
 
 class IEngineMultiplexer(Interface):
     """Interface to multiple engines implementing IEngineCore/Serialized/Queued.
     
     This class simply acts as a multiplexer of methods that are in the 
     various IEngines* interfaces.  Thus the methods here are jut like those 
     in the IEngine* interfaces, but with an extra first argument, targets.  
     The targets argument can have the following forms:
     
     * targets = 10            # Engines are indexed by ints
     * targets = [0,1,2,3]     # A list of ints
     * targets = 'all'         # A string to indicate all targets
     
     If targets is bad in any way, an InvalidEngineID will be raised.  This
     includes engines not being registered.
     
     All IEngineMultiplexer multiplexer methods must return a Deferred to a list 
     with length equal to the number of targets.  The elements of the list will 
     correspond to the return of the corresponding IEngine method.
     
     Failures are aggressive, meaning that if an action fails for any target, 
     the overall action will fail immediately with that Failure.
     
     :Parameters:
         targets : int, list of ints, or 'all'
             Engine ids the action will apply to.
     
     :Returns: Deferred to a list of results for each engine.
     
     :Exception:
         InvalidEngineID
             If the targets argument is bad or engines aren't registered.
         NoEnginesRegistered
             If there are no engines registered and targets='all'
     """
         
     #---------------------------------------------------------------------------
     # Mutiplexed methods
     #---------------------------------------------------------------------------
      
     def execute(lines, targets='all'):
         """Execute lines of Python code on targets.
               
         See the class docstring for information about targets and possible
         exceptions this method can raise.
               
         :Parameters:
             lines : str
                 String of python code to be executed on targets.
         """
             
     def push(namespace, targets='all'):
         """Push dict namespace into the user's namespace on targets.
         
         See the class docstring for information about targets and possible
         exceptions this method can raise.
         
         :Parameters:
             namspace : dict
                 Dict of key value pairs to be put into the users namspace.
         """
         
     def pull(keys, targets='all'):
         """Pull values out of the user's namespace on targets by keys.
         
         See the class docstring for information about targets and possible
         exceptions this method can raise.
         
         :Parameters:
             keys : tuple of strings
                 Sequence of keys to be pulled from user's namespace.
         """
           
     def push_function(namespace, targets='all'):
         """"""
         
     def pull_function(keys, targets='all'):
         """"""
                 
     def get_result(i=None, targets='all'):
         """Get the result for command i from targets.
         
         See the class docstring for information about targets and possible
         exceptions this method can raise.
         
         :Parameters:
             i : int or None
                 Command index or None to indicate most recent command.                
         """
         
     def reset(targets='all'):
         """Reset targets.
         
         This clears the users namespace of the Engines, but won't cause
         modules to be reloaded.
         """
         
     def keys(targets='all'):
         """Get variable names defined in user's namespace on targets."""
         
     def kill(controller=False, targets='all'):
         """Kill the targets Engines and possibly the controller.
         
         :Parameters:
             controller : boolean
                 Should the controller be killed as well.  If so all the 
                 engines will be killed first no matter what targets is.
         """
         
     def push_serialized(namespace, targets='all'):
         """Push a namespace of Serialized objects to targets.
         
         :Parameters:
             namespace : dict
                 A dict whose keys are the variable names and whose values
                 are serialized version of the objects.
         """
         
     def pull_serialized(keys, targets='all'):
         """Pull Serialized objects by keys from targets.
         
         :Parameters:
             keys : tuple of strings
                 Sequence of variable names to pull as serialized objects.
         """
         
     def clear_queue(targets='all'):
         """Clear the queue of pending command for targets."""
         
     def queue_status(targets='all'):
         """Get the status of the queue on the targets."""
     
     def set_properties(properties, targets='all'):
         """set properties by key and value"""
     
     def get_properties(keys=None, targets='all'):
         """get a list of properties by `keys`, if no keys specified, get all"""
     
     def del_properties(keys, targets='all'):
         """delete properties by `keys`"""
     
     def has_properties(keys, targets='all'):
         """get a list of bool values for whether `properties` has `keys`"""
     
     def clear_properties(targets='all'):
         """clear the properties dict"""
 
 
 class IMultiEngine(IEngineMultiplexer):
     """A controller that exposes an explicit interface to all of its engines.
     
     This is the primary inteface for interactive usage.
     """
     
     def get_ids():
         """Return list of currently registered ids.
                 
         :Returns:  A Deferred to a list of registered engine ids.
         """
 
 
 
 #-------------------------------------------------------------------------------
 # Implementation of the core MultiEngine classes
 #-------------------------------------------------------------------------------
 
 class MultiEngine(ControllerAdapterBase):
     """The representation of a ControllerService as a IMultiEngine.
     
     Although it is not implemented currently, this class would be where a
     client/notification API is implemented.  It could inherit from something
     like results.NotifierParent and then use the notify method to send
     notifications.
     """
     
     implements(IMultiEngine)
     
     def __init(self, controller):
         ControllerAdapterBase.__init__(self, controller)
     
     #---------------------------------------------------------------------------
     # Helper methods
     #---------------------------------------------------------------------------
     
     def engineList(self, targets):
         """Parse the targets argument into a list of valid engine objects.
         
         :Parameters:
             targets : int, list of ints or 'all'
                 The targets argument to be parsed.
                 
         :Returns: List of engine objects.
         
         :Exception:
             InvalidEngineID
                 If targets is not valid or if an engine is not registered.
         """
         if isinstance(targets, int):
             if targets not in self.engines.keys():
                 log.msg("Engine with id %i is not registered" % targets)
                 raise error.InvalidEngineID("Engine with id %i is not registered" % targets)
             else: 
                 return [self.engines[targets]]
         elif isinstance(targets, (list, tuple)):
             for id in targets:
                 if id not in self.engines.keys():
                     log.msg("Engine with id %r is not registered" % id)
                     raise error.InvalidEngineID("Engine with id %r is not registered" % id)  
             return map(self.engines.get, targets)
         elif targets == 'all':
             eList = self.engines.values()
             if len(eList) == 0:
                 msg = """There are no engines registered.
                      Check the logs in ~/.ipython/log if you think there should have been."""
                 raise error.NoEnginesRegistered(msg)
             else:
                 return eList
         else:
             raise error.InvalidEngineID("targets argument is not an int, list of ints or 'all': %r"%targets)
     
     def _performOnEngines(self, methodName, *args, **kwargs):
         """Calls a method on engines and returns deferred to list of results.
         
         :Parameters:
             methodName : str
                 Name of the method to be called.
             targets : int, list of ints, 'all'
                 The targets argument to be parsed into a list of engine objects.
             args
                 The positional keyword arguments to be passed to the engines.
             kwargs
                 The keyword arguments passed to the method
                 
         :Returns: List of deferreds to the results on each engine
         
         :Exception:
             InvalidEngineID
                 If the targets argument is bad in any way.
             AttributeError
                 If the method doesn't exist on one of the engines.
         """
         targets = kwargs.pop('targets')
         log.msg("Performing %s on %r" % (methodName, targets))
         # log.msg("Performing %s(%r, %r) on %r" % (methodName, args, kwargs, targets))
         # This will and should raise if targets is not valid!
         engines = self.engineList(targets)
         dList = []
         for e in engines:
             meth = getattr(e, methodName, None)
             if meth is not None:
                 dList.append(meth(*args, **kwargs))
             else:
                 raise AttributeError("Engine %i does not have method %s" % (e.id, methodName))
         return dList
         
     def _performOnEnginesAndGatherBoth(self, methodName, *args, **kwargs):
         """Called _performOnEngines and wraps result/exception into deferred."""
         try:
             dList = self._performOnEngines(methodName, *args, **kwargs)
         except (error.InvalidEngineID, AttributeError, KeyError, error.NoEnginesRegistered):
             return defer.fail(failure.Failure())
         else:
             # Having fireOnOneErrback is causing problems with the determinacy
             # of the system.  Basically, once a single engine has errbacked, this
             # method returns.  In some cases, this will cause client to submit
             # another command.  Because the previous command is still running
             # on some engines, this command will be queued.  When those commands
             # then errback, the second command will raise QueueCleared.  Ahhh!
             d = gatherBoth(dList, 
                            fireOnOneErrback=0,
                            consumeErrors=1,
                            logErrors=0)
             d.addCallback(error.collect_exceptions, methodName)
             return d
               
     #---------------------------------------------------------------------------
     # General IMultiEngine methods
     #---------------------------------------------------------------------------
     
     def get_ids(self):
         return defer.succeed(self.engines.keys())
     
     #---------------------------------------------------------------------------
     # IEngineMultiplexer methods
     #---------------------------------------------------------------------------
     
     def execute(self, lines, targets='all'):
         return self._performOnEnginesAndGatherBoth('execute', lines, targets=targets)
     
     def push(self, ns, targets='all'):
         return self._performOnEnginesAndGatherBoth('push', ns, targets=targets)
         
     def pull(self, keys, targets='all'):
         return self._performOnEnginesAndGatherBoth('pull', keys, targets=targets)
     
     def push_function(self, ns, targets='all'):
         return self._performOnEnginesAndGatherBoth('push_function', ns, targets=targets)
         
     def pull_function(self, keys, targets='all'):
         return self._performOnEnginesAndGatherBoth('pull_function', keys, targets=targets)
     
     def get_result(self, i=None, targets='all'):
         return self._performOnEnginesAndGatherBoth('get_result', i, targets=targets)
     
     def reset(self, targets='all'):
         return self._performOnEnginesAndGatherBoth('reset', targets=targets)
     
     def keys(self, targets='all'):
         return self._performOnEnginesAndGatherBoth('keys', targets=targets)
     
     def kill(self, controller=False, targets='all'):
         if controller:
             targets = 'all'
         d = self._performOnEnginesAndGatherBoth('kill', targets=targets)
         if controller:
             log.msg("Killing controller")
             d.addCallback(lambda _: reactor.callLater(2.0, reactor.stop))
             # Consume any weird stuff coming back
             d.addBoth(lambda _: None)
         return d
     
     def push_serialized(self, namespace, targets='all'):
         for k, v in namespace.iteritems():
             log.msg("Pushed object %s is %f MB" % (k, v.getDataSize()))
         d = self._performOnEnginesAndGatherBoth('push_serialized', namespace, targets=targets)      
         return d
         
     def pull_serialized(self, keys, targets='all'):
         try:
             dList = self._performOnEngines('pull_serialized', keys, targets=targets)
         except (error.InvalidEngineID, AttributeError, error.NoEnginesRegistered):
             return defer.fail(failure.Failure())
         else:
             for d in dList:
                 d.addCallback(self._logSizes)
             d = gatherBoth(dList, 
                            fireOnOneErrback=0,
                            consumeErrors=1,
                            logErrors=0)
             d.addCallback(error.collect_exceptions, 'pull_serialized')
             return d  
                               
     def _logSizes(self, listOfSerialized):
         if isinstance(listOfSerialized, (list, tuple)):
             for s in listOfSerialized:
                 log.msg("Pulled object is %f MB" % s.getDataSize())
         else:
             log.msg("Pulled object is %f MB" % listOfSerialized.getDataSize())
         return listOfSerialized
     
     def clear_queue(self, targets='all'):
         return self._performOnEnginesAndGatherBoth('clear_queue', targets=targets)         
     
     def queue_status(self, targets='all'):
         log.msg("Getting queue status on %r" % targets)
         try:
             engines = self.engineList(targets)
         except (error.InvalidEngineID, AttributeError, error.NoEnginesRegistered):
             return defer.fail(failure.Failure())            
         else:
             dList = []
             for e in engines:
                 dList.append(e.queue_status().addCallback(lambda s:(e.id, s)))
             d = gatherBoth(dList, 
                            fireOnOneErrback=0,
                            consumeErrors=1,
                            logErrors=0)
             d.addCallback(error.collect_exceptions, 'queue_status')
             return d 
     
     def get_properties(self, keys=None, targets='all'):
         log.msg("Getting properties on %r" % targets)
         try:
             engines = self.engineList(targets)
         except (error.InvalidEngineID, AttributeError, error.NoEnginesRegistered):
             return defer.fail(failure.Failure())            
         else:
             dList = [e.get_properties(keys) for e in engines]
             d = gatherBoth(dList, 
                            fireOnOneErrback=0,
                            consumeErrors=1,
                            logErrors=0)
             d.addCallback(error.collect_exceptions, 'get_properties')
             return d
     
     def set_properties(self, properties, targets='all'):
         log.msg("Setting properties on %r" % targets)
         try:
             engines = self.engineList(targets)
         except (error.InvalidEngineID, AttributeError, error.NoEnginesRegistered):
             return defer.fail(failure.Failure())            
         else:
             dList = [e.set_properties(properties) for e in engines]
             d = gatherBoth(dList, 
                            fireOnOneErrback=0,
                            consumeErrors=1,
                            logErrors=0)
             d.addCallback(error.collect_exceptions, 'set_properties')
             return d
     
     def has_properties(self, keys, targets='all'):
         log.msg("Checking properties on %r" % targets)
         try:
             engines = self.engineList(targets)
         except (error.InvalidEngineID, AttributeError, error.NoEnginesRegistered):
             return defer.fail(failure.Failure())            
         else:
             dList = [e.has_properties(keys) for e in engines]
             d = gatherBoth(dList, 
                            fireOnOneErrback=0,
                            consumeErrors=1,
                            logErrors=0)
             d.addCallback(error.collect_exceptions, 'has_properties')
             return d
     
     def del_properties(self, keys, targets='all'):
         log.msg("Deleting properties on %r" % targets)
         try:
             engines = self.engineList(targets)
         except (error.InvalidEngineID, AttributeError, error.NoEnginesRegistered):
             return defer.fail(failure.Failure())            
         else:
             dList = [e.del_properties(keys) for e in engines]
             d = gatherBoth(dList, 
                            fireOnOneErrback=0,
                            consumeErrors=1,
                            logErrors=0)
             d.addCallback(error.collect_exceptions, 'del_properties')
             return d
     
     def clear_properties(self, targets='all'):
         log.msg("Clearing properties on %r" % targets)
         try:
             engines = self.engineList(targets)
         except (error.InvalidEngineID, AttributeError, error.NoEnginesRegistered):
             return defer.fail(failure.Failure())            
         else:
             dList = [e.clear_properties() for e in engines]
             d = gatherBoth(dList, 
                            fireOnOneErrback=0,
                            consumeErrors=1,
                            logErrors=0)
             d.addCallback(error.collect_exceptions, 'clear_properties')
             return d
 
 
 components.registerAdapter(MultiEngine,
                            IControllerBase, 
                            IMultiEngine)
 
 
 #-------------------------------------------------------------------------------
 # Interfaces for the Synchronous MultiEngine
 #-------------------------------------------------------------------------------
 
 class ISynchronousEngineMultiplexer(Interface):
     pass
 
 
 class ISynchronousMultiEngine(ISynchronousEngineMultiplexer):
     """Synchronous, two-phase version of IMultiEngine.
     
     Methods in this interface are identical to those of IMultiEngine, but they
     take one additional argument:
     
     execute(lines, targets='all') -> execute(lines, targets='all, block=True)
     
     :Parameters:
         block : boolean
             Should the method return a deferred to a deferredID or the 
             actual result.  If block=False a deferred to a deferredID is 
             returned and the user must call `get_pending_deferred` at a later
             point.  If block=True, a deferred to the actual result comes back.
     """
     def get_pending_deferred(deferredID, block=True):
         """"""
     
     def clear_pending_deferreds():
         """"""
 
 
 #-------------------------------------------------------------------------------
 # Implementation of the Synchronous MultiEngine
 #-------------------------------------------------------------------------------
 
 class SynchronousMultiEngine(PendingDeferredManager):
     """Adapt an `IMultiEngine` -> `ISynchronousMultiEngine`
     
     Warning, this class uses a decorator that currently uses **kwargs.  
     Because of this block must be passed as a kwarg, not positionally.
     """
     
     implements(ISynchronousMultiEngine)
     
     def __init__(self, multiengine):
         self.multiengine = multiengine
         PendingDeferredManager.__init__(self)
     
     #---------------------------------------------------------------------------
     # Decorated pending deferred methods
     #---------------------------------------------------------------------------
     
+    @profile
     @two_phase
     def execute(self, lines, targets='all'):
         d = self.multiengine.execute(lines, targets)
         return d
     
     @two_phase
     def push(self, namespace, targets='all'):
         return self.multiengine.push(namespace, targets)
     
     @two_phase
     def pull(self, keys, targets='all'):
         d = self.multiengine.pull(keys, targets)
         return d
     
     @two_phase
     def push_function(self, namespace, targets='all'):
         return self.multiengine.push_function(namespace, targets)
     
     @two_phase
     def pull_function(self, keys, targets='all'):
         d = self.multiengine.pull_function(keys, targets)
         return d
     
     @two_phase
     def get_result(self, i=None, targets='all'):
         return self.multiengine.get_result(i, targets='all')
     
     @two_phase
     def reset(self, targets='all'):
         return self.multiengine.reset(targets)
     
     @two_phase
     def keys(self, targets='all'):
         return self.multiengine.keys(targets)
     
     @two_phase
     def kill(self, controller=False, targets='all'):
         return self.multiengine.kill(controller, targets)
     
     @two_phase
     def push_serialized(self, namespace, targets='all'):
         return self.multiengine.push_serialized(namespace, targets)
     
     @two_phase
     def pull_serialized(self, keys, targets='all'):
         return self.multiengine.pull_serialized(keys, targets)
     
     @two_phase
     def clear_queue(self, targets='all'):
         return self.multiengine.clear_queue(targets)
     
     @two_phase
     def queue_status(self, targets='all'):
         return self.multiengine.queue_status(targets)
     
     @two_phase
     def set_properties(self, properties, targets='all'):
         return self.multiengine.set_properties(properties, targets)
     
     @two_phase
     def get_properties(self, keys=None, targets='all'):
         return self.multiengine.get_properties(keys, targets)
     
     @two_phase
     def has_properties(self, keys, targets='all'):
         return self.multiengine.has_properties(keys, targets)
     
     @two_phase
     def del_properties(self, keys, targets='all'):
         return self.multiengine.del_properties(keys, targets)
     
     @two_phase
     def clear_properties(self, targets='all'):
         return self.multiengine.clear_properties(targets)
     
     #---------------------------------------------------------------------------
     # IMultiEngine methods
     #---------------------------------------------------------------------------
     
     def get_ids(self):
         """Return a list of registered engine ids.
         
         Never use the two phase block/non-block stuff for this.
         """
         return self.multiengine.get_ids()
 
 
 components.registerAdapter(SynchronousMultiEngine, IMultiEngine, ISynchronousMultiEngine)
 
 
 #-------------------------------------------------------------------------------
 # Various high-level interfaces that can be used as MultiEngine mix-ins
 #-------------------------------------------------------------------------------
 
 #-------------------------------------------------------------------------------
 # IMultiEngineCoordinator
 #-------------------------------------------------------------------------------
 
 class IMultiEngineCoordinator(Interface):
     """Methods that work on multiple engines explicitly."""
        
     def scatter(key, seq, dist='b', flatten=False, targets='all'):
         """Partition and distribute a sequence to targets."""
         
     def gather(key, dist='b', targets='all'):
         """Gather object key from targets."""
     
     def raw_map(func, seqs, dist='b', targets='all'):
         """
         A parallelized version of Python's builtin `map` function.
         
         This has a slightly different syntax than the builtin `map`.
         This is needed because we need to have keyword arguments and thus
         can't use *args to capture all the sequences.  Instead, they must
         be passed in a list or tuple.
         
         The equivalence is:
         
         raw_map(func, seqs) -> map(func, seqs[0], seqs[1], ...)
         
         Most users will want to use parallel functions or the `mapper`
         and `map` methods for an API that follows that of the builtin
         `map`.
         """
 
 
 class ISynchronousMultiEngineCoordinator(IMultiEngineCoordinator):
     """Methods that work on multiple engines explicitly."""
 
     def scatter(key, seq, dist='b', flatten=False, targets='all', block=True):
         """Partition and distribute a sequence to targets."""
         
     def gather(key, dist='b', targets='all', block=True):
         """Gather object key from targets"""
     
     def raw_map(func, seqs, dist='b', targets='all', block=True):
         """
         A parallelized version of Python's builtin map.
         
         This has a slightly different syntax than the builtin `map`.
         This is needed because we need to have keyword arguments and thus
         can't use *args to capture all the sequences.  Instead, they must
         be passed in a list or tuple.
         
         raw_map(func, seqs) -> map(func, seqs[0], seqs[1], ...)
         
         Most users will want to use parallel functions or the `mapper`
         and `map` methods for an API that follows that of the builtin
         `map`.
         """
 
 
 #-------------------------------------------------------------------------------
 # IMultiEngineExtras
 #-------------------------------------------------------------------------------
 
 class IMultiEngineExtras(Interface):
     
     def zip_pull(targets, keys):
         """
         Pull, but return results in a different format from `pull`.
         
         This method basically returns zip(pull(targets, *keys)), with a few 
         edge cases handled differently.  Users of chainsaw will find this format 
         familiar.
         """
     
     def run(targets, fname):
         """Run a .py file on targets."""
 
 
 class ISynchronousMultiEngineExtras(IMultiEngineExtras):
     def zip_pull(targets, keys, block=True):
         """
         Pull, but return results in a different format from `pull`.
         
         This method basically returns zip(pull(targets, *keys)), with a few 
         edge cases handled differently.  Users of chainsaw will find this format 
         familiar.
         """
     
     def run(targets, fname, block=True):
         """Run a .py file on targets."""
 
 #-------------------------------------------------------------------------------
 # The full MultiEngine interface
 #-------------------------------------------------------------------------------
 
 class IFullMultiEngine(IMultiEngine, 
     IMultiEngineCoordinator, 
     IMultiEngineExtras):
     pass
 
 
 class IFullSynchronousMultiEngine(ISynchronousMultiEngine, 
     ISynchronousMultiEngineCoordinator,
     ISynchronousMultiEngineExtras):
     pass