Merging Brian's trunk-dev.

* Fixed Foolscap import warnings (#525805).
* Improved Monte-Carlo options pricing example.

IPython/kernel/enginefc.py

@@ -1,541 +1,544 b''
 # encoding: utf-8
 # -*- test-case-name: IPython.kernel.test.test_enginepb -*-
 
 """
 Expose the IPython EngineService using the Foolscap network protocol.
 
 Foolscap is a high-performance and secure network protocol.
 """
 __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 cPickle as pickle
 
 from twisted.python import components, log, failure
 from twisted.internet import defer, threads
 from zope.interface import Interface, implements
 
 from twisted.internet.base import DelayedCall
 DelayedCall.debug = True
 
+try:
+    from foolscap.api import Referenceable, DeadReferenceError
+except ImportError:
 from foolscap import Referenceable, DeadReferenceError
 from foolscap.referenceable import RemoteReference
 
 from IPython.kernel.pbutil import packageFailure, unpackageFailure
 from IPython.kernel.controllerservice import IControllerBase
 from IPython.kernel.engineservice import (
     IEngineBase,
     IEngineQueued,
     StrictDict
 )
 from IPython.kernel.pickleutil import (
     can,
     canDict,
     canSequence,
     uncan,
     uncanDict,
     uncanSequence
 )
 
 
 #-------------------------------------------------------------------------------
 # The client (Engine) side of things
 #-------------------------------------------------------------------------------
 
 # Expose a FC interface to the EngineService
      
 class IFCEngine(Interface):
     """An interface that exposes an EngineService over Foolscap.
     
     The methods in this interface are similar to those from IEngine, 
     but their arguments and return values slightly different to reflect
     that FC cannot send arbitrary objects.  We handle this by pickling/
     unpickling that the two endpoints.
     
     If a remote or local exception is raised, the appropriate Failure
     will be returned instead.
     """
     pass
     
 
 class FCEngineReferenceFromService(Referenceable, object):
     """Adapt an `IEngineBase` to an `IFCEngine` implementer.
     
     This exposes an `IEngineBase` to foolscap by adapting it to a
     `foolscap.Referenceable`.
     
     See the documentation of the `IEngineBase` methods for more details.
     """
         
     implements(IFCEngine)
     
     def __init__(self, service):
         assert IEngineBase.providedBy(service), \
             "IEngineBase is not provided by" + repr(service)
         self.service = service
         self.collectors = {}
     
     def remote_get_id(self):
         return self.service.id
     
     def remote_set_id(self, id):
         self.service.id = id
     
     def _checkProperties(self, result):
         dosync = self.service.properties.modified
         self.service.properties.modified = False
         return (dosync and pickle.dumps(self.service.properties, 2)), result
     
     def remote_execute(self, lines):
         d = self.service.execute(lines)
         d.addErrback(packageFailure)
         d.addCallback(self._checkProperties)
         d.addErrback(packageFailure)
         #d.addCallback(lambda r: log.msg("Got result: " + str(r)))
         return d
         
     #---------------------------------------------------------------------------
     # Old version of push
     #---------------------------------------------------------------------------
         
     def remote_push(self, pNamespace):
         try:
             namespace = pickle.loads(pNamespace)
         except:
             return defer.fail(failure.Failure()).addErrback(packageFailure)
         else:
             return self.service.push(namespace).addErrback(packageFailure)
     
     #---------------------------------------------------------------------------
     # pull
     #---------------------------------------------------------------------------     
     
     def remote_pull(self, keys):
         d = self.service.pull(keys)
         d.addCallback(pickle.dumps, 2)
         d.addErrback(packageFailure)
         return d
     
     #---------------------------------------------------------------------------
     # push/pullFuction
     #---------------------------------------------------------------------------
     
     def remote_push_function(self, pNamespace):
         try:
             namespace = pickle.loads(pNamespace)
         except:
             return defer.fail(failure.Failure()).addErrback(packageFailure)
         else:
             # The usage of globals() here is an attempt to bind any pickled functions
             # to the globals of this module.  What we really want is to have it bound
             # to the globals of the callers module.  This will require walking the 
             # stack.  BG 10/3/07.
             namespace = uncanDict(namespace, globals())
             return self.service.push_function(namespace).addErrback(packageFailure)
     
     def remote_pull_function(self, keys):
         d = self.service.pull_function(keys)
         if len(keys)>1:
             d.addCallback(canSequence)
         elif len(keys)==1:
             d.addCallback(can)
         d.addCallback(pickle.dumps, 2)
         d.addErrback(packageFailure)
         return d
 
     #---------------------------------------------------------------------------
     # Other methods
     #---------------------------------------------------------------------------
     
     def remote_get_result(self, i=None):
         return self.service.get_result(i).addErrback(packageFailure)
     
     def remote_reset(self):
         return self.service.reset().addErrback(packageFailure)
     
     def remote_kill(self):
         return self.service.kill().addErrback(packageFailure)
     
     def remote_keys(self):
         return self.service.keys().addErrback(packageFailure)
     
     #---------------------------------------------------------------------------
     # push/pull_serialized
     #---------------------------------------------------------------------------
     
     def remote_push_serialized(self, pNamespace):
         try:
             namespace = pickle.loads(pNamespace)
         except:
             return defer.fail(failure.Failure()).addErrback(packageFailure)
         else:
             d = self.service.push_serialized(namespace)
             return d.addErrback(packageFailure)
     
     def remote_pull_serialized(self, keys):
         d = self.service.pull_serialized(keys)
         d.addCallback(pickle.dumps, 2)
         d.addErrback(packageFailure)
         return d
     
     #---------------------------------------------------------------------------
     # Properties interface
     #---------------------------------------------------------------------------
     
     def remote_set_properties(self, pNamespace):
         try:
             namespace = pickle.loads(pNamespace)
         except:
             return defer.fail(failure.Failure()).addErrback(packageFailure)
         else:
             return self.service.set_properties(namespace).addErrback(packageFailure)
     
     def remote_get_properties(self, keys=None):
         d = self.service.get_properties(keys)
         d.addCallback(pickle.dumps, 2)
         d.addErrback(packageFailure)
         return d
     
     def remote_has_properties(self, keys):
         d = self.service.has_properties(keys)
         d.addCallback(pickle.dumps, 2)
         d.addErrback(packageFailure)
         return d
     
     def remote_del_properties(self, keys):
         d = self.service.del_properties(keys)
         d.addErrback(packageFailure)
         return d
     
     def remote_clear_properties(self):
         d = self.service.clear_properties()
         d.addErrback(packageFailure)
         return d
     
 
 components.registerAdapter(FCEngineReferenceFromService,
                            IEngineBase,
                            IFCEngine)
 
 
 #-------------------------------------------------------------------------------
 # Now the server (Controller) side of things
 #-------------------------------------------------------------------------------
 
 class EngineFromReference(object):
     """Adapt a `RemoteReference` to an `IEngineBase` implementing object.
     
     When an engine connects to a controller, it calls the `register_engine`
     method of the controller and passes the controller a `RemoteReference` to
     itself.  This class is used to adapt this `RemoteReference` to an object
     that implements the full `IEngineBase` interface.
     
     See the documentation of `IEngineBase` for details on the methods.
     """
     
     implements(IEngineBase)
     
     def __init__(self, reference):
         self.reference = reference
         self._id = None
         self._properties = StrictDict()
         self.currentCommand = None
     
     def callRemote(self, *args, **kwargs):
         try:
             return self.reference.callRemote(*args, **kwargs)
         except DeadReferenceError:
             self.notifier()
             self.stopNotifying(self.notifier)
             return defer.fail()
     
     def get_id(self):
         """Return the Engines id."""
         return self._id
     
     def set_id(self, id):
         """Set the Engines id."""
         self._id = id
         return self.callRemote('set_id', id)
     
     id = property(get_id, set_id)
     
     def syncProperties(self, r):
         try:
             psync, result = r
         except (ValueError, TypeError):
             return r
         else:
             if psync:
                 log.msg("sync properties")
                 pick = self.checkReturnForFailure(psync)
                 if isinstance(pick, failure.Failure):
                     self.properties = pick
                     return pick
                 else:
                     self.properties = pickle.loads(pick)
             return result
     
     def _set_properties(self, dikt):
         self._properties.clear()
         self._properties.update(dikt)
     
     def _get_properties(self):
         if isinstance(self._properties, failure.Failure):
             self._properties.raiseException()
         return self._properties
     
     properties = property(_get_properties, _set_properties)
     
     #---------------------------------------------------------------------------
     # Methods from IEngine
     #---------------------------------------------------------------------------
     
     #---------------------------------------------------------------------------
     # execute
     #---------------------------------------------------------------------------
     
     def execute(self, lines):
         # self._needProperties = True
         d = self.callRemote('execute', lines)
         d.addCallback(self.syncProperties)
         return d.addCallback(self.checkReturnForFailure)
     
     #---------------------------------------------------------------------------
     # push
     #---------------------------------------------------------------------------
     
     def push(self, namespace):
         try:
             package = pickle.dumps(namespace, 2)
         except:
             return defer.fail(failure.Failure())
         else:
             if isinstance(package, failure.Failure):
                 return defer.fail(package)
             else:
                 d = self.callRemote('push', package)
                 return d.addCallback(self.checkReturnForFailure)
     
     #---------------------------------------------------------------------------
     # pull
     #---------------------------------------------------------------------------
     
     def pull(self, keys):
         d = self.callRemote('pull', keys)
         d.addCallback(self.checkReturnForFailure)
         d.addCallback(pickle.loads)
         return d
     
     #---------------------------------------------------------------------------
     # push/pull_function
     #---------------------------------------------------------------------------
     
     def push_function(self, namespace):
         try:
             package = pickle.dumps(canDict(namespace), 2)
         except:
             return defer.fail(failure.Failure())
         else:
             if isinstance(package, failure.Failure):
                 return defer.fail(package)
             else:
                 d = self.callRemote('push_function', package)
                 return d.addCallback(self.checkReturnForFailure)    
     
     def pull_function(self, keys):
         d = self.callRemote('pull_function', keys)
         d.addCallback(self.checkReturnForFailure)
         d.addCallback(pickle.loads)
         # The usage of globals() here is an attempt to bind any pickled functions
         # to the globals of this module.  What we really want is to have it bound
         # to the globals of the callers module.  This will require walking the 
         # stack.  BG 10/3/07.
         if len(keys)==1:
             d.addCallback(uncan, globals())
         elif len(keys)>1:
             d.addCallback(uncanSequence, globals())            
         return d
     
     #---------------------------------------------------------------------------
     # Other methods
     #---------------------------------------------------------------------------
     
     def get_result(self, i=None):
         return self.callRemote('get_result', i).addCallback(self.checkReturnForFailure)
     
     def reset(self):
         self._refreshProperties = True
         d = self.callRemote('reset')
         d.addCallback(self.syncProperties)
         return d.addCallback(self.checkReturnForFailure)
     
     def kill(self):
         #this will raise pb.PBConnectionLost on success
         d = self.callRemote('kill')
         d.addCallback(self.syncProperties)
         d.addCallback(self.checkReturnForFailure)
         d.addErrback(self.killBack)
         return d
     
     def killBack(self, f):
         log.msg('filling engine: %s' % f)
         return None
     
     def keys(self):
         return self.callRemote('keys').addCallback(self.checkReturnForFailure)
     
     #---------------------------------------------------------------------------
     # Properties methods
     #---------------------------------------------------------------------------
     
     def set_properties(self, properties):
         try:
             package = pickle.dumps(properties, 2)
         except:
             return defer.fail(failure.Failure())
         else:
             if isinstance(package, failure.Failure):
                 return defer.fail(package)
             else:
                 d = self.callRemote('set_properties', package)
                 return d.addCallback(self.checkReturnForFailure)
         return d
     
     def get_properties(self, keys=None):
         d = self.callRemote('get_properties', keys)
         d.addCallback(self.checkReturnForFailure)
         d.addCallback(pickle.loads)
         return d
     
     def has_properties(self, keys):
         d = self.callRemote('has_properties', keys)
         d.addCallback(self.checkReturnForFailure)
         d.addCallback(pickle.loads)
         return d
     
     def del_properties(self, keys):
         d = self.callRemote('del_properties', keys)
         d.addCallback(self.checkReturnForFailure)
         # d.addCallback(pickle.loads)
         return d
     
     def clear_properties(self):
         d = self.callRemote('clear_properties')
         d.addCallback(self.checkReturnForFailure)
         return d
     
     #---------------------------------------------------------------------------
     # push/pull_serialized
     #---------------------------------------------------------------------------
     
     def push_serialized(self, namespace):
         """Older version of pushSerialize."""
         try:
             package = pickle.dumps(namespace, 2)
         except:
             return defer.fail(failure.Failure())
         else:
             if isinstance(package, failure.Failure):
                 return defer.fail(package)
             else:
                 d = self.callRemote('push_serialized', package)
                 return d.addCallback(self.checkReturnForFailure)
     
     def pull_serialized(self, keys):
         d = self.callRemote('pull_serialized', keys)
         d.addCallback(self.checkReturnForFailure)
         d.addCallback(pickle.loads)
         return d
     
     #---------------------------------------------------------------------------
     # Misc
     #---------------------------------------------------------------------------
      
     def checkReturnForFailure(self, r):
         """See if a returned value is a pickled Failure object.
         
         To distinguish between general pickled objects and pickled Failures, the
         other side should prepend the string FAILURE: to any pickled Failure.
         """
         return unpackageFailure(r)
     
 
 components.registerAdapter(EngineFromReference,
     RemoteReference,
     IEngineBase)
 
 
 #-------------------------------------------------------------------------------
 # Now adapt an IControllerBase to incoming FC connections
 #-------------------------------------------------------------------------------
 
 
 class IFCControllerBase(Interface):
     """
     Interface that tells how an Engine sees a Controller.
     
     In our architecture, the Controller listens for Engines to connect
     and register.  This interface defines that registration method as it is 
     exposed over the Foolscap network protocol
     """
     
     def remote_register_engine(self, engineReference, id=None, pid=None, pproperties=None):
         """
         Register new engine on the controller.
         
         Engines must call this upon connecting to the controller if they
         want to do work for the controller.
         
         See the documentation of `IControllerCore` for more details.
         """
     
 
 class FCRemoteEngineRefFromService(Referenceable):
     """
     Adapt an `IControllerBase` to an `IFCControllerBase`.
     """
     
     implements(IFCControllerBase)
     
     def __init__(self, service):
         assert IControllerBase.providedBy(service), \
             "IControllerBase is not provided by " + repr(service)
         self.service = service
     
     def remote_register_engine(self, engine_reference, id=None, pid=None, pproperties=None):
         # First adapt the engine_reference to a basic non-queued engine
         engine = IEngineBase(engine_reference)
         if pproperties:
             engine.properties = pickle.loads(pproperties)
         # Make it an IQueuedEngine before registration
         remote_engine = IEngineQueued(engine)
         # Get the ip/port of the remote side
         peer_address = engine_reference.tracker.broker.transport.getPeer()
         ip = peer_address.host
         port = peer_address.port
         reg_dict = self.service.register_engine(remote_engine, id, ip, port, pid)
         # Now setup callback for disconnect and unregistering the engine
         def notify(*args):
             return self.service.unregister_engine(reg_dict['id'])
         engine_reference.tracker.broker.notifyOnDisconnect(notify)
         
         engine.notifier = notify
         engine.stopNotifying = engine_reference.tracker.broker.dontNotifyOnDisconnect
         
         return reg_dict
 
 
 components.registerAdapter(FCRemoteEngineRefFromService,
                         IControllerBase,
                         IFCControllerBase)

IPython/kernel/fcutil.py

@@ -1,293 +1,296 b''
 #!/usr/bin/env python
 # encoding: utf-8
 """
 Foolscap related utilities.
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008-2009  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 __future__ import with_statement
 
 import os
 import tempfile
 
 from twisted.internet import reactor, defer
 from twisted.python import log
 
 import foolscap
+try:
+    from foolscap.api import Tub, UnauthenticatedTub
+except ImportError:
 from foolscap import Tub, UnauthenticatedTub
 
 from IPython.config.loader import Config
 from IPython.kernel.configobjfactory import AdaptedConfiguredObjectFactory
 from IPython.kernel.error import SecurityError
 
 from IPython.utils.importstring import import_item
 from IPython.utils.path import expand_path
 from IPython.utils.traitlets import Int, Str, Bool, Instance
 
 #-----------------------------------------------------------------------------
 # Code
 #-----------------------------------------------------------------------------
 
 
 # We do this so if a user doesn't have OpenSSL installed, it will try to use
 # an UnauthenticatedTub.  But, they will still run into problems if they
 # try to use encrypted furls.
 try:
     import OpenSSL
 except:
     Tub = UnauthenticatedTub
     have_crypto = False
 else:
     have_crypto = True
 
 
 class FURLError(Exception):
     pass
 
 
 def check_furl_file_security(furl_file, secure):
     """Remove the old furl_file if changing security modes."""
     furl_file = expand_path(furl_file)
     if os.path.isfile(furl_file):
         with open(furl_file, 'r') as f:
             oldfurl = f.read().strip()
         if (oldfurl.startswith('pb://') and not secure) or (oldfurl.startswith('pbu://') and secure):
             os.remove(furl_file)
 
 
 def is_secure(furl):
     """Is the given FURL secure or not."""
     if is_valid_furl(furl):
         if furl.startswith("pb://"):
             return True
         elif furl.startswith("pbu://"):
             return False
     else:
         raise FURLError("invalid FURL: %s" % furl)
 
 
 def is_valid_furl(furl):
     """Is the str a valid FURL or not."""
     if isinstance(furl, str):
         if furl.startswith("pb://") or furl.startswith("pbu://"):
             return True
         else:
             return False
     else:
         return False
 
 
 def is_valid_furl_file(furl_or_file):
     """See if furl_or_file exists and contains a valid FURL.
 
     This doesn't try to read the contents because often we have to validate
     FURL files that are created, but don't yet have a FURL written to them.
     """
     if isinstance(furl_or_file, (str, unicode)):
         path, furl_filename = os.path.split(furl_or_file)
         if os.path.isdir(path) and furl_filename.endswith('.furl'):
             return True
     return False
 
 
 def find_furl(furl_or_file):
     """Find, validate and return a FURL in a string or file.
 
     This calls :func:`IPython.utils.path.expand_path` on the argument to
     properly handle ``~`` and ``$`` variables in the path.
     """
     if is_valid_furl(furl_or_file):
         return furl_or_file
     furl_or_file = expand_path(furl_or_file)
     if is_valid_furl_file(furl_or_file):
         with open(furl_or_file, 'r') as f:
             furl = f.read().strip()
         if is_valid_furl(furl):
             return furl
     raise FURLError("Not a valid FURL or FURL file: %r" % furl_or_file)
 
 
 def is_valid_furl_or_file(furl_or_file):
     """Validate a FURL or a FURL file.
 
     If ``furl_or_file`` looks like a file, we simply make sure its directory
     exists and that it has a ``.furl`` file extension.  We don't try to see
     if the FURL file exists or to read its contents. This is useful for
     cases where auto re-connection is being used.
     """
     if is_valid_furl(furl_or_file) or is_valid_furl_file(furl_or_file):
         return True
     else:
         return False
 
 
 def validate_furl_or_file(furl_or_file):
     """Like :func:`is_valid_furl_or_file`, but raises an error."""
     if not is_valid_furl_or_file(furl_or_file):
         raise FURLError('Not a valid FURL or FURL file: %r' % furl_or_file)
 
 
 def get_temp_furlfile(filename):
     """Return a temporary FURL file."""
     return tempfile.mktemp(dir=os.path.dirname(filename),
                            prefix=os.path.basename(filename))
 
 
 def make_tub(ip, port, secure, cert_file):
     """Create a listening tub given an ip, port, and cert_file location.
     
     Parameters
     ----------
     ip : str
         The ip address or hostname that the tub should listen on.  
         Empty means all interfaces.
     port : int
         The port that the tub should listen on.  A value of 0 means
         pick a random port
     secure: bool
         Will the connection be secure (in the Foolscap sense).
     cert_file: str
         A filename of a file to be used for theSSL certificate.
 
     Returns
     -------
     A tub, listener tuple.
     """
     if secure:
         if have_crypto:
             tub = Tub(certFile=cert_file)
         else:
             raise SecurityError("OpenSSL/pyOpenSSL is not available, so we "
                                 "can't run in secure mode.  Try running without "
                                 "security using 'ipcontroller -xy'.")
     else:
         tub = UnauthenticatedTub()
     
     # Set the strport based on the ip and port and start listening
     if ip == '':
         strport = "tcp:%i" % port
     else:
         strport = "tcp:%i:interface=%s" % (port, ip)
     log.msg("Starting listener with [secure=%r] on: %s" % (secure, strport))
     listener = tub.listenOn(strport)
     
     return tub, listener
 
 
 class FCServiceFactory(AdaptedConfiguredObjectFactory):
     """This class creates a tub with various services running in it.
 
     The basic idea is that :meth:`create` returns a running :class:`Tub`
     instance that has a number of Foolscap references registered in it.
     This class is a subclass of :class:`IPython.core.component.Component`
     so the IPython configuration and component system are used.
 
     Attributes
     ----------
     interfaces : Config
         A Config instance whose values are sub-Config objects having two
         keys: furl_file and interface_chain.
 
     The other attributes are the standard ones for Foolscap.
     """
 
     ip = Str('', config=True)
     port = Int(0, config=True)
     secure = Bool(True, config=True)
     cert_file = Str('', config=True)
     location = Str('', config=True)
     reuse_furls = Bool(False, config=True)
     interfaces = Instance(klass=Config, kw={}, allow_none=False, config=True)
 
     def __init__(self, config, adaptee):
         super(FCServiceFactory, self).__init__(config, adaptee)
         self._check_reuse_furls()
 
     def _ip_changed(self, name, old, new):
         if new == 'localhost' or new == '127.0.0.1':
             self.location = '127.0.0.1'
 
     def _check_reuse_furls(self):
         furl_files = [i.furl_file for i in self.interfaces.values()]
         for ff in furl_files:
             fullfile = self._get_security_file(ff)
             if self.reuse_furls:
                 if self.port==0:
                     raise FURLError("You are trying to reuse the FURL file "
                         "for this connection, but the port for this connection "
                         "is set to 0 (autoselect). To reuse the FURL file "
                         "you need to specify specific port to listen on."
                     )
                 else:
                     log.msg("Reusing FURL file: %s" % fullfile)
             else:
                 if os.path.isfile(fullfile):
                     log.msg("Removing old FURL file: %s" % fullfile)
                     os.remove(fullfile)
 
     def _get_security_file(self, filename):
         return os.path.join(self.config.Global.security_dir, filename)
 
     def create(self):
         """Create and return the Foolscap tub with everything running."""
 
         self.tub, self.listener = make_tub(
             self.ip, self.port, self.secure, 
             self._get_security_file(self.cert_file)
         )
         # log.msg("Interfaces to register [%r]: %r" % \
         #     (self.__class__, self.interfaces))
         if not self.secure:
             log.msg("WARNING: running with no security: %s" % \
                 self.__class__.__name__)
         reactor.callWhenRunning(self.set_location_and_register)
         return self.tub
 
     def set_location_and_register(self):
         """Set the location for the tub and return a deferred."""
 
         if self.location == '':
             d = self.tub.setLocationAutomatically()
         else:
             d = defer.maybeDeferred(self.tub.setLocation,
                 "%s:%i" % (self.location, self.listener.getPortnum()))
         self.adapt_to_interfaces(d)
 
     def adapt_to_interfaces(self, d):
         """Run through the interfaces, adapt and register."""
 
         for ifname, ifconfig in self.interfaces.iteritems():
             ff = self._get_security_file(ifconfig.furl_file)
             log.msg("Adapting [%s] to interface: %s" % \
                 (self.adaptee.__class__.__name__, ifname))
             log.msg("Saving FURL for interface [%s] to file: %s" % (ifname, ff))
             check_furl_file_security(ff, self.secure)
             adaptee = self.adaptee
             for i in ifconfig.interface_chain:
                 adaptee = import_item(i)(adaptee)
             d.addCallback(self.register, adaptee, furl_file=ff)
 
     def register(self, empty, ref, furl_file):
         """Register the reference with the FURL file.
 
         The FURL file is created and then moved to make sure that when the
         file appears, the buffer has been flushed and the file closed. This
         is not done if we are re-using FURLS however.
         """
         if self.reuse_furls:
             self.tub.registerReference(ref, furlFile=furl_file)
         else:
             temp_furl_file = get_temp_furlfile(furl_file)
             self.tub.registerReference(ref, furlFile=temp_furl_file)
             os.rename(temp_furl_file, furl_file)
 

IPython/kernel/multiengineclient.py

@@ -1,901 +1,905 b''
 # encoding: utf-8
 # -*- test-case-name: IPython.kernel.test.test_multiengineclient -*-
 
 """General Classes for IMultiEngine clients."""
 
 __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 sys
 import warnings
 
 from twisted.python import components
 from twisted.python.failure import Failure
 from zope.interface import Interface, implements, Attribute
+
+try:
+    from foolscap.api import DeadReferenceError
+except ImportError:
 from foolscap import DeadReferenceError
 
 from IPython.utils.coloransi import TermColors
 
 from IPython.kernel.twistedutil import blockingCallFromThread
 from IPython.kernel import error
 from IPython.kernel.parallelfunction import ParallelFunction
 from IPython.kernel.mapper import (
     MultiEngineMapper, 
     IMultiEngineMapperFactory,
     IMapper
 )
 
 from IPython.kernel.multiengine import IFullSynchronousMultiEngine
 
 
 #-------------------------------------------------------------------------------
 # Pending Result things
 #-------------------------------------------------------------------------------
 
 class IPendingResult(Interface):
     """A representation of a result that is pending.
     
     This class is similar to Twisted's `Deferred` object, but is designed to be
     used in a synchronous context.
     """
     
     result_id=Attribute("ID of the deferred on the other side")
     client=Attribute("A client that I came from")
     r=Attribute("An attribute that is a property that calls and returns get_result")
     
     def get_result(default=None, block=True):
         """
         Get a result that is pending.
                 
         :Parameters:
             default
                 The value to return if the result is not ready.
             block : boolean
                 Should I block for the result.
                 
         :Returns: The actual result or the default value.
         """
         
     def add_callback(f, *args, **kwargs):
         """
         Add a callback that is called with the result.
         
         If the original result is foo, adding a callback will cause
         f(foo, *args, **kwargs) to be returned instead.  If multiple
         callbacks are registered, they are chained together: the result of
         one is passed to the next and so on.  
         
         Unlike Twisted's Deferred object, there is no errback chain.  Thus
         any exception raised will not be caught and handled.  User must 
         catch these by hand when calling `get_result`.
         """
 
 
 class PendingResult(object):
     """A representation of a result that is not yet ready.
     
     A user should not create a `PendingResult` instance by hand.
     
     Methods:
     
     * `get_result`
     * `add_callback`
     
     Properties:
     
     * `r`
     """
     
     def __init__(self, client, result_id):
         """Create a PendingResult with a result_id and a client instance.
         
         The client should implement `_getPendingResult(result_id, block)`.
         """
         self.client = client
         self.result_id = result_id
         self.called = False
         self.raised = False
         self.callbacks = []
         
     def get_result(self, default=None, block=True):
         """Get a result that is pending.
                 
         This method will connect to an IMultiEngine adapted controller
         and see if the result is ready.  If the action triggers an exception
         raise it and record it.  This method records the result/exception once it is 
         retrieved.  Calling `get_result` again will get this cached result or will
         re-raise the exception.  The .r attribute is a property that calls
         `get_result` with block=True.
         
         :Parameters:
             default
                 The value to return if the result is not ready.
             block : boolean
                 Should I block for the result.
                 
         :Returns: The actual result or the default value.
         """
         
         if self.called:
             if self.raised:
                 raise self.result[0], self.result[1], self.result[2]
             else:
                 return self.result
         try:
             result = self.client.get_pending_deferred(self.result_id, block)
         except error.ResultNotCompleted:
             return default
         except:
             # Reraise other error, but first record them so they can be reraised
             # later if .r or get_result is called again.
             self.result = sys.exc_info()
             self.called = True
             self.raised = True
             raise
         else:
             for cb in self.callbacks:
                 result = cb[0](result, *cb[1], **cb[2])
             self.result = result
             self.called = True
             return result
         
     def add_callback(self, f, *args, **kwargs):
         """Add a callback that is called with the result.
         
         If the original result is result, adding a callback will cause
         f(result, *args, **kwargs) to be returned instead.  If multiple
         callbacks are registered, they are chained together: the result of
         one is passed to the next and so on.  
         
         Unlike Twisted's Deferred object, there is no errback chain.  Thus
         any exception raised will not be caught and handled.  User must 
         catch these by hand when calling `get_result`.
         """
         assert callable(f)
         self.callbacks.append((f, args, kwargs))
         
     def __cmp__(self, other):
         if self.result_id < other.result_id:
             return -1
         else:
             return 1
             
     def _get_r(self):
         return self.get_result(block=True)
     
     r = property(_get_r)
     """This property is a shortcut to a `get_result(block=True)`."""
 
 
 #-------------------------------------------------------------------------------
 # Pretty printing wrappers for certain lists
 #-------------------------------------------------------------------------------    
     
 class ResultList(list):
     """A subclass of list that pretty prints the output of `execute`/`get_result`."""
     
     def __repr__(self):
         output = []
         # These colored prompts were not working on Windows
         if sys.platform == 'win32':
             blue = normal = red = green = ''
         else:
             blue = TermColors.Blue
             normal = TermColors.Normal
             red = TermColors.Red
             green = TermColors.Green
         output.append("<Results List>\n")
         for cmd in self:
             if isinstance(cmd, Failure):
                 output.append(cmd)
             else:
                 target = cmd.get('id',None)
                 cmd_num = cmd.get('number',None)
                 cmd_stdin = cmd.get('input',{}).get('translated','No Input')
                 cmd_stdout = cmd.get('stdout', None)
                 cmd_stderr = cmd.get('stderr', None)
                 output.append("%s[%i]%s In [%i]:%s %s\n" % \
                     (green, target,
                     blue, cmd_num, normal, cmd_stdin))
                 if cmd_stdout:
                     output.append("%s[%i]%s Out[%i]:%s %s\n" % \
                         (green, target,
                         red, cmd_num, normal, cmd_stdout))
                 if cmd_stderr:
                     output.append("%s[%i]%s Err[%i]:\n%s %s" % \
                         (green, target,
                         red, cmd_num, normal, cmd_stderr))
         return ''.join(output)
 
 
 def wrapResultList(result):
     """A function that wraps the output of `execute`/`get_result` -> `ResultList`."""
     if len(result) == 0:
         result = [result]
     return ResultList(result)
 
 
 class QueueStatusList(list):
     """A subclass of list that pretty prints the output of `queue_status`."""
     
     def __repr__(self):
         output = []
         output.append("<Queue Status List>\n")
         for e in self:
             output.append("Engine: %s\n" % repr(e[0]))
             output.append("    Pending: %s\n" % repr(e[1]['pending']))
             for q in e[1]['queue']:
                 output.append("    Command: %s\n" % repr(q))
         return ''.join(output)
 
 
 #-------------------------------------------------------------------------------
 # InteractiveMultiEngineClient
 #-------------------------------------------------------------------------------    
 
 class InteractiveMultiEngineClient(object):
     """A mixin class that add a few methods to a multiengine client.
     
     The methods in this mixin class are designed for interactive usage.
     """
                                         
     def activate(self):
         """Make this `MultiEngineClient` active for parallel magic commands.
         
         IPython has a magic command syntax to work with `MultiEngineClient` objects.
         In a given IPython session there is a single active one.  While
         there can be many `MultiEngineClient` created and used by the user, 
         there is only one active one.  The active `MultiEngineClient` is used whenever 
         the magic commands %px and %autopx are used.
         
         The activate() method is called on a given `MultiEngineClient` to make it 
         active.  Once this has been done, the magic commands can be used.
         """
         
         try:
             # This is injected into __builtins__.
             ip = get_ipython()
         except NameError:
             print "The IPython parallel magics (%result, %px, %autopx) only work within IPython."
         else:
             pmagic = ip.get_component('parallel_magic')
             if pmagic is not None:
                 pmagic.active_multiengine_client = self
             else:
                 print "You must first load the parallelmagic extension " \
                       "by doing '%load_ext parallelmagic'"
 
     def __setitem__(self, key, value):
         """Add a dictionary interface for pushing/pulling.
         
         This functions as a shorthand for `push`.
         
         :Parameters:
             key : str 
                 What to call the remote object.
             value : object
                 The local Python object to push.
         """
         targets, block = self._findTargetsAndBlock()
         return self.push({key:value}, targets=targets, block=block)
     
     def __getitem__(self, key):
         """Add a dictionary interface for pushing/pulling.
         
         This functions as a shorthand to `pull`.
         
         :Parameters:
          - `key`: A string representing the key.
         """
         if isinstance(key, str):
             targets, block = self._findTargetsAndBlock()
             return self.pull(key, targets=targets, block=block)
         else:
             raise TypeError("__getitem__ only takes strs")
             
     def __len__(self):
         """Return the number of available engines."""
         return len(self.get_ids())
 
 
 #-------------------------------------------------------------------------------
 # The top-level MultiEngine client adaptor
 #-------------------------------------------------------------------------------
 
 
 _prop_warn = """\
 
 We are currently refactoring the task dependency system.  This might
 involve the removal of this method and other methods related to engine
 properties.  Please see the docstrings for IPython.kernel.TaskRejectError 
 for more information."""
 
 
 class IFullBlockingMultiEngineClient(Interface):
     pass
 
 
 class FullBlockingMultiEngineClient(InteractiveMultiEngineClient):
     """
     A blocking client to the `IMultiEngine` controller interface.
     
     This class allows users to use a set of engines for a parallel
     computation through the `IMultiEngine` interface.  In this interface,
     each engine has a specific id (an int) that is used to refer to the
     engine, run code on it, etc.
     """
     
     implements(
         IFullBlockingMultiEngineClient,
         IMultiEngineMapperFactory,
         IMapper
     )
     
     def __init__(self, smultiengine):
         self.smultiengine = smultiengine
         self.block = True
         self.targets = 'all'
     
     def _findBlock(self, block=None):
         if block is None:
             return self.block
         else:
             if block in (True, False):
                 return block
             else:
                 raise ValueError("block must be True or False")
     
     def _findTargets(self, targets=None):
         if targets is None:
             return self.targets
         else:
             if not isinstance(targets, (str,list,tuple,int)):
                 raise ValueError("targets must be a str, list, tuple or int")
             return targets
     
     def _findTargetsAndBlock(self, targets=None, block=None):
         return self._findTargets(targets), self._findBlock(block) 
 
     def _bcft(self, *args, **kwargs):
         try:
             result = blockingCallFromThread(*args, **kwargs)
         except DeadReferenceError:
             raise error.ConnectionError(
                 """A connection error has occurred in trying to connect to the
                 controller. This is usually caused by the controller dying or 
                 being restarted. To resolve this issue try recreating the 
                 multiengine client."""
             )
         else:
             return result
 
     def _blockFromThread(self, function, *args, **kwargs):
         block = kwargs.get('block', None)
         if block is None:
             raise error.MissingBlockArgument("'block' keyword argument is missing")
         result = self._bcft(function, *args, **kwargs)
         if not block:
             result = PendingResult(self, result)
         return result
     
     def get_pending_deferred(self, deferredID, block):
         return self._bcft(self.smultiengine.get_pending_deferred, deferredID, block)
     
     def barrier(self, pendingResults):
         """Synchronize a set of `PendingResults`.
         
         This method is a synchronization primitive that waits for a set of
         `PendingResult` objects to complete.  More specifically, barier does
         the following.
         
         * The `PendingResult`s are sorted by result_id.
         * The `get_result` method is called for each `PendingResult` sequentially
           with block=True.
         * If a `PendingResult` gets a result that is an exception, it is 
           trapped and can be re-raised later by calling `get_result` again.
         * The `PendingResult`s are flushed from the controller.
                 
         After barrier has been called on a `PendingResult`, its results can 
         be retrieved by calling `get_result` again or accesing the `r` attribute
         of the instance.
         """
         
         # Convert to list for sorting and check class type 
         prList = list(pendingResults)
         for pr in prList:
             if not isinstance(pr, PendingResult):
                 raise error.NotAPendingResult("Objects passed to barrier must be PendingResult instances")
                             
         # Sort the PendingResults so they are in order
         prList.sort()
         # Block on each PendingResult object
         for pr in prList:
             try:
                 result = pr.get_result(block=True)
             except Exception:
                 pass
     
     def flush(self):
         """
         Clear all pending deferreds/results from the controller.
         
         For each `PendingResult` that is created by this client, the controller
         holds on to the result for that `PendingResult`.  This can be a problem
         if there are a large number of `PendingResult` objects that are created.
         
         Once the result of the `PendingResult` has been retrieved, the result
         is removed from the controller, but if a user doesn't get a result (
         they just ignore the `PendingResult`) the result is kept forever on the
         controller.  This method allows the user to clear out all un-retrieved
         results on the controller. 
         """
         r = self._bcft(self.smultiengine.clear_pending_deferreds)
         return r
     
     clear_pending_results = flush
     
     #---------------------------------------------------------------------------
     # IEngineMultiplexer related methods
     #---------------------------------------------------------------------------
     
     def execute(self, lines, targets=None, block=None):
         """
         Execute code on a set of engines.
         
         :Parameters:
             lines : str
                 The Python code to execute as a string
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         result = self._bcft(self.smultiengine.execute, lines,
             targets=targets, block=block)
         if block:
             result = ResultList(result)
         else:
             result = PendingResult(self, result)
             result.add_callback(wrapResultList)
         return result
     
     def push(self, namespace, targets=None, block=None):
         """
         Push a dictionary of keys and values to engines namespace.
         
         Each engine has a persistent namespace.  This method is used to push
         Python objects into that namespace.
         
         The objects in the namespace must be pickleable.
         
         :Parameters:
             namespace : dict
                 A dict that contains Python objects to be injected into
                 the engine persistent namespace.
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.push, namespace,
             targets=targets, block=block)
     
     def pull(self, keys, targets=None, block=None):
         """
         Pull Python objects by key out of engines namespaces.
         
         :Parameters:
             keys : str or list of str
                 The names of the variables to be pulled
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.pull, keys, targets=targets, block=block)
     
     def push_function(self, namespace, targets=None, block=None):
         """
         Push a Python function to an engine.
         
         This method is used to push a Python function to an engine.  This
         method can then be used in code on the engines.  Closures are not supported.
         
         :Parameters:
             namespace : dict
                 A dict whose values are the functions to be pushed.  The keys give
                 that names that the function will appear as in the engines
                 namespace.
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.push_function, namespace, targets=targets, block=block)
     
     def pull_function(self, keys, targets=None, block=None):
         """
         Pull a Python function from an engine.
         
         This method is used to pull a Python function from an engine.
         Closures are not supported.
         
         :Parameters:
             keys : str or list of str
                 The names of the functions to be pulled
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.pull_function, keys, targets=targets, block=block)
     
     def push_serialized(self, namespace, targets=None, block=None):
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.push_serialized, namespace, targets=targets, block=block)
     
     def pull_serialized(self, keys, targets=None, block=None):
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.pull_serialized, keys, targets=targets, block=block)
     
     def get_result(self, i=None, targets=None, block=None):
         """
         Get a previous result.
         
         When code is executed in an engine, a dict is created and returned.  This
         method retrieves that dict for previous commands.
         
         :Parameters:
             i : int
                 The number of the result to get
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         result = self._bcft(self.smultiengine.get_result, i, targets=targets, block=block)
         if block:
             result = ResultList(result)
         else:
             result = PendingResult(self, result)
             result.add_callback(wrapResultList)
         return result
     
     def reset(self, targets=None, block=None):
         """
         Reset an engine.
         
         This method clears out the namespace of an engine.
 
         :Parameters:
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.reset, targets=targets, block=block)
     
     def keys(self, targets=None, block=None):
         """
         Get a list of all the variables in an engine's namespace.
 
         :Parameters:
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.keys, targets=targets, block=block)
     
     def kill(self, controller=False, targets=None, block=None):
         """
         Kill the engines and controller.
         
         This method is used to stop the engine and controller by calling
         `reactor.stop`.
         
         :Parameters:
             controller : boolean
                 If True, kill the engines and controller.  If False, just the 
                 engines
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.kill, controller, targets=targets, block=block)
     
     def clear_queue(self, targets=None, block=None):
         """
         Clear out the controller's queue for an engine.
         
         The controller maintains a queue for each engine.  This clear it out.
         
         :Parameters:
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.clear_queue, targets=targets, block=block)
     
     def queue_status(self, targets=None, block=None):
         """
         Get the status of an engines queue.
         
         :Parameters:
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.queue_status, targets=targets, block=block)
     
     def set_properties(self, properties, targets=None, block=None):
         warnings.warn(_prop_warn)
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.set_properties, properties, targets=targets, block=block)
     
     def get_properties(self, keys=None, targets=None, block=None):
         warnings.warn(_prop_warn)
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.get_properties, keys, targets=targets, block=block)
     
     def has_properties(self, keys, targets=None, block=None):
         warnings.warn(_prop_warn)
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.has_properties, keys, targets=targets, block=block)
     
     def del_properties(self, keys, targets=None, block=None):
         warnings.warn(_prop_warn)
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.del_properties, keys, targets=targets, block=block)
     
     def clear_properties(self, targets=None, block=None):
         warnings.warn(_prop_warn)
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.clear_properties, targets=targets, block=block)
     
     #---------------------------------------------------------------------------
     # IMultiEngine related methods
     #---------------------------------------------------------------------------
     
     def get_ids(self):
         """
         Returns the ids of currently registered engines.
         """
         result = self._bcft(self.smultiengine.get_ids)
         return result
         
     #---------------------------------------------------------------------------
     # IMultiEngineCoordinator
     #---------------------------------------------------------------------------
              
     def scatter(self, key, seq, dist='b', flatten=False, targets=None, block=None):
         """
         Partition a Python sequence and send the partitions to a set of engines.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.scatter, key, seq, 
             dist, flatten, targets=targets, block=block)
     
     def gather(self, key, dist='b', targets=None, block=None):
         """
         Gather a partitioned sequence on a set of engines as a single local seq.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.gather, key, dist, 
             targets=targets, block=block)
     
     def raw_map(self, func, seq, dist='b', targets=None, block=None):
         """
         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`.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.raw_map, func, seq, 
             dist, targets=targets, block=block)
     
     def map(self, func, *sequences):
         """
         A parallel version of Python's builtin `map` function.
         
         This method applies a function to sequences of arguments.  It 
         follows the same syntax as the builtin `map`.
         
         This method creates a mapper objects by calling `self.mapper` with
         no arguments and then uses that mapper to do the mapping.  See
         the documentation of `mapper` for more details.
         """
         return self.mapper().map(func, *sequences)
     
     def mapper(self, dist='b', targets='all', block=None):
         """
         Create a mapper object that has a `map` method.
         
         This method returns an object that implements the `IMapper` 
         interface.  This method is a factory that is used to control how 
         the map happens.
         
         :Parameters:
             dist : str
                 What decomposition to use, 'b' is the only one supported
                 currently
             targets : str, int, sequence of ints
                 Which engines to use for the map
             block : boolean
                 Should calls to `map` block or not
         """
         return MultiEngineMapper(self, dist, targets, block)
     
     def parallel(self, dist='b', targets=None, block=None):
         """
         A decorator that turns a function into a parallel function.
         
         This can be used as:
         
         @parallel()
         def f(x, y)
             ...
         
         f(range(10), range(10))
         
         This causes f(0,0), f(1,1), ... to be called in parallel.
         
         :Parameters:
             dist : str
                 What decomposition to use, 'b' is the only one supported
                 currently
             targets : str, int, sequence of ints
                 Which engines to use for the map
             block : boolean
                 Should calls to `map` block or not
         """
         targets, block = self._findTargetsAndBlock(targets, block)       
         mapper = self.mapper(dist, targets, block)
         pf = ParallelFunction(mapper)
         return pf
     
     #---------------------------------------------------------------------------
     # IMultiEngineExtras
     #---------------------------------------------------------------------------
     
     def zip_pull(self, keys, targets=None, block=None):
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.zip_pull, keys, 
             targets=targets, block=block)
     
     def run(self, filename, targets=None, block=None):
         """
         Run a Python code in a file on the engines.
         
         :Parameters:
             filename : str
                 The name of the local file to run
             targets : id or list of ids
                 The engine to use for the execution
             block : boolean
                 If False, this method will return the actual result.  If False,
                 a `PendingResult` is returned which can be used to get the result
                 at a later time.
         """
         targets, block = self._findTargetsAndBlock(targets, block)
         return self._blockFromThread(self.smultiengine.run, filename,
             targets=targets, block=block)
     
     def benchmark(self, push_size=10000):
         """
         Run performance benchmarks for the current IPython cluster.
         
         This method tests both the latency of sending command and data to the
         engines as well as the throughput of sending large objects to the
         engines using push.  The latency is measured by having one or more
         engines execute the command 'pass'.  The throughput is measure by 
         sending an NumPy array of size `push_size` to one or more engines.
         
         These benchmarks will vary widely on different hardware and networks
         and thus can be used to get an idea of the performance characteristics
         of a particular configuration of an IPython controller and engines.
         
         This function is not testable within our current testing framework.
         """
         import timeit, __builtin__
         __builtin__._mec_self = self
         benchmarks = {}
         repeat = 3
         count = 10
 
         timer = timeit.Timer('_mec_self.execute("pass",0)')
         result = 1000*min(timer.repeat(repeat,count))/count
         benchmarks['single_engine_latency'] = (result,'msec')
 
         timer = timeit.Timer('_mec_self.execute("pass")')
         result = 1000*min(timer.repeat(repeat,count))/count
         benchmarks['all_engine_latency'] = (result,'msec')
 
         try:
             import numpy as np
         except:
             pass
         else:
             timer = timeit.Timer(
                 "_mec_self.push(d)",
                 "import numpy as np; d = dict(a=np.zeros(%r,dtype='float64'))" % push_size
             )
             result = min(timer.repeat(repeat,count))/count
             benchmarks['all_engine_push'] = (1e-6*push_size*8/result, 'MB/sec')
 
         try:
             import numpy as np
         except:
             pass
         else:
             timer = timeit.Timer(
                 "_mec_self.push(d,0)",
                 "import numpy as np; d = dict(a=np.zeros(%r,dtype='float64'))" % push_size
             )
             result = min(timer.repeat(repeat,count))/count
             benchmarks['single_engine_push'] = (1e-6*push_size*8/result, 'MB/sec')
 
         return benchmarks
 
 
 components.registerAdapter(FullBlockingMultiEngineClient,
             IFullSynchronousMultiEngine, IFullBlockingMultiEngineClient)
 
 
 
 

IPython/kernel/multienginefc.py

@@ -1,757 +1,760 b''
 # encoding: utf-8
 
 """
 Expose the multiengine controller over the Foolscap network protocol.
 """
 
 __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 cPickle as pickle
 from types import FunctionType
 
 from zope.interface import Interface, implements
 from twisted.internet import defer
 from twisted.python import components, failure
 
+try:
+    from foolscap.api import Referenceable
+except ImportError:
 from foolscap import Referenceable
 
 from IPython.kernel import error 
 from IPython.kernel import map as Map
 from IPython.kernel.parallelfunction import ParallelFunction
 from IPython.kernel.mapper import (
     MultiEngineMapper, 
     IMultiEngineMapperFactory,
     IMapper
 )
 from IPython.kernel.twistedutil import gatherBoth
 from IPython.kernel.multiengine import (
     IMultiEngine,
     IFullSynchronousMultiEngine,
     ISynchronousMultiEngine)
 from IPython.kernel.pendingdeferred import PendingDeferredManager
 from IPython.kernel.pickleutil import (
     canDict,
     canSequence, uncanDict, uncanSequence
 )
 
 from IPython.kernel.clientinterfaces import (
     IFCClientInterfaceProvider, 
     IBlockingClientAdaptor
 )
 
 # Needed to access the true globals from __main__.__dict__ 
 import __main__
 
 #-------------------------------------------------------------------------------
 # The Controller side of things
 #-------------------------------------------------------------------------------
 
 def packageResult(wrappedMethod):
     
     def wrappedPackageResult(self, *args, **kwargs):
         d = wrappedMethod(self, *args, **kwargs)
         d.addCallback(self.packageSuccess)
         d.addErrback(self.packageFailure)
         return d
     return wrappedPackageResult
 
 
 class IFCSynchronousMultiEngine(Interface):
     """Foolscap interface to `ISynchronousMultiEngine`.  
     
     The methods in this interface are similar to those of 
     `ISynchronousMultiEngine`, but their arguments and return values are pickled
     if they are not already simple Python types that can be send over XML-RPC.
     
     See the documentation of `ISynchronousMultiEngine` and `IMultiEngine` for 
     documentation about the methods.
     
     Most methods in this interface act like the `ISynchronousMultiEngine`
     versions and can be called in blocking or non-blocking mode.
     """
     pass
 
 
 class FCSynchronousMultiEngineFromMultiEngine(Referenceable):
     """Adapt `IMultiEngine` -> `ISynchronousMultiEngine` -> `IFCSynchronousMultiEngine`.
     """
     
     implements(IFCSynchronousMultiEngine, IFCClientInterfaceProvider)
     
     addSlash = True
     
     def __init__(self, multiengine):
         # Adapt the raw multiengine to `ISynchronousMultiEngine` before saving
         # it.  This allow this class to do two adaptation steps.
         self.smultiengine = ISynchronousMultiEngine(multiengine)
         self._deferredIDCallbacks = {}
     
     #---------------------------------------------------------------------------
     # Non interface methods
     #---------------------------------------------------------------------------
     
     def packageFailure(self, f):
         f.cleanFailure()
         return self.packageSuccess(f)
     
     def packageSuccess(self, obj):
         serial = pickle.dumps(obj, 2)
         return serial
     
     #---------------------------------------------------------------------------
     # Things related to PendingDeferredManager
     #---------------------------------------------------------------------------
     
     @packageResult
     def remote_get_pending_deferred(self, deferredID, block):
         d = self.smultiengine.get_pending_deferred(deferredID, block)
         try:
             callback = self._deferredIDCallbacks.pop(deferredID)
         except KeyError:
             callback = None
         if callback is not None:
             d.addCallback(callback[0], *callback[1], **callback[2])
         return d
        
     @packageResult
     def remote_clear_pending_deferreds(self):
         return defer.maybeDeferred(self.smultiengine.clear_pending_deferreds)
     
     def _addDeferredIDCallback(self, did, callback, *args, **kwargs):
         self._deferredIDCallbacks[did] = (callback, args, kwargs)
         return did
     
     #---------------------------------------------------------------------------
     # IEngineMultiplexer related methods
     #---------------------------------------------------------------------------
     
     @packageResult
     def remote_execute(self, lines, targets, block):
         return self.smultiengine.execute(lines, targets=targets, block=block)
     
     @packageResult    
     def remote_push(self, binaryNS, targets, block):
         try:
             namespace = pickle.loads(binaryNS)
         except:
             d = defer.fail(failure.Failure())
         else:
             d = self.smultiengine.push(namespace, targets=targets, block=block)
         return d
     
     @packageResult
     def remote_pull(self, keys, targets, block):
         d = self.smultiengine.pull(keys, targets=targets, block=block)
         return d
     
     @packageResult    
     def remote_push_function(self, binaryNS, targets, block):
         try:
             namespace = pickle.loads(binaryNS)
         except:
             d = defer.fail(failure.Failure())
         else:
             namespace = uncanDict(namespace)
             d = self.smultiengine.push_function(namespace, targets=targets, block=block)
         return d
     
     def _canMultipleKeys(self, result):
         return [canSequence(r) for r in result]
     
     @packageResult
     def remote_pull_function(self, keys, targets, block):
         def can_functions(r, keys):
             if len(keys)==1 or isinstance(keys, str):
                 result = canSequence(r)
             elif len(keys)>1:
                 result = [canSequence(s) for s in r]
             return result
         d = self.smultiengine.pull_function(keys, targets=targets, block=block)
         if block:
             d.addCallback(can_functions, keys)
         else:
             d.addCallback(lambda did: self._addDeferredIDCallback(did, can_functions, keys))
         return d
     
     @packageResult    
     def remote_push_serialized(self, binaryNS, targets, block):
         try:
             namespace = pickle.loads(binaryNS)
         except:
             d = defer.fail(failure.Failure())
         else:
             d = self.smultiengine.push_serialized(namespace, targets=targets, block=block)
         return d
     
     @packageResult
     def remote_pull_serialized(self, keys, targets, block):
         d = self.smultiengine.pull_serialized(keys, targets=targets, block=block)
         return d
     
     @packageResult
     def remote_get_result(self, i, targets, block):
         if i == 'None':
             i = None
         return self.smultiengine.get_result(i, targets=targets, block=block)
     
     @packageResult
     def remote_reset(self, targets, block):
         return self.smultiengine.reset(targets=targets, block=block)
     
     @packageResult
     def remote_keys(self, targets, block):
         return self.smultiengine.keys(targets=targets, block=block)
     
     @packageResult
     def remote_kill(self, controller, targets, block):
         return self.smultiengine.kill(controller, targets=targets, block=block)
     
     @packageResult
     def remote_clear_queue(self, targets, block):
         return self.smultiengine.clear_queue(targets=targets, block=block)
     
     @packageResult
     def remote_queue_status(self, targets, block):
         return self.smultiengine.queue_status(targets=targets, block=block)
     
     @packageResult
     def remote_set_properties(self, binaryNS, targets, block):
         try:
             ns = pickle.loads(binaryNS)
         except:
             d = defer.fail(failure.Failure())
         else:
             d = self.smultiengine.set_properties(ns, targets=targets, block=block)
         return d
     
     @packageResult
     def remote_get_properties(self, keys, targets, block):
         if keys=='None':
             keys=None
         return self.smultiengine.get_properties(keys, targets=targets, block=block)
     
     @packageResult
     def remote_has_properties(self, keys, targets, block):
         return self.smultiengine.has_properties(keys, targets=targets, block=block)
     
     @packageResult
     def remote_del_properties(self, keys, targets, block):
         return self.smultiengine.del_properties(keys, targets=targets, block=block)
     
     @packageResult
     def remote_clear_properties(self, targets, block):
         return self.smultiengine.clear_properties(targets=targets, block=block)
     
     #---------------------------------------------------------------------------
     # IMultiEngine related methods
     #---------------------------------------------------------------------------
     
     def remote_get_ids(self):
         """Get the ids of the registered engines.
         
         This method always blocks.
         """
         return self.smultiengine.get_ids()
     
     #---------------------------------------------------------------------------
     # IFCClientInterfaceProvider related methods
     #---------------------------------------------------------------------------
     
     def remote_get_client_name(self):
         return 'IPython.kernel.multienginefc.FCFullSynchronousMultiEngineClient'
 
 
 # The __init__ method of `FCMultiEngineFromMultiEngine` first adapts the
 # `IMultiEngine` to `ISynchronousMultiEngine` so this is actually doing a
 # two phase adaptation.
 components.registerAdapter(FCSynchronousMultiEngineFromMultiEngine,
             IMultiEngine, IFCSynchronousMultiEngine)
 
 
 #-------------------------------------------------------------------------------
 # The Client side of things
 #-------------------------------------------------------------------------------
 
 
 class FCFullSynchronousMultiEngineClient(object):
     
     implements(
         IFullSynchronousMultiEngine, 
         IBlockingClientAdaptor,
         IMultiEngineMapperFactory,
         IMapper
     )
     
     def __init__(self, remote_reference):
         self.remote_reference = remote_reference
         self._deferredIDCallbacks = {}
         # This class manages some pending deferreds through this instance.  This
         # is required for methods like gather/scatter as it enables us to
         # create our own pending deferreds for composite operations.
         self.pdm = PendingDeferredManager()
     
     #---------------------------------------------------------------------------
     # Non interface methods
     #---------------------------------------------------------------------------
                  
     def unpackage(self, r):
         return pickle.loads(r)
     
     #---------------------------------------------------------------------------
     # Things related to PendingDeferredManager
     #---------------------------------------------------------------------------
     
     def get_pending_deferred(self, deferredID, block=True):
         
         # Because we are managing some pending deferreds locally (through
         # self.pdm) and some remotely (on the controller), we first try the 
         # local one and then the remote one.
         if self.pdm.quick_has_id(deferredID):
             d = self.pdm.get_pending_deferred(deferredID, block)
             return d
         else:
             d = self.remote_reference.callRemote('get_pending_deferred', deferredID, block)
             d.addCallback(self.unpackage)
             try:
                 callback = self._deferredIDCallbacks.pop(deferredID)
             except KeyError:
                 callback = None
             if callback is not None:
                 d.addCallback(callback[0], *callback[1], **callback[2])
             return d
     
     def clear_pending_deferreds(self):
         
         # This clear both the local (self.pdm) and remote pending deferreds
         self.pdm.clear_pending_deferreds()
         d2 = self.remote_reference.callRemote('clear_pending_deferreds')
         d2.addCallback(self.unpackage)
         return d2
     
     def _addDeferredIDCallback(self, did, callback, *args, **kwargs):
         self._deferredIDCallbacks[did] = (callback, args, kwargs)
         return did
        
     #---------------------------------------------------------------------------
     # IEngineMultiplexer related methods
     #---------------------------------------------------------------------------
     
     def execute(self, lines, targets='all', block=True):
         d = self.remote_reference.callRemote('execute', lines, targets, block)
         d.addCallback(self.unpackage)
         return d
     
     def push(self, namespace, targets='all', block=True):
         serial = pickle.dumps(namespace, 2)
         d =  self.remote_reference.callRemote('push', serial, targets, block)
         d.addCallback(self.unpackage)
         return d
     
     def pull(self, keys, targets='all', block=True):
         d = self.remote_reference.callRemote('pull', keys, targets, block)
         d.addCallback(self.unpackage)
         return d
     
     def push_function(self, namespace, targets='all', block=True):
         cannedNamespace = canDict(namespace)
         serial = pickle.dumps(cannedNamespace, 2)
         d = self.remote_reference.callRemote('push_function', serial, targets, block)
         d.addCallback(self.unpackage)
         return d
     
     def pull_function(self, keys, targets='all', block=True):
         def uncan_functions(r, keys):
             if len(keys)==1 or isinstance(keys, str):
                 return uncanSequence(r)
             elif len(keys)>1:
                 return [uncanSequence(s) for s in r]
         d = self.remote_reference.callRemote('pull_function', keys, targets, block)
         if block:
             d.addCallback(self.unpackage)
             d.addCallback(uncan_functions, keys)
         else:
             d.addCallback(self.unpackage)
             d.addCallback(lambda did: self._addDeferredIDCallback(did, uncan_functions, keys))
         return d
     
     def push_serialized(self, namespace, targets='all', block=True):
         cannedNamespace = canDict(namespace)
         serial = pickle.dumps(cannedNamespace, 2)
         d =  self.remote_reference.callRemote('push_serialized', serial, targets, block)
         d.addCallback(self.unpackage)
         return d
     
     def pull_serialized(self, keys, targets='all', block=True):
         d = self.remote_reference.callRemote('pull_serialized', keys, targets, block)
         d.addCallback(self.unpackage)
         return d
         
     def get_result(self, i=None, targets='all', block=True):
         if i is None: # This is because None cannot be marshalled by xml-rpc
             i = 'None'
         d = self.remote_reference.callRemote('get_result', i, targets, block)
         d.addCallback(self.unpackage)
         return d
     
     def reset(self, targets='all', block=True):
         d = self.remote_reference.callRemote('reset', targets, block)
         d.addCallback(self.unpackage)
         return d        
     
     def keys(self, targets='all', block=True):
         d = self.remote_reference.callRemote('keys', targets, block)
         d.addCallback(self.unpackage)
         return d
     
     def kill(self, controller=False, targets='all', block=True):
         d = self.remote_reference.callRemote('kill', controller, targets, block)
         d.addCallback(self.unpackage)
         return d
     
     def clear_queue(self, targets='all', block=True):
         d = self.remote_reference.callRemote('clear_queue', targets, block)
         d.addCallback(self.unpackage)
         return d
     
     def queue_status(self, targets='all', block=True):
         d = self.remote_reference.callRemote('queue_status', targets, block)
         d.addCallback(self.unpackage)
         return d
     
     def set_properties(self, properties, targets='all', block=True):
         serial = pickle.dumps(properties, 2)
         d = self.remote_reference.callRemote('set_properties', serial, targets, block)
         d.addCallback(self.unpackage)
         return d
     
     def get_properties(self, keys=None, targets='all', block=True):
         if keys==None:
             keys='None'
         d = self.remote_reference.callRemote('get_properties', keys, targets, block)
         d.addCallback(self.unpackage)
         return d
     
     def has_properties(self, keys, targets='all', block=True):
         d = self.remote_reference.callRemote('has_properties', keys, targets, block)
         d.addCallback(self.unpackage)
         return d
     
     def del_properties(self, keys, targets='all', block=True):
         d = self.remote_reference.callRemote('del_properties', keys, targets, block)
         d.addCallback(self.unpackage)
         return d
     
     def clear_properties(self, targets='all', block=True):
         d = self.remote_reference.callRemote('clear_properties', targets, block)
         d.addCallback(self.unpackage)
         return d
     
     #---------------------------------------------------------------------------
     # IMultiEngine related methods
     #---------------------------------------------------------------------------
     
     def get_ids(self):
         d = self.remote_reference.callRemote('get_ids')
         return d
     
     #---------------------------------------------------------------------------
     # ISynchronousMultiEngineCoordinator related methods
     #---------------------------------------------------------------------------
 
     def _process_targets(self, targets):
         def create_targets(ids):
             if isinstance(targets, int):
                 engines = [targets]
             elif targets=='all':
                 engines = ids
             elif isinstance(targets, (list, tuple)):
                 engines = targets
             for t in engines:
                 if not t in ids:
                     raise error.InvalidEngineID("engine with id %r does not exist"%t)
             return engines
         
         d = self.get_ids()
         d.addCallback(create_targets)
         return d
     
     def scatter(self, key, seq, dist='b', flatten=False, targets='all', block=True):
         
         # Note: scatter and gather handle pending deferreds locally through self.pdm.
         # This enables us to collect a bunch fo deferred ids and make a secondary 
         # deferred id that corresponds to the entire group.  This logic is extremely
         # difficult to get right though.
         def do_scatter(engines):
             nEngines = len(engines)
             mapClass = Map.dists[dist]
             mapObject = mapClass()
             d_list = []
             # Loop through and push to each engine in non-blocking mode.
             # This returns a set of deferreds to deferred_ids
             for index, engineid in enumerate(engines):
                 partition = mapObject.getPartition(seq, index, nEngines)
                 if flatten and len(partition) == 1:
                     d = self.push({key: partition[0]}, targets=engineid, block=False)
                 else:
                     d = self.push({key: partition}, targets=engineid, block=False)
                 d_list.append(d)
             # Collect the deferred to deferred_ids
             d = gatherBoth(d_list,
                            fireOnOneErrback=0,
                            consumeErrors=1,
                            logErrors=0)
             # Now d has a list of deferred_ids or Failures coming
             d.addCallback(error.collect_exceptions, 'scatter')
             def process_did_list(did_list):
                 """Turn a list of deferred_ids into a final result or failure."""
                 new_d_list = [self.get_pending_deferred(did, True) for did in did_list]
                 final_d = gatherBoth(new_d_list,
                                      fireOnOneErrback=0,
                                      consumeErrors=1,
                                      logErrors=0)
                 final_d.addCallback(error.collect_exceptions, 'scatter')
                 final_d.addCallback(lambda lop: [i[0] for i in lop])
                 return final_d
             # Now, depending on block, we need to handle the list deferred_ids
             # coming down the pipe diferently.
             if block:
                 # If we are blocking register a callback that will transform the
                 # list of deferred_ids into the final result.
                 d.addCallback(process_did_list)
                 return d
             else:
                 # Here we are going to use a _local_ PendingDeferredManager.
                 deferred_id = self.pdm.get_deferred_id()
                 # This is the deferred we will return to the user that will fire
                 # with the local deferred_id AFTER we have received the list of 
                 # primary deferred_ids
                 d_to_return = defer.Deferred()
                 def do_it(did_list):
                     """Produce a deferred to the final result, but first fire the
                     deferred we will return to the user that has the local
                     deferred id."""
                     d_to_return.callback(deferred_id)
                     return process_did_list(did_list)
                 d.addCallback(do_it)
                 # Now save the deferred to the final result
                 self.pdm.save_pending_deferred(d, deferred_id)
                 return d_to_return
 
         d = self._process_targets(targets)
         d.addCallback(do_scatter)
         return d
 
     def gather(self, key, dist='b', targets='all', block=True):
         
         # Note: scatter and gather handle pending deferreds locally through self.pdm.
         # This enables us to collect a bunch fo deferred ids and make a secondary 
         # deferred id that corresponds to the entire group.  This logic is extremely
         # difficult to get right though.
         def do_gather(engines):
             nEngines = len(engines)
             mapClass = Map.dists[dist]
             mapObject = mapClass()
             d_list = []
             # Loop through and push to each engine in non-blocking mode.
             # This returns a set of deferreds to deferred_ids
             for index, engineid in enumerate(engines):
                 d = self.pull(key, targets=engineid, block=False)
                 d_list.append(d)
             # Collect the deferred to deferred_ids
             d = gatherBoth(d_list,
                            fireOnOneErrback=0,
                            consumeErrors=1,
                            logErrors=0)
             # Now d has a list of deferred_ids or Failures coming
             d.addCallback(error.collect_exceptions, 'scatter')
             def process_did_list(did_list):
                 """Turn a list of deferred_ids into a final result or failure."""
                 new_d_list = [self.get_pending_deferred(did, True) for did in did_list]
                 final_d = gatherBoth(new_d_list,
                                      fireOnOneErrback=0,
                                      consumeErrors=1,
                                      logErrors=0)
                 final_d.addCallback(error.collect_exceptions, 'gather')
                 final_d.addCallback(lambda lop: [i[0] for i in lop])
                 final_d.addCallback(mapObject.joinPartitions)
                 return final_d
             # Now, depending on block, we need to handle the list deferred_ids
             # coming down the pipe diferently.
             if block:
                 # If we are blocking register a callback that will transform the
                 # list of deferred_ids into the final result.
                 d.addCallback(process_did_list)
                 return d
             else:
                 # Here we are going to use a _local_ PendingDeferredManager.
                 deferred_id = self.pdm.get_deferred_id()
                 # This is the deferred we will return to the user that will fire
                 # with the local deferred_id AFTER we have received the list of 
                 # primary deferred_ids
                 d_to_return = defer.Deferred()
                 def do_it(did_list):
                     """Produce a deferred to the final result, but first fire the
                     deferred we will return to the user that has the local
                     deferred id."""
                     d_to_return.callback(deferred_id)
                     return process_did_list(did_list)
                 d.addCallback(do_it)
                 # Now save the deferred to the final result
                 self.pdm.save_pending_deferred(d, deferred_id)
                 return d_to_return
 
         d = self._process_targets(targets)
         d.addCallback(do_gather)
         return d
 
     def raw_map(self, func, sequences, 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`.
         """
         if not isinstance(sequences, (list, tuple)):
             raise TypeError('sequences must be a list or tuple')
         max_len = max(len(s) for s in sequences)
         for s in sequences:
             if len(s)!=max_len:
                 raise ValueError('all sequences must have equal length')
         if isinstance(func, FunctionType):
             d = self.push_function(dict(_ipython_map_func=func), targets=targets, block=False)
             d.addCallback(lambda did: self.get_pending_deferred(did, True))
             sourceToRun = '_ipython_map_seq_result = map(_ipython_map_func, *zip(*_ipython_map_seq))'
         elif isinstance(func, str):
             d = defer.succeed(None)
             sourceToRun = \
                 '_ipython_map_seq_result = map(%s, *zip(*_ipython_map_seq))' % func
         else:
             raise TypeError("func must be a function or str")
         
         d.addCallback(lambda _: self.scatter('_ipython_map_seq', zip(*sequences), dist, targets=targets))
         d.addCallback(lambda _: self.execute(sourceToRun, targets=targets, block=False))
         d.addCallback(lambda did: self.get_pending_deferred(did, True))
         d.addCallback(lambda _: self.gather('_ipython_map_seq_result', dist, targets=targets, block=block))
         return d
 
     def map(self, func, *sequences):
         """
         A parallel version of Python's builtin `map` function.
         
         This method applies a function to sequences of arguments.  It 
         follows the same syntax as the builtin `map`.
         
         This method creates a mapper objects by calling `self.mapper` with
         no arguments and then uses that mapper to do the mapping.  See
         the documentation of `mapper` for more details.
         """
         return self.mapper().map(func, *sequences)
     
     def mapper(self, dist='b', targets='all', block=True):
         """
         Create a mapper object that has a `map` method.
         
         This method returns an object that implements the `IMapper` 
         interface.  This method is a factory that is used to control how 
         the map happens.
         
         :Parameters:
             dist : str
                 What decomposition to use, 'b' is the only one supported
                 currently
             targets : str, int, sequence of ints
                 Which engines to use for the map
             block : boolean
                 Should calls to `map` block or not
         """
         return MultiEngineMapper(self, dist, targets, block)
 
     def parallel(self, dist='b', targets='all', block=True):
         """
         A decorator that turns a function into a parallel function.
         
         This can be used as:
         
         @parallel()
         def f(x, y)
             ...
         
         f(range(10), range(10))
         
         This causes f(0,0), f(1,1), ... to be called in parallel.
         
         :Parameters:
             dist : str
                 What decomposition to use, 'b' is the only one supported
                 currently
             targets : str, int, sequence of ints
                 Which engines to use for the map
             block : boolean
                 Should calls to `map` block or not
         """
         mapper = self.mapper(dist, targets, block)
         pf = ParallelFunction(mapper)
         return pf
     
     #---------------------------------------------------------------------------
     # ISynchronousMultiEngineExtras related methods
     #---------------------------------------------------------------------------
     
     def _transformPullResult(self, pushResult, multitargets, lenKeys):
         if not multitargets:
             result = pushResult[0]
         elif lenKeys > 1:
             result = zip(*pushResult)
         elif lenKeys is 1:
             result = list(pushResult)
         return result
     
     def zip_pull(self, keys, targets='all', block=True):
         multitargets = not isinstance(targets, int) and len(targets) > 1
         lenKeys = len(keys)
         d = self.pull(keys, targets=targets, block=block)
         if block:
             d.addCallback(self._transformPullResult, multitargets, lenKeys)
         else:
             d.addCallback(lambda did: self._addDeferredIDCallback(did, self._transformPullResult, multitargets, lenKeys))
         return d
     
     def run(self, fname, targets='all', block=True):
         fileobj = open(fname,'r')
         source = fileobj.read()
         fileobj.close()
         # if the compilation blows, we get a local error right away
         try:
             code = compile(source,fname,'exec')
         except:
             return defer.fail(failure.Failure()) 
         # Now run the code
         d = self.execute(source, targets=targets, block=block)
         return d
     
     #---------------------------------------------------------------------------
     # IBlockingClientAdaptor related methods
     #---------------------------------------------------------------------------
     
     def adapt_to_blocking_client(self):
         from IPython.kernel.multiengineclient import IFullBlockingMultiEngineClient
         return IFullBlockingMultiEngineClient(self)

