update with new client registration reply

IPython/zmq/parallel/client.py

@@ -30,7 +30,7 b' import streamsession as ss'
 from clusterdir import ClusterDir, ClusterDirError
 # from remotenamespace import RemoteNamespace
 from view import DirectView, LoadBalancedView
-from dependency import Dependency, depend, require
+from dependency import Dependency, depend, require, dependent
 import error
 import map as Map
 from asyncresult import AsyncResult, AsyncMapResult
@@ -893,13 +893,11 b' class Client(object):'
         block : bool (default: self.block)
             Whether to wait for the result, or return immediately.
             False:
-                returns msg_id(s)
-                if multiple targets:
-                    list of ids
+                returns AsyncResult
             True:
                 returns actual result(s) of f(*args, **kwargs)
                 if multiple targets:
-                    dict of results, by engine ID
+                    list of results, matching `targets`
         targets : int,list of ints, 'all', None
             Specify the destination of the job.
             if None:
@@ -930,16 +928,13 b' class Client(object):'
         Returns
         -------
         if block is False:
-            if single target:
-                return msg_id
-            else:
-                return list of msg_ids
-                ? (should this be dict like block=True) ?
+            return AsyncResult wrapping msg_ids
+            output of AsyncResult.get() is identical to that of `apply(...block=True)`
         else:
             if single target:
-                return result of f(*args, **kwargs)
+                return result of `f(*args, **kwargs)`
             else:
-                return dict of results, keyed by engine
+                return list of results, matching `targets`
         """
         
         # defaults:
@@ -977,6 +972,18 b' class Client(object):'
                             after=None, follow=None, timeout=None):
         """The underlying method for applying functions in a load balanced
         manner, via the task queue."""
+        
+        if self._task_scheme == 'pure':
+            # pure zmq scheme doesn't support dependencies
+            msg = "Pure ZMQ scheduler doesn't support dependencies"
+            if (follow or after):
+                # hard fail on DAG dependencies
+                raise RuntimeError(msg)
+            if isinstance(f, dependent):
+                # soft warn on functional dependencies
+                warnings.warn(msg, RuntimeWarning)
+            
+        
         subheader = dict(after=after, follow=follow, timeout=timeout)
         bufs = ss.pack_apply_message(f,args,kwargs)
         content = dict(bound=bound)

IPython/zmq/parallel/controller.py

@@ -100,7 +100,7 b' class ControllerFactory(HubFactory):'
             
         else:
             self.log.info("task::using Python %s Task scheduler"%self.scheme)
-            sargs = (self.client_info['task'], self.engine_info['task'], self.monitor_url, self.client_info['notification'])
+            sargs = (self.client_info['task'][1], self.engine_info['task'], self.monitor_url, self.client_info['notification'])
             kwargs = dict(scheme=self.scheme,logname=self.log.name, loglevel=self.log.level, config=self.config)
             q = Process(target=launch_scheduler, args=sargs, kwargs=kwargs)
             q.daemon=True

docs/source/development/parallel_messages.txt

@@ -86,7 +86,7 b' Message type: ``connection_reply``::'
         'status' : 'ok', # or 'error'
         # if ok:
         'queue' : 'tcp://127.0.0.1:12345', # connection for client side of the MUX queue
-        'task' : 'tcp...', # addr for task queue, or None if no task queue running
+        'task' : ('lru','tcp...'), # routing scheme and addr for task queue (len 2 tuple)
         'query' : 'tcp...', # addr for methods to query the hub, like queue_request, etc.
         'control' : 'tcp...', # addr for control methods, like abort, etc.
     }