added dependency decorator

IPython/zmq/parallel/dependency.py

@@ -0,0 +1,66
+"""Dependency utilities"""
+
+from IPython.external.decorator import decorator
+
+# flags
+ALL = 1 << 0
+ANY = 1 << 1
+HERE = 1 << 2
+ANYWHERE = 1 << 3
+
+class UnmetDependency(Exception):
+    pass
+
+class depend2(object):
+    """dependency decorator"""
+    def __init__(self, f, *args, **kwargs):
+        print "Inside __init__()"
+        self.dependency = (f,args,kwargs)
+    
+    def __call__(self, f, *args, **kwargs):
+        f._dependency = self.dependency
+        return decorator(_depend_wrapper, f)
+    
+class depend(object):
+    """Dependency decorator, for use with tasks."""
+    def __init__(self, f, *args, **kwargs):
+        print "Inside __init__()"
+        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 = self.f.func_name
+        self.df = df
+        self.dargs = dargs
+        self.dkwargs = dkwargs
+
+    def __call__(self, *args, **kwargs):
+        if self.df(*self.dargs, **self.dkwargs) is False:
+            raise UnmetDependency()
+        return self.f(*args, **kwargs)
+
+def evaluate_dependency(deps):
+    """Evaluate wheter dependencies are met.
+    
+    Parameters
+    ----------
+    deps : dict
+    """
+    pass
+
+def _check_dependency(flag):
+    pass
+
+
+__all__ = ['UnmetDependency', 'depend', 'evaluate_dependencies']
  No newline at end of file

IPython/zmq/parallel/streamkernel.py

@@ -16,9 +16,11 from code import CommandCompiler
 import zmq
 from zmq.eventloop import ioloop, zmqstream
 
+from IPython.zmq.completer import KernelCompleter
+
 from streamsession import StreamSession, Message, extract_header, serialize_object,\
                 unpack_apply_message
-from IPython.zmq.completer import KernelCompleter
+from dependency import UnmetDependency
 
 def printer(*args):
     pprint(args)
@@ -155,7 +157,7 class Kernel(object):
         for msg_type in ['execute_request', 'complete_request', 'apply_request']:
             self.queue_handlers[msg_type] = getattr(self, msg_type)
         
-        for msg_type in ['kill_request', 'abort_request']:
+        for msg_type in ['kill_request', 'abort_request']+self.queue_handlers.keys():
             self.control_handlers[msg_type] = getattr(self, msg_type)
 
     #-------------------- control handlers -----------------------------
@@ -273,13 +275,6 class Kernel(object):
     def check_aborted(self, msg_id):
         return msg_id in self.aborted
     
-    def unmet_dependencies(self, stream, idents, msg):
-        reply_type = msg['msg_type'].split('_')[0] + '_reply'
-        content = dict(status='resubmitted', reason='unmet dependencies')
-        reply_msg = self.session.send(stream, reply_type, 
-                    content=content, parent=msg, ident=idents)
-        ### TODO: actually resubmit it ###
-    
     #-------------------- queue handlers -----------------------------
     
     def execute_request(self, stream, ident, parent):
@@ -344,6 +339,7 class Kernel(object):
         # pyin_msg = self.session.msg(u'pyin',{u'code':code}, parent=parent)
         # self.pub_stream.send(pyin_msg)
         # self.session.send(self.pub_stream, u'pyin', {u'code':code},parent=parent)
+        sub = {'dependencies_met' : True}
         try:
             # allow for not overriding displayhook
             if hasattr(sys.displayhook, 'set_parent'):
@@ -393,15 +389,19 class Kernel(object):
             self.session.send(self.pub_stream, u'pyerr', exc_content, parent=parent)
             reply_content = exc_content
             result_buf = []
+            
+            if etype is UnmetDependency:
+                sub = {'dependencies_met' : False}
         else:
             reply_content = {'status' : 'ok'}
         # reply_msg = self.session.msg(u'execute_reply', reply_content, parent)
         # self.reply_socket.send(ident, zmq.SNDMORE)
         # self.reply_socket.send_json(reply_msg)
-        reply_msg = self.session.send(stream, u'apply_reply', reply_content, parent=parent, ident=ident,buffers=result_buf)
+        reply_msg = self.session.send(stream, u'apply_reply', reply_content, 
+                    parent=parent, ident=ident,buffers=result_buf, subheader=sub)
         # print>>sys.__stdout__, Message(reply_msg)
-        if reply_msg['content']['status'] == u'error':
-            self.abort_queues()
+        # if reply_msg['content']['status'] == u'error':
+        #     self.abort_queues()
     
     def dispatch_queue(self, stream, msg):
         self.control_stream.flush()
@@ -410,7 +410,6 class Kernel(object):
         
         header = msg['header']
         msg_id = header['msg_id']
-        dependencies = header.get('dependencies', [])
         if self.check_aborted(msg_id):
             self.aborted.remove(msg_id)
             # is it safe to assume a msg_id will not be resubmitted?
@@ -418,8 +417,6 class Kernel(object):
             reply_msg = self.session.send(stream, reply_type, 
                         content={'status' : 'aborted'}, parent=msg, ident=idents)
             return
-        if not self.check_dependencies(dependencies):
-            return self.unmet_dependencies(stream, idents, msg)
         handler = self.queue_handlers.get(msg['msg_type'], None)
         if handler is None:
             print >> sys.__stderr__, "UNKNOWN MESSAGE TYPE:", msg

IPython/zmq/parallel/streamsession.py

@@ -230,7 +230,7 def unpack_apply_message(bufs, g=None, copy=True):
     sargs = list(pickle.loads(bufs.pop(0)))
     skwargs = dict(pickle.loads(bufs.pop(0)))
     # print sargs, skwargs
-    f = cf.getFunction(g)
+    f = uncan(cf, g)
     for sa in sargs:
         if sa.data is None:
             m = bufs.pop(0)

IPython/zmq/pickleutil.py

@@ -19,10 +19,25 from types import FunctionType
 
 # contents of codeutil should either be in here, or codeutil belongs in IPython/util
 from IPython.kernel import codeutil
+from IPython.zmq.parallel.dependency import dependent
 
 class CannedObject(object):
-    pass
+    def __init__(self, obj, keys=[]):
+        self.keys = keys
+        self.obj = obj
+        for key in keys:
+            setattr(obj, key, can(getattr(obj, key)))
+            
     
+    def getObject(self, g=None):
+        if g is None:
+            g = globals()
+        for key in self.keys:
+            setattr(self.obj, key, uncan(getattr(self.obj, key), g))
+        return self.obj
+
+        
+
 class CannedFunction(CannedObject):
     
     def __init__(self, f):
@@ -41,6 +56,9 class CannedFunction(CannedObject):
 def can(obj):
     if isinstance(obj, FunctionType):
         return CannedFunction(obj)
+    elif isinstance(obj, dependent):
+        keys = ('f','df')
+        return CannedObject(obj, keys=keys)
     elif isinstance(obj,dict):
         return canDict(obj)
     elif isinstance(obj, (list,tuple)):
@@ -67,6 +85,8 def canSequence(obj):
 def uncan(obj, g=None):
     if isinstance(obj, CannedFunction):
         return obj.getFunction(g)
+    elif isinstance(obj, CannedObject):
+        return obj.getObject(g)
     elif isinstance(obj,dict):
         return uncanDict(obj)
     elif isinstance(obj, (list,tuple)):