IPython/kernel/taskclient.py

@@ -1,194 +1,198 b''
 # encoding: utf-8
 # -*- test-case-name: IPython.kernel.tests.test_taskcontrollerxmlrpc -*-
 
 """
 A blocking version of the task client.
 """
 
 __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 zope.interface import Interface, implements
 from twisted.python import components
+
+try:
+    from foolscap.api import DeadReferenceError
+except ImportError:
 from foolscap import DeadReferenceError
 
 from IPython.kernel.twistedutil import blockingCallFromThread
 from IPython.kernel import task, error
 from IPython.kernel.mapper import (
     SynchronousTaskMapper,
     ITaskMapperFactory,
     IMapper
 )
 from IPython.kernel.parallelfunction import (
     ParallelFunction, 
     ITaskParallelDecorator
 )
 
 #-------------------------------------------------------------------------------
 # The task client
 #-------------------------------------------------------------------------------
 
 class IBlockingTaskClient(Interface):
     """
     A vague interface of the blocking task client
     """
     pass
 
 class BlockingTaskClient(object):
     """
     A blocking task client that adapts a non-blocking one.
     """
     
     implements(
         IBlockingTaskClient,
         ITaskMapperFactory,
         IMapper,
         ITaskParallelDecorator
     )
     
     def __init__(self, task_controller):
         self.task_controller = task_controller
         self.block = True
 
     def _bcft(self, *args, **kwargs):
         try:
             result = blockingCallFromThread(*args, **kwargs)
         except DeadReferenceError:
             raise error.ConnectionError(
                 """A connection error has occurred in trying to connect to the
                 controller. This is usually caused by the controller dying or 
                 being restarted. To resolve this issue try recreating the 
                 task client."""
             )
         else:
             return result
 
     def run(self, task, block=False):
         """Run a task on the `TaskController`.
         
         See the documentation of the `MapTask` and `StringTask` classes for
         details on how to build a task of different types.
         
         :Parameters:
             task : an `ITask` implementer
         
         :Returns: The int taskid of the submitted task.  Pass this to 
             `get_task_result` to get the `TaskResult` object.
         """
         tid = self._bcft(self.task_controller.run, task)
         if block:
             return self.get_task_result(tid, block=True)
         else:
             return tid
     
     def get_task_result(self, taskid, block=False):
         """
         Get a task result by taskid.
         
         :Parameters:
             taskid : int
                 The taskid of the task to be retrieved.
             block : boolean
                 Should I block until the task is done?
         
         :Returns: A `TaskResult` object that encapsulates the task result.
         """
         return self._bcft(self.task_controller.get_task_result,
             taskid, block)
     
     def abort(self, taskid):
         """
         Abort a task by taskid.
         
         :Parameters:
             taskid : int
                 The taskid of the task to be aborted.
         """
         return self._bcft(self.task_controller.abort, taskid)
     
     def barrier(self, taskids):
         """Block until a set of tasks are completed.
         
         :Parameters:
             taskids : list, tuple
                 A sequence of taskids to block on.
         """
         return self._bcft(self.task_controller.barrier, taskids)
     
     def spin(self):
         """
         Touch the scheduler, to resume scheduling without submitting a task.
         
         This method only needs to be called in unusual situations where the
         scheduler is idle for some reason. 
         """
         return self._bcft(self.task_controller.spin)
     
     def queue_status(self, verbose=False):
         """
         Get a dictionary with the current state of the task queue.
         
         :Parameters:
             verbose : boolean
                 If True, return a list of taskids.  If False, simply give
                 the number of tasks with each status.
         
         :Returns:
             A dict with the queue status.
         """
         return self._bcft(self.task_controller.queue_status, verbose)
     
     def clear(self):
         """
         Clear all previously run tasks from the task controller.
         
         This is needed because the task controller keep all task results
         in memory.  This can be a problem is there are many completed
         tasks.  Users should call this periodically to clean out these
         cached task results.
         """
         return self._bcft(self.task_controller.clear)
     
     def map(self, func, *sequences):
         """
         Apply func to *sequences elementwise.  Like Python's builtin map.
         
         This version is load balanced.
         """
         return self.mapper().map(func, *sequences)
 
     def mapper(self, clear_before=False, clear_after=False, retries=0, 
                 recovery_task=None, depend=None, block=True):
         """
         Create an `IMapper` implementer with a given set of arguments.
         
         The `IMapper` created using a task controller is load balanced.
         
         See the documentation for `IPython.kernel.task.BaseTask` for 
         documentation on the arguments to this method.
         """
         return SynchronousTaskMapper(self, clear_before=clear_before, 
             clear_after=clear_after, retries=retries, 
             recovery_task=recovery_task, depend=depend, block=block)
     
     def parallel(self, clear_before=False, clear_after=False, retries=0, 
         recovery_task=None, depend=None, block=True):
         mapper = self.mapper(clear_before, clear_after, retries,
             recovery_task, depend, block)
         pf = ParallelFunction(mapper)
         return pf
 
 components.registerAdapter(BlockingTaskClient,
             task.ITaskController, IBlockingTaskClient)
 
 

