use canning hook in dependent

allows the dependency function to be called even before the real task arguments are unpacked.

IPython/parallel/controller/dependency.py

@@ -1,223 +1,225 b''
 """Dependency utilities
 
 Authors:
 
 * Min RK
 """
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2013  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.
 #-----------------------------------------------------------------------------
 
 from types import ModuleType
 
 from IPython.parallel.client.asyncresult import AsyncResult
 from IPython.parallel.error import UnmetDependency
 from IPython.parallel.util import interactive
 from IPython.utils import py3compat
 from IPython.utils.pickleutil import can, uncan
 
 class depend(object):
     """Dependency decorator, for use with tasks.
     
     `@depend` lets you define a function for engine dependencies
     just like you use `apply` for tasks.
     
     
     Examples
     --------
     ::
     
         @depend(df, a,b, c=5)
         def f(m,n,p)
         
         view.apply(f, 1,2,3)
     
     will call df(a,b,c=5) on the engine, and if it returns False or
     raises an UnmetDependency error, then the task will not be run
     and another engine will be tried.
     """
     def __init__(self, f, *args, **kwargs):
         self.f = f
         self.args = args
         self.kwargs = kwargs
     
     def __call__(self, f):
         return dependent(f, self.f, *self.args, **self.kwargs)
 
 class dependent(object):
     """A function that depends on another function.
     This is an object to prevent the closure used
     in traditional decorators, which are not picklable.
     """
 
     def __init__(self, f, df, *dargs, **dkwargs):
         self.f = f
         self.func_name = getattr(f, '__name__', 'f')
         self.df = df
         self.dargs = dargs
         self.dkwargs = dkwargs
-
-    def __call__(self, *args, **kwargs):
+    
+    def check_dependency(self):
         if self.df(*self.dargs, **self.dkwargs) is False:
             raise UnmetDependency()
+    
+    def __call__(self, *args, **kwargs):
         return self.f(*args, **kwargs)
     
     if not py3compat.PY3:
         @property
         def __name__(self):
             return self.func_name
 
 @interactive
 def _require(*modules, **mapping):
     """Helper for @require decorator."""
     from IPython.parallel.error import UnmetDependency
     from IPython.utils.pickleutil import uncan
     user_ns = globals()
     for name in modules:
         try:
             exec 'import %s' % name in user_ns
         except ImportError:
             raise UnmetDependency(name)
             
     for name, cobj in mapping.items():
         user_ns[name] = uncan(cobj, user_ns)
     return True
 
 def require(*objects, **mapping):
     """Simple decorator for requiring local objects and modules to be available
     when the decorated function is called on the engine.
     
     Modules specified by name or passed directly will be imported
     prior to calling the decorated function.
     
     Objects other than modules will be pushed as a part of the task.
     Functions can be passed positionally,
     and will be pushed to the engine with their __name__.
     Other objects can be passed by keyword arg.
     
     Examples
     --------
     
     In [1]: @require('numpy')
        ...: def norm(a):
        ...:     return numpy.linalg.norm(a,2)
 
     In [2]: foo = lambda x: x*x
     In [3]: @require(foo)
        ...: def bar(a):
        ...:     return foo(1-a)
     """
     names = []
     for obj in objects:
         if isinstance(obj, ModuleType):
             obj = obj.__name__
         
         if isinstance(obj, basestring):
             names.append(obj)
         elif hasattr(obj, '__name__'):
             mapping[obj.__name__] = obj
         else:
             raise TypeError("Objects other than modules and functions "
                 "must be passed by kwarg, but got: %s" % type(obj)
             )
     
     for name, obj in mapping.items():
         mapping[name] = can(obj)
     return depend(_require, *names, **mapping)
 
 class Dependency(set):
     """An object for representing a set of msg_id dependencies.
     
     Subclassed from set().
     
     Parameters
     ----------
     dependencies: list/set of msg_ids or AsyncResult objects or output of Dependency.as_dict()
         The msg_ids to depend on
     all : bool [default True]
         Whether the dependency should be considered met when *all* depending tasks have completed
         or only when *any* have been completed.
     success : bool [default True]
         Whether to consider successes as fulfilling dependencies.
     failure : bool [default False]
         Whether to consider failures as fulfilling dependencies.
     
     If `all=success=True` and `failure=False`, then the task will fail with an ImpossibleDependency
         as soon as the first depended-upon task fails.
     """
     
     all=True
     success=True
     failure=True
     
     def __init__(self, dependencies=[], all=True, success=True, failure=False):
         if isinstance(dependencies, dict):
             # load from dict
             all = dependencies.get('all', True)
             success = dependencies.get('success', success)
             failure = dependencies.get('failure', failure)
             dependencies = dependencies.get('dependencies', [])
         ids = []
         
         # extract ids from various sources:
         if isinstance(dependencies, (basestring, AsyncResult)):
             dependencies = [dependencies]
         for d in dependencies:
             if isinstance(d, basestring):
                 ids.append(d)
             elif isinstance(d, AsyncResult):
                 ids.extend(d.msg_ids)
             else:
                 raise TypeError("invalid dependency type: %r"%type(d))
         
         set.__init__(self, ids)
         self.all = all
         if not (success or failure):
             raise ValueError("Must depend on at least one of successes or failures!")
         self.success=success
         self.failure = failure
     
     def check(self, completed, failed=None):
         """check whether our dependencies have been met."""
         if len(self) == 0:
             return True
         against = set()
         if self.success:
             against = completed
         if failed is not None and self.failure:
             against = against.union(failed)
         if self.all:
             return self.issubset(against)
         else:
             return not self.isdisjoint(against)
     
     def unreachable(self, completed, failed=None):
         """return whether this dependency has become impossible."""
         if len(self) == 0:
             return False
         against = set()
         if not self.success:
             against = completed
         if failed is not None and not self.failure:
             against = against.union(failed)
         if self.all:
             return not self.isdisjoint(against)
         else:
             return self.issubset(against)
         
     
     def as_dict(self):
         """Represent this dependency as a dict. For json compatibility."""
         return dict(
             dependencies=list(self),
             all=self.all,
             success=self.success,
             failure=self.failure
         )
 
 
 __all__ = ['depend', 'require', 'dependent', 'Dependency']
 

