Add pluggable DependencyResolvers

docs/reference.rst

@@ -14,6 +14,9 @@ Core
     parsl.dataflow.futures.AppFuture
     parsl.dataflow.dflow.DataFlowKernelLoader
     parsl.monitoring.MonitoringHub
+    parsl.dataflow.dependency_resolvers.DependencyResolver
+    parsl.dataflow.dependency_resolvers.DEEP_DEPENDENCY_RESOLVER
+    parsl.dataflow.dependency_resolvers.SHALLOW_DEPENDENCY_RESOLVER
 
 Configuration
 =============

docs/userguide/plugins.rst

@@ -86,3 +86,19 @@ from invoking a Parsl app. This includes as the return value of a
 
 An specific example of this is integrating Globus Compute tasks into a Parsl
 task graph. See :ref:`label-join-globus-compute`
+
+Dependency resolution
+---------------------
+
+When Parsl examines the arguments to an app, it uses a `DependencyResolver`.
+The default `DependencyResolver` will cause Parsl to wait for
+``concurrent.futures.Future`` instances (including `AppFuture` and
+`DataFuture`), and pass through other arguments without waiting.
+
+This behaviour is pluggable: Parsl comes with another dependency resolver,
+`DEEP_DEPENDENCY_RESOLVER` which knows about futures contained with structures
+such as tuples, lists, sets and dicts.
+
+This plugin interface might be used to interface other task-like or future-like
+objects to the Parsl dependency mechanism, by describing how they can be
+interpreted as a Future.

parsl/config.py

@@ -5,6 +5,7 @@
 from typing_extensions import Literal
 
 from parsl.utils import RepresentationMixin
+from parsl.dataflow.dependency_resolvers import DependencyResolver
 from parsl.executors.base import ParslExecutor
 from parsl.executors.threads import ThreadPoolExecutor
 from parsl.errors import ConfigurationError
@@ -35,6 +36,8 @@ class Config(RepresentationMixin, UsageInformation):
     checkpoint_period : str, optional
         Time interval (in "HH:MM:SS") at which to checkpoint completed tasks. Only has an effect if
         ``checkpoint_mode='periodic'``.
+    dependency_resolver: plugin point for custom dependency resolvers. Default: only resolve Futures,
+        using the `SHALLOW_DEPENDENCY_RESOLVER`.
     garbage_collect : bool. optional.
         Delete task records from DFK when tasks have completed. Default: True
     internal_tasks_max_threads : int, optional
@@ -88,6 +91,7 @@ def __init__(self,
                                         Literal['dfk_exit'],
                                         Literal['manual']] = None,
                  checkpoint_period: Optional[str] = None,
+                 dependency_resolver: Optional[DependencyResolver] = None,
                  garbage_collect: bool = True,
                  internal_tasks_max_threads: int = 10,
                  retries: int = 0,
@@ -123,6 +127,7 @@ def __init__(self,
         if checkpoint_mode == 'periodic' and checkpoint_period is None:
             checkpoint_period = "00:30:00"
         self.checkpoint_period = checkpoint_period
+        self.dependency_resolver = dependency_resolver
         self.garbage_collect = garbage_collect
         self.internal_tasks_max_threads = internal_tasks_max_threads
         self.retries = retries
@@ -152,4 +157,5 @@ def _validate_executors(self) -> None:
                 ', '.join(['label={}'.format(repr(d)) for d in duplicates])))
 
     def get_usage_information(self):
-        return {"executors_len": len(self.executors)}
+        return {"executors_len": len(self.executors),
+                "dependency_resolver": self.dependency_resolver is not None}

parsl/dataflow/dependency_resolvers.py

@@ -0,0 +1,115 @@
+from concurrent.futures import Future
+from dataclasses import dataclass
+from functools import singledispatch
+from typing import Callable
+
+
+@dataclass
+class DependencyResolver:
+    """A DependencyResolver describes how app dependencies can be resolved.
+    It is specified as two functions: `traverse_to_gather` which turns an
+    app parameter into a list of futures which must be waited for before
+    the task can be executed (for example, in the case of
+    `DEEP_DEPENDENCY_RESOLVER` this traverses structures such as lists to
+    find every contained ``Future``), and `traverse_to_unwrap` which turns an
+    app parameter into it's value to be passed to the app on execution
+    (for example in the case of `DEEP_DEPENDENCY_RESOLVER` this replaces a
+    list containing futures with a new list containing the values of those
+    resolved futures).
+
+    By default, Parsl will use `SHALLOW_DEPENDENCY_RESOLVER` which only
+    resolves Futures passed directly as arguments.
+    """
+    traverse_to_gather: Callable
+    traverse_to_unwrap: Callable
+
+
+@singledispatch
+def shallow_traverse_to_gather(o):
+    # objects in general do not expose futures that we can see
+    return []
+
+
+@singledispatch
+def shallow_traverse_to_unwrap(o):
+    # objects in general unwrap to themselves
+    return o
+
+
+@shallow_traverse_to_gather.register
+def _(fut: Future):
+    return [fut]
+
+
+@shallow_traverse_to_unwrap.register
+@singledispatch
+def _(fut: Future):
+    return fut.result()
+
+
+@singledispatch
+def deep_traverse_to_gather(o):
+    # objects in general do not expose futures that we can see
+    return []
+
+
+@singledispatch
+def deep_traverse_to_unwrap(o):
+    # objects in general unwrap to themselves
+    return o
+
+
+@deep_traverse_to_gather.register
+def _(fut: Future):
+    return [fut]
+
+
+@deep_traverse_to_unwrap.register
+@singledispatch
+def _(fut: Future):
+    return fut.result()
+
+
+@deep_traverse_to_gather.register(tuple)
+@deep_traverse_to_gather.register(list)
+@deep_traverse_to_gather.register(set)
+def _(iterable):
+    return [e for v in iterable for e in deep_traverse_to_gather(v) if isinstance(e, Future)]
+
+
+@deep_traverse_to_unwrap.register(tuple)
+@deep_traverse_to_unwrap.register(list)
+@deep_traverse_to_unwrap.register(set)
+@singledispatch
+def _(iterable):
+
+    type_ = type(iterable)
+    return type_(map(deep_traverse_to_unwrap, iterable))
+
+
+@deep_traverse_to_gather.register(dict)
+def _(dictionary):
+    futures = []
+    for key, value in dictionary.items():
+        if isinstance(key, Future):
+            futures.append(key)
+        if isinstance(value, Future):
+            futures.append(value)
+    return futures
+
+
+@deep_traverse_to_unwrap.register(dict)
+def _(dictionary):
+    unwrapped_dict = {}
+    for key, value in dictionary.items():
+        key = deep_traverse_to_unwrap(key)
+        value = deep_traverse_to_unwrap(value)
+        unwrapped_dict[key] = value
+    return unwrapped_dict
+
+
+DEEP_DEPENDENCY_RESOLVER = DependencyResolver(traverse_to_gather=deep_traverse_to_gather,
+                                              traverse_to_unwrap=deep_traverse_to_unwrap)
+
+SHALLOW_DEPENDENCY_RESOLVER = DependencyResolver(traverse_to_gather=shallow_traverse_to_gather,
+                                                 traverse_to_unwrap=shallow_traverse_to_unwrap)