IPython/kernel/taskfc.py

@@ -1,326 +1,329 b''
 # encoding: utf-8
 # -*- test-case-name: IPython.kernel.tests.test_taskxmlrpc -*-
 """A Foolscap interface to a TaskController.
 
 This class lets Foolscap clients talk to a TaskController.
 """
 
 __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 cPickle as pickle
 
 from zope.interface import Interface, implements
 from twisted.internet import defer
 from twisted.python import components
 
+try:
+    from foolscap.api import Referenceable
+except ImportError:
 from foolscap import Referenceable
 
 from IPython.kernel import task as taskmodule
 from IPython.kernel.clientinterfaces import (
     IFCClientInterfaceProvider, 
     IBlockingClientAdaptor
 )
 from IPython.kernel.mapper import (
     TaskMapper,
     ITaskMapperFactory,
     IMapper
 )
 from IPython.kernel.parallelfunction import (
     ParallelFunction, 
     ITaskParallelDecorator
 )
 
 #-------------------------------------------------------------------------------
 # The Controller side of things
 #-------------------------------------------------------------------------------
 
 
 class IFCTaskController(Interface):
     """Foolscap interface to task controller.
         
     See the documentation of `ITaskController` for more information.
     """
     def remote_run(binTask):
         """"""
     
     def remote_abort(taskid):
         """"""
         
     def remote_get_task_result(taskid, block=False):
         """"""
         
     def remote_barrier(taskids):
         """"""
     
     def remote_spin():
         """"""
     
     def remote_queue_status(verbose):
         """"""
     
     def remote_clear():
         """"""
 
 
 class FCTaskControllerFromTaskController(Referenceable):
     """
     Adapt a `TaskController` to an `IFCTaskController`
     
     This class is used to expose a `TaskController` over the wire using
     the Foolscap network protocol.
     """
     
     implements(IFCTaskController, IFCClientInterfaceProvider)
     
     def __init__(self, taskController):
         self.taskController = taskController
     
     #---------------------------------------------------------------------------
     # Non interface methods
     #---------------------------------------------------------------------------
     
     def packageFailure(self, f):
         f.cleanFailure()
         return self.packageSuccess(f)
     
     def packageSuccess(self, obj):
         serial = pickle.dumps(obj, 2)
         return serial
     
     #---------------------------------------------------------------------------
     # ITaskController related methods
     #---------------------------------------------------------------------------
     
     def remote_run(self, ptask):
         try:
             task = pickle.loads(ptask)
             task.uncan_task()
         except:
             d = defer.fail(pickle.UnpickleableError("Could not unmarshal task"))
         else:
             d = self.taskController.run(task)
         d.addCallback(self.packageSuccess)
         d.addErrback(self.packageFailure)
         return d
     
     def remote_abort(self, taskid):
         d = self.taskController.abort(taskid)
         d.addCallback(self.packageSuccess)
         d.addErrback(self.packageFailure)
         return d
     
     def remote_get_task_result(self, taskid, block=False):
         d = self.taskController.get_task_result(taskid, block)
         d.addCallback(self.packageSuccess)
         d.addErrback(self.packageFailure)
         return d
     
     def remote_barrier(self, taskids):
         d = self.taskController.barrier(taskids)
         d.addCallback(self.packageSuccess)
         d.addErrback(self.packageFailure)
         return d        
     
     def remote_spin(self):
         d = self.taskController.spin()
         d.addCallback(self.packageSuccess)
         d.addErrback(self.packageFailure)
         return d        
     
     def remote_queue_status(self, verbose):
         d = self.taskController.queue_status(verbose)
         d.addCallback(self.packageSuccess)
         d.addErrback(self.packageFailure)
         return d
     
     def remote_clear(self):
         return self.taskController.clear()
     
     def remote_get_client_name(self):
         return 'IPython.kernel.taskfc.FCTaskClient'
     
 components.registerAdapter(FCTaskControllerFromTaskController,
             taskmodule.ITaskController, IFCTaskController)
 
 
 #-------------------------------------------------------------------------------
 # The Client side of things
 #-------------------------------------------------------------------------------
 
 class FCTaskClient(object):
     """
     Client class for Foolscap exposed `TaskController`.
     
     This class is an adapter that makes a `RemoteReference` to a 
     `TaskController` look like an actual `ITaskController` on the client side.
 
     This class also implements `IBlockingClientAdaptor` so that clients can 
     automatically get a blocking version of this class.
     """
     
     implements(
         taskmodule.ITaskController, 
         IBlockingClientAdaptor,
         ITaskMapperFactory,
         IMapper,
         ITaskParallelDecorator
     )
     
     def __init__(self, remote_reference):
         self.remote_reference = remote_reference
     
     #---------------------------------------------------------------------------
     # Non interface methods
     #---------------------------------------------------------------------------
     
     def unpackage(self, r):
         return pickle.loads(r)
     
     #---------------------------------------------------------------------------
     # ITaskController related methods
     #---------------------------------------------------------------------------
     def run(self, task):
         """Run a task on the `TaskController`.
         
         See the documentation of the `MapTask` and `StringTask` classes for 
         details on how to build a task of different types.
         
         :Parameters:
             task : an `ITask` implementer
         
         :Returns: The int taskid of the submitted task.  Pass this to 
             `get_task_result` to get the `TaskResult` object.
         """
         assert isinstance(task, taskmodule.BaseTask), "task must be a Task object!"
         task.can_task()
         ptask = pickle.dumps(task, 2)
         task.uncan_task()
         d = self.remote_reference.callRemote('run', ptask)
         d.addCallback(self.unpackage)
         return d
     
     def get_task_result(self, taskid, block=False):
         """
         Get a task result by taskid.
         
         :Parameters:
             taskid : int
                 The taskid of the task to be retrieved.
             block : boolean
                 Should I block until the task is done?
         
         :Returns: A `TaskResult` object that encapsulates the task result.
         """
         d = self.remote_reference.callRemote('get_task_result', taskid, block)
         d.addCallback(self.unpackage)
         return d 
     
     def abort(self, taskid):
         """
         Abort a task by taskid.
         
         :Parameters:
             taskid : int
                 The taskid of the task to be aborted.
         """
         d = self.remote_reference.callRemote('abort', taskid)
         d.addCallback(self.unpackage)
         return d 
         
     def barrier(self, taskids):
         """Block until a set of tasks are completed.
         
         :Parameters:
             taskids : list, tuple
                 A sequence of taskids to block on.
         """
         d = self.remote_reference.callRemote('barrier', taskids)
         d.addCallback(self.unpackage)
         return d 
     
     def spin(self):
         """
         Touch the scheduler, to resume scheduling without submitting a task.
         
         This method only needs to be called in unusual situations where the
         scheduler is idle for some reason.
         """
         d = self.remote_reference.callRemote('spin')
         d.addCallback(self.unpackage)
         return d
     
     def queue_status(self, verbose=False):
         """
         Get a dictionary with the current state of the task queue.
         
         :Parameters:
             verbose : boolean
                 If True, return a list of taskids.  If False, simply give
                 the number of tasks with each status.
         
         :Returns:
             A dict with the queue status.
         """
         d = self.remote_reference.callRemote('queue_status', verbose)
         d.addCallback(self.unpackage)
         return d
     
     def clear(self):
         """
         Clear all previously run tasks from the task controller.
         
         This is needed because the task controller keep all task results
         in memory.  This can be a problem is there are many completed
         tasks.  Users should call this periodically to clean out these
         cached task results.
         """
         d = self.remote_reference.callRemote('clear')
         return d
     
     def adapt_to_blocking_client(self):
         """
         Wrap self in a blocking version that implements `IBlockingTaskClient.
         """
         from IPython.kernel.taskclient import IBlockingTaskClient
         return IBlockingTaskClient(self)
     
     def map(self, func, *sequences):
         """
         Apply func to *sequences elementwise.  Like Python's builtin map.
         
         This version is load balanced.
         """
         return self.mapper().map(func, *sequences)
     
     def mapper(self, clear_before=False, clear_after=False, retries=0, 
                 recovery_task=None, depend=None, block=True):
         """
         Create an `IMapper` implementer with a given set of arguments.
         
         The `IMapper` created using a task controller is load balanced.
         
         See the documentation for `IPython.kernel.task.BaseTask` for 
         documentation on the arguments to this method.
         """
         return TaskMapper(self, clear_before=clear_before, 
             clear_after=clear_after, retries=retries, 
             recovery_task=recovery_task, depend=depend, block=block)
     
     def parallel(self, clear_before=False, clear_after=False, retries=0, 
         recovery_task=None, depend=None, block=True):
         mapper = self.mapper(clear_before, clear_after, retries,
             recovery_task, depend, block)
         pf = ParallelFunction(mapper)
         return pf
 