IPython/utils/pickleutil.py

@@ -1,333 +1,338 b''
 # encoding: utf-8
 
 """Pickle related utilities. Perhaps this should be called 'can'."""
 
 __docformat__ = "restructuredtext en"
 
 #-------------------------------------------------------------------------------
 #  Copyright (C) 2008-2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-------------------------------------------------------------------------------
 
 #-------------------------------------------------------------------------------
 # Imports
 #-------------------------------------------------------------------------------
 
 import copy
 import logging
 import sys
 from types import FunctionType
 
 try:
     import cPickle as pickle
 except ImportError:
     import pickle
 
 try:
     import numpy
 except:
     numpy = None
 
 import codeutil
 import py3compat
 from importstring import import_item
 
 from IPython.config import Application
 
 if py3compat.PY3:
     buffer = memoryview
     class_type = type
 else:
     from types import ClassType
     class_type = (type, ClassType)
 
 #-------------------------------------------------------------------------------
 # Classes
 #-------------------------------------------------------------------------------
 
 
 class CannedObject(object):
     def __init__(self, obj, keys=[], hook=None):
         self.keys = keys
         self.obj = copy.copy(obj)
         self.hook = can(hook)
         for key in keys:
             setattr(self.obj, key, can(getattr(obj, key)))
         
         self.buffers = []
 
     def get_object(self, g=None):
         if g is None:
             g = {}
         obj = self.obj
         for key in self.keys:
             setattr(obj, key, uncan(getattr(obj, key), g))