parsl/dataflow/dflow.py

@@ -26,6 +26,7 @@
 from parsl.config import Config
 from parsl.data_provider.data_manager import DataManager
 from parsl.data_provider.files import File
+from parsl.dataflow.dependency_resolvers import SHALLOW_DEPENDENCY_RESOLVER
 from parsl.dataflow.errors import BadCheckpoint, DependencyError, JoinError
 from parsl.dataflow.futures import AppFuture
 from parsl.dataflow.memoization import Memoizer
@@ -203,6 +204,9 @@ def __init__(self, config: Config) -> None:
         self.tasks: Dict[int, TaskRecord] = {}
         self.submitter_lock = threading.Lock()
 
+        self.dependency_resolver = self.config.dependency_resolver if self.config.dependency_resolver is not None \
+            else SHALLOW_DEPENDENCY_RESOLVER
+
         atexit.register(self.atexit_cleanup)
 
     def __enter__(self):
@@ -852,8 +856,11 @@ def _gather_all_deps(self, args: Sequence[Any], kwargs: Dict[str, Any]) -> List[
         depends: List[Future] = []
 
         def check_dep(d: Any) -> None:
-            if isinstance(d, Future):
-                depends.extend([d])
+            try:
+                depends.extend(self.dependency_resolver.traverse_to_gather(d))
+            except Exception:
+                logger.exception("Exception in dependency_resolver.traverse_to_gather")
+                raise
 
         # Check the positional args
         for dep in args:
@@ -905,34 +912,27 @@ def append_failure(e: Exception, dep: Future) -> None:
         # Replace item in args
         new_args = []
         for dep in args:
-            if isinstance(dep, Future):
-                try:
-                    new_args.extend([dep.result()])
-                except Exception as e:
-                    append_failure(e, dep)
-            else:
-                new_args.extend([dep])
+            try:
+                new_args.extend([self.dependency_resolver.traverse_to_unwrap(dep)])
+            except Exception as e:
+                append_failure(e, dep)
 
         # Check for explicit kwargs ex, fu_1=<fut>
         for key in kwargs:
             dep = kwargs[key]
-            if isinstance(dep, Future):
-                try:
-                    kwargs[key] = dep.result()
-                except Exception as e:
-                    append_failure(e, dep)
+            try:
+                kwargs[key] = self.dependency_resolver.traverse_to_unwrap(dep)
+            except Exception as e:
+                append_failure(e, dep)
 
         # Check for futures in inputs=[<fut>...]
         if 'inputs' in kwargs:
             new_inputs = []
             for dep in kwargs['inputs']:
-                if isinstance(dep, Future):
-                    try:
-                        new_inputs.extend([dep.result()])
-                    except Exception as e:
-                        append_failure(e, dep)
-                else:
-                    new_inputs.extend([dep])
+                try:
+                    new_inputs.extend([self.dependency_resolver.traverse_to_unwrap(dep)])
+                except Exception as e:
+                    append_failure(e, dep)
             kwargs['inputs'] = new_inputs
 
         return new_args, kwargs, dep_failures
@@ -1037,6 +1037,8 @@ def submit(self,
 
         func = self._add_output_deps(executor, app_args, app_kwargs, app_fu, func)
 
+        logger.debug("Added output dependencies")
+
         # Replace the function invocation in the TaskRecord with whatever file-staging
         # substitutions have been made.
         task_record.update({
@@ -1048,8 +1050,10 @@ def submit(self,
 
         self.tasks[task_id] = task_record
 
+        logger.debug("Gathering dependencies")
         # Get the list of dependencies for the task
         depends = self._gather_all_deps(app_args, app_kwargs)
+        logger.debug("Gathered dependencies")
         task_record['depends'] = depends
 
         depend_descs = []