docs/examples/kernel/mcdriver.py

@@ -1,71 +1,148 b''
 #!/usr/bin/env python
 """Run a Monte-Carlo options pricer in parallel."""
 
+#-----------------------------------------------------------------------------
+# Imports
+#-----------------------------------------------------------------------------
+
+import sys
+import time
 from IPython.kernel import client
 import numpy as np
 from mcpricer import price_options
+from matplotlib import pyplot as plt
+
+#-----------------------------------------------------------------------------
+# Setup parameters for the run
+#-----------------------------------------------------------------------------
+
+def ask_question(text, the_type, default):
+    s = '%s [%r]: ' % (text, the_type(default))
+    result = raw_input(s)
+    if result:
+        return the_type(result)
+    else:
+        return the_type(default)
+
+cluster_profile = ask_question("Cluster profile", str, "default")
+price = ask_question("Initial price", float, 100.0)
+rate = ask_question("Interest rate", float, 0.05)
+days = ask_question("Days to expiration", int, 260)
+paths = ask_question("Number of MC paths", int, 10000)
+n_strikes = ask_question("Number of strike values", int, 5)
+min_strike = ask_question("Min strike price", float, 90.0)
+max_strike = ask_question("Max strike price", float, 110.0)
+n_sigmas = ask_question("Number of volatility values", int, 5)
+min_sigma = ask_question("Min volatility", float, 0.1)
+max_sigma = ask_question("Max volatility", float, 0.4)
+
+strike_vals = np.linspace(min_strike, max_strike, n_strikes)
+sigma_vals = np.linspace(min_sigma, max_sigma, n_sigmas)
+
+#-----------------------------------------------------------------------------
+# Setup for parallel calculation
+#-----------------------------------------------------------------------------
 
 # The MultiEngineClient is used to setup the calculation and works with all
 # engine.
-mec = client.MultiEngineClient(profile='mycluster')
+mec = client.MultiEngineClient(profile=cluster_profile)
 
 # The TaskClient is an interface to the engines that provides dynamic load 
 # balancing at the expense of not knowing which engine will execute the code.
-tc = client.TaskClient(profile='mycluster')
+tc = client.TaskClient(profile=cluster_profile)
 
 # Initialize the common code on the engines. This Python module has the
 # price_options function that prices the options.
 mec.run('mcpricer.py')
 
-# Define the function that will make up our tasks. We basically want to
-# call the price_options function with all but two arguments (K, sigma)
-# fixed.
-def my_prices(K, sigma):
-    S = 100.0
-    r = 0.05
-    days = 260
-    paths = 100000
-    return price_options(S, K, sigma, r, days, paths)
-
-# Create arrays of strike prices and volatilities
-nK = 10
-nsigma = 10
-K_vals = np.linspace(90.0, 100.0, nK)
-sigma_vals = np.linspace(0.1, 0.4, nsigma)
-
-# Submit tasks to the TaskClient for each (K, sigma) pair as a MapTask.
-# The MapTask simply applies a function (my_prices) to the arguments:
-# my_prices(K, sigma) and returns the result.
+#-----------------------------------------------------------------------------
+# Perform parallel calculation
+#-----------------------------------------------------------------------------
+
+print "Running parallel calculation over strike prices and volatilities..."
+print "Strike prices: ", strike_vals
+print "Volatilities: ", sigma_vals
+sys.stdout.flush()
+
+# Submit tasks to the TaskClient for each (strike, sigma) pair as a MapTask.
+t1 = time.time()
 taskids = []
