Suppress foolscap deprecation warnings, bug reported by Satra.

Now we import foolscap components according to their api.  I think
these four files contain all the naked 'from foolscap import'
statements we had, so all such warnings should be gone now.

IPython/kernel/enginefc.py

@@ -1,548 +1,554 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 os, time
 import cPickle as pickle
 
 from twisted.python import components, log, failure
 from twisted.python.failure import Failure
 from twisted.internet import defer, reactor, threads
 from twisted.internet.interfaces import IProtocolFactory
 from zope.interface import Interface, implements, Attribute
 
 from twisted.internet.base import DelayedCall
 DelayedCall.debug = True
 
-from foolscap import Referenceable, DeadReferenceError
+try:
+    # This is preferred in foolscap v > 0.4.3
+    from foolscap.api import Referenceable, DeadReferenceError
+except ImportError:
+    # Fallback for older versions
+    from foolscap import Referenceable, DeadReferenceError
+
 from foolscap.referenceable import RemoteReference
 
 from IPython.kernel.pbutil import packageFailure, unpackageFailure
 from IPython.kernel.util import printer
 from IPython.kernel.twistedutil import gatherBoth
 from IPython.kernel import newserialized
 from IPython.kernel.error import  ProtocolError
 from IPython.kernel import controllerservice
 from IPython.kernel.controllerservice import IControllerBase
 from IPython.kernel.engineservice import \
     IEngineBase, \
     IEngineQueued, \
     EngineService, \
     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,69 +1,74 b''
 # encoding: utf-8
 
 """Foolscap related utilities."""
 
 __docformat__ = "restructuredtext en"
 
 #-------------------------------------------------------------------------------
 #  Copyright (C) 2008  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-------------------------------------------------------------------------------
 
 #-------------------------------------------------------------------------------
 # Imports
 #-------------------------------------------------------------------------------
 
 import os
 
-from foolscap import Tub, UnauthenticatedTub
+try:
+    # This is preferred in foolscap v > 0.4.3
+    from foolscap.api import Tub, UnauthenticatedTub
+except ImportError:
+    # Fallback for older versions
+    from foolscap import Tub, UnauthenticatedTub
 
 def check_furl_file_security(furl_file, secure):
     """Remove the old furl_file if changing security modes."""
     
     if os.path.isfile(furl_file):
         f = open(furl_file, 'r')
         oldfurl = f.read().strip()
         f.close()
         if (oldfurl.startswith('pb://') and not secure) or (oldfurl.startswith('pbu://') and secure):
             os.remove(furl_file)
 
 def is_secure(furl):
     if is_valid(furl):
         if furl.startswith("pb://"):
             return True
         elif furl.startswith("pbu://"):
             return False
     else:
         raise ValueError("invalid furl: %s" % furl)
 
 def is_valid(furl):
     if isinstance(furl, str):
         if furl.startswith("pb://") or furl.startswith("pbu://"):
             return True
     else:
         return False
 
 def find_furl(furl_or_file):
     if isinstance(furl_or_file, str):
         if is_valid(furl_or_file):
             return furl_or_file
     if os.path.isfile(furl_or_file):
         furl = open(furl_or_file, 'r').read().strip()
         if is_valid(furl):
             return furl
     raise ValueError("not a furl or a file containing a furl: %s" % furl_or_file)
 
 # 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
 
 

IPython/kernel/multienginefc.py

@@ -1,757 +1,762 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, log
 
-from foolscap import Referenceable
+try:
+    # This is preferred in foolscap v > 0.4.3
+    from foolscap.api import Referenceable
+except ImportError:
+    # Fallback for older versions
+    from foolscap import Referenceable
 
 from IPython.kernel import error 
 from IPython.kernel.util import printer
 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 (MultiEngine,
     IMultiEngine,
     IFullSynchronousMultiEngine,
     ISynchronousMultiEngine)
 from IPython.kernel.multiengineclient import wrapResultList
 from IPython.kernel.pendingdeferred import PendingDeferredManager
 from IPython.kernel.pickleutil import (can, canDict,
     canSequence, uncan, 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/taskfc.py

@@ -1,340 +1,345 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
 import xmlrpclib, copy
 
 from zope.interface import Interface, implements
 from twisted.internet import defer
 from twisted.python import components, failure
 
-from foolscap import Referenceable
+try:
+    # This is preferred in foolscap v > 0.4.3
+    from foolscap.api import Referenceable
+except ImportError:
+    # Fallback for older versions
+    from foolscap import Referenceable
 
 from IPython.kernel.twistedutil import blockingCallFromThread
 from IPython.kernel import error, task as taskmodule, taskclient
 from IPython.kernel.pickleutil import can, uncan
 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(taskids=None):
         """"""
 
 
 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,taskids=None):
         d = self.taskController.clear(taskids)
         d.addCallback(self.packageSuccess)
         d.addErrback(self.packageFailure)
         return d
     
     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,taskids=None):
         """
         Clear previously run tasks from the task controller.
         :Parameters:
             taskids : list, tuple, None
                 A sequence of taskids whose results we should drop.
                 if None: clear all results
         
         :Returns:
             An int, the number of tasks cleared
         
         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', taskids)
         d.addCallback(self.unpackage)
         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