-
+        
         if self.hook:
             self.hook = uncan(self.hook, g)
             self.hook(obj, g)
         return self.obj
     
 
 class Reference(CannedObject):
     """object for wrapping a remote reference by name."""
     def __init__(self, name):
         if not isinstance(name, basestring):
             raise TypeError("illegal name: %r"%name)
         self.name = name
         self.buffers = []
 
     def __repr__(self):
         return "<Reference: %r>"%self.name
 
     def get_object(self, g=None):
         if g is None:
             g = {}
         
         return eval(self.name, g)
 
 
 class CannedFunction(CannedObject):
 
     def __init__(self, f):
         self._check_type(f)
         self.code = f.func_code
         if f.func_defaults:
             self.defaults = [ can(fd) for fd in f.func_defaults ]
         else:
             self.defaults = None
         self.module = f.__module__ or '__main__'
         self.__name__ = f.__name__
         self.buffers = []
 
     def _check_type(self, obj):
         assert isinstance(obj, FunctionType), "Not a function type"
 
     def get_object(self, g=None):
         # try to load function back into its module:
         if not self.module.startswith('__'):
             __import__(self.module)
             g = sys.modules[self.module].__dict__
 
         if g is None:
             g = {}
         if self.defaults:
             defaults = tuple(uncan(cfd, g) for cfd in self.defaults)
         else:
             defaults = None
         newFunc = FunctionType(self.code, g, self.__name__, defaults)
         return newFunc
 
 class CannedClass(CannedObject):
 
     def __init__(self, cls):
         self._check_type(cls)
         self.name = cls.__name__
         self.old_style = not isinstance(cls, type)
         self._canned_dict = {}
         for k,v in cls.__dict__.items():
             if k not in ('__weakref__', '__dict__'):
                 self._canned_dict[k] = can(v)
         if self.old_style:
             mro = []
         else:
             mro = cls.mro()
         
         self.parents = [ can(c) for c in mro[1:] ]
         self.buffers = []
 
     def _check_type(self, obj):
         assert isinstance(obj, class_type), "Not a class type"
 
     def get_object(self, g=None):
         parents = tuple(uncan(p, g) for p in self.parents)
         return type(self.name, parents, uncan_dict(self._canned_dict, g=g))
 
 class CannedArray(CannedObject):
     def __init__(self, obj):
         self.shape = obj.shape
         self.dtype = obj.dtype.descr if obj.dtype.fields else obj.dtype.str
         if sum(obj.shape) == 0:
             # just pickle it
             self.buffers = [pickle.dumps(obj, -1)]
         else:
             # ensure contiguous
             obj = numpy.ascontiguousarray(obj, dtype=None)
             self.buffers = [buffer(obj)]
     
     def get_object(self, g=None):
         data = self.buffers[0]
         if sum(self.shape) == 0:
             # no shape, we just pickled it
             return pickle.loads(data)
         else:
             return numpy.frombuffer(data, dtype=self.dtype).reshape(self.shape)
 
 
 class CannedBytes(CannedObject):
     wrap = bytes
     def __init__(self, obj):
         self.buffers = [obj]
     
     def get_object(self, g=None):
         data = self.buffers[0]
         return self.wrap(data)
 
 def CannedBuffer(CannedBytes):
     wrap = buffer
 
 #-------------------------------------------------------------------------------
 # Functions
 #-------------------------------------------------------------------------------
 
 def _logger():
     """get the logger for the current Application
     
     the root logger will be used if no Application is running
     """
     if Application.initialized():
         logger = Application.instance().log
     else:
         logger = logging.getLogger()
         if not logger.handlers:
             logging.basicConfig()
     
     return logger
 
 def _import_mapping(mapping, original=None):
     """import any string-keys in a type mapping
     
     """
     log = _logger()
     log.debug("Importing canning map")
     for key,value in mapping.items():
         if isinstance(key, basestring):
             try:
                 cls = import_item(key)
             except Exception:
                 if original and key not in original:
                     # only message on user-added classes
                     log.error("cannning class not importable: %r", key, exc_info=True)
                 mapping.pop(key)
             else:
                 mapping[cls] = mapping.pop(key)
 
 def istype(obj, check):
     """like isinstance(obj, check), but strict
     
     This won't catch subclasses.
     """
     if isinstance(check, tuple):
         for cls in check:
             if type(obj) is cls:
                 return True
         return False
     else:
         return type(obj) is check
 
 def can(obj):
     """prepare an object for pickling"""
     
     import_needed = False
     
     for cls,canner in can_map.iteritems():
         if isinstance(cls, basestring):
             import_needed = True
             break
         elif istype(obj, cls):
             return canner(obj)
     
     if import_needed:
         # perform can_map imports, then try again
         # this will usually only happen once
         _import_mapping(can_map, _original_can_map)
         return can(obj)
     
     return obj
 
 def can_class(obj):
     if isinstance(obj, class_type) and obj.__module__ == '__main__':
         return CannedClass(obj)
     else:
         return obj
 
 def can_dict(obj):
     """can the *values* of a dict"""
     if istype(obj, dict):
         newobj = {}
         for k, v in obj.iteritems():
             newobj[k] = can(v)
         return newobj
     else:
         return obj
 
 sequence_types = (list, tuple, set)
 
 def can_sequence(obj):
     """can the elements of a sequence"""
     if istype(obj, sequence_types):
         t = type(obj)
         return t([can(i) for i in obj])
     else:
         return obj
 
 def uncan(obj, g=None):
     """invert canning"""
     
     import_needed = False
     for cls,uncanner in uncan_map.iteritems():
         if isinstance(cls, basestring):
             import_needed = True
             break
         elif isinstance(obj, cls):
             return uncanner(obj, g)
     
     if import_needed:
         # perform uncan_map imports, then try again
         # this will usually only happen once
         _import_mapping(uncan_map, _original_uncan_map)
         return uncan(obj, g)
     
     return obj
 
 def uncan_dict(obj, g=None):
     if istype(obj, dict):
         newobj = {}
         for k, v in obj.iteritems():
             newobj[k] = uncan(v,g)
         return newobj
     else:
         return obj
 
 def uncan_sequence(obj, g=None):
     if istype(obj, sequence_types):
         t = type(obj)
         return t([uncan(i,g) for i in obj])
     else:
         return obj
 
+def _uncan_dependent_hook(dep, g=None):
+    dep.check_dependency()
+    
+def can_dependent(obj):
+    return CannedObject(obj, keys=('f', 'df'), hook=_uncan_dependent_hook)
 
 #-------------------------------------------------------------------------------
 # API dictionaries
 #-------------------------------------------------------------------------------
 
 # These dicts can be extended for custom serialization of new objects
 
 can_map = {
-    'IPython.parallel.dependent' : lambda obj: CannedObject(obj, keys=('f','df')),
+    'IPython.parallel.dependent' : can_dependent,
     'numpy.ndarray' : CannedArray,
     FunctionType : CannedFunction,
     bytes : CannedBytes,
     buffer : CannedBuffer,
     class_type : can_class,
 }
 
 uncan_map = {
     CannedObject : lambda obj, g: obj.get_object(g),
 }
 
 # for use in _import_mapping:
 _original_can_map = can_map.copy()
 _original_uncan_map = uncan_map.copy()