-for K in K_vals:
+for strike in strike_vals:
     for sigma in sigma_vals:
-        t = client.MapTask(my_prices, args=(K, sigma))
+        t = client.MapTask(
+            price_options, 
+            args=(price, strike, sigma, rate, days, paths)
+        )
         taskids.append(tc.run(t))
 
 print "Submitted tasks: ", len(taskids)
+sys.stdout.flush()
 
 # Block until all tasks are completed.
 tc.barrier(taskids)
+t2 = time.time()
+t = t2-t1
+
+print "Parallel calculation completed, time = %s s" % t
+print "Collecting results..."
 
 # Get the results using TaskClient.get_task_result.
 results = [tc.get_task_result(tid) for tid in taskids]
 
 # Assemble the result into a structured NumPy array.
-prices = np.empty(nK*nsigma,
+prices = np.empty(n_strikes*n_sigmas,
     dtype=[('ecall',float),('eput',float),('acall',float),('aput',float)]
 )
+
 for i, price_tuple in enumerate(results):
     prices[i] = price_tuple
-prices.shape = (nK, nsigma)
-K_vals, sigma_vals = np.meshgrid(K_vals, sigma_vals)
 
-def plot_options(sigma_vals, K_vals, prices):
+prices.shape = (n_strikes, n_sigmas)
+strike_mesh, sigma_mesh = np.meshgrid(strike_vals, sigma_vals)
+
+print "Results are available: strike_mesh, sigma_mesh, prices"
+print "To plot results type 'plot_options(sigma_mesh, strike_mesh, prices)'"
+
+#-----------------------------------------------------------------------------
+# Utilities
+#-----------------------------------------------------------------------------
+
+def plot_options(sigma_mesh, strike_mesh, prices):
     """
-    Make a contour plot of the option price in (sigma, K) space.
+    Make a contour plot of the option price in (sigma, strike) space.
     """
-    from matplotlib import pyplot as plt
-    plt.contourf(sigma_vals, K_vals, prices)
+    plt.figure(1)
+    
+    plt.subplot(221)
+    plt.contourf(sigma_mesh, strike_mesh, prices['ecall'])
+    plt.axis('tight')
+    plt.colorbar()
+    plt.title('European Call')
+    plt.ylabel("Strike Price")
+
+    plt.subplot(222)
+    plt.contourf(sigma_mesh, strike_mesh, prices['acall'])
+    plt.axis('tight')
     plt.colorbar()
-    plt.title("Option Price")
+    plt.title("Asian Call")
+
+    plt.subplot(223)
+    plt.contourf(sigma_mesh, strike_mesh, prices['eput'])
+    plt.axis('tight')
+    plt.colorbar()
+    plt.title("European Put")
     plt.xlabel("Volatility")
     plt.ylabel("Strike Price")
+
+    plt.subplot(224)
+    plt.contourf(sigma_mesh, strike_mesh, prices['aput'])
+    plt.axis('tight')
+    plt.colorbar()
+    plt.title("Asian Put")
+    plt.xlabel("Volatility")
+
+
+
+    
+
+