Add more unit tests in the get_app function, document chaining (#4014)

# Description

I wanted to make sure that Parsl futures created with ParslPoolExecutor
work with Parsl function chaining.
As expected, they do. These changes make the compatibility between PPE
and other Parsl routes clear in the documentation too

# Changed Behaviour

No changes in functionality.

I did make a method public so the unit tests were clearer and open that
route as an acceptable option in a workflow I'm tinkering with.

# Fixes

N/A

## Type of change

- Update to human readable text: Documentation/error messages/comments

Author: WardLT

docs/index.rst

@@ -38,7 +38,7 @@ Parsl lets you chain functions together and will launch each function as inputs
 
 Start with the `configuration quickstart <quickstart.html#getting-started>`_ to learn how to tell Parsl how to use your computing resource,
 see if `a template configuration for your supercomputer <userguide/configuration/examples.html>`_ is already available,
-then explore the `parallel computing patterns <userguide/workflows/workflows.html>`_ to determine how to use parallelism best in your application.
+then explore the `parallel computing patterns <userguide/workflows/workflow.html>`_ to determine how to use parallelism best in your application.
 
 Parsl is an open-source code, and available on GitHub: https://github.com/parsl/parsl/
 

docs/reference.rst

@@ -13,6 +13,7 @@ Core
     parsl.app.app.join_app
     parsl.dataflow.futures.AppFuture
     parsl.dataflow.dflow.DataFlowKernelLoader
+    parsl.concurrent.ParslPoolExecutor
     parsl.dataflow.dependency_resolvers.DependencyResolver
     parsl.dataflow.dependency_resolvers.DEEP_DEPENDENCY_RESOLVER
     parsl.dataflow.dependency_resolvers.SHALLOW_DEPENDENCY_RESOLVER

parsl/concurrent/__init__.py

@@ -23,7 +23,26 @@ class ParslPoolExecutor(Executor, AbstractContextManager):
     2. Supplying an already-started Parsl :class:`~parsl.DataFlowKernel` (DFK).
        The executor assumes you will start and stop the Parsl DFK outside the Executor.
 
-    Note: Parsl does not support canceling tasks. The :meth:`map` method does not cancel work
+    The futures returned by :meth:`submit` and :meth:`map` are Parsl futures and will work
+    with the same function chaining mechanisms as when using Parsl with decorators.
+
+    .. code-block:: python
+
+        def f(x):
+            return x + 1
+
+        @python_app
+        def parity(x):
+            return 'odd' if x % 2 == 1 else 'even'
+
+        with ParslPoolExecutor(config=my_parsl_config) as executor:
+            future_1 = executor.submit(f, 1)
+            assert parity(future_1) == 'even'  # Function chaining, as expected
+
+            future_2 = executor.submit(f, future_1)
+            assert future_2.result() == 3  # Chaining works with `submit` too
+
+    Parsl does not support canceling tasks. The :meth:`map` method does not cancel work
     when one member of the run fails or a timeout is reached
     and :meth:`shutdown` does not cancel work on completion.
     """
@@ -63,7 +82,7 @@ def app_count(self):
         """Number of functions currently registered with the executor"""
         return len(self._app_cache)
 
-    def _get_app(self, fn: Callable) -> PythonApp:
+    def get_app(self, fn: Callable) -> PythonApp:
         """Create a PythonApp for a function
 
         Args:
@@ -78,19 +97,48 @@ def _get_app(self, fn: Callable) -> PythonApp:
         return app
 
     def submit(self, fn, *args, **kwargs):
+        """Submits a callable to be executed with the given arguments.
+
+        Schedules the callable to be executed as ``fn(*args, **kwargs)`` and returns
+        a Future instance representing the execution of the callable.
+
+        Returns:
+            A Future representing the given call.
+        """
+
         if self._dfk is None:
             raise RuntimeError('Executor has been shut down.')
-        app = self._get_app(fn)
+        app = self.get_app(fn)
         return app(*args, **kwargs)
 
     # TODO (wardlt): This override can go away when Parsl supports cancel
     def map(self, fn: Callable, *iterables: Iterable, timeout: Optional[float] = None, chunksize: int = 1) -> Iterator:
+        """Returns an iterator equivalent to map(fn, iter).
+
+        Args:
+            fn: A callable that will take as many arguments as there are
+                passed iterables.
+            timeout: The maximum number of seconds to wait. If None, then there
+                is no limit on the wait time.
+            chunksize: If greater than one, the iterables will be chopped into
+                chunks of size chunksize and submitted to the process pool.
+                If set to one, the items in the list will be sent one at a time.
+
+        Returns:
+            An iterator equivalent to: map(func, ``*iterables``) but the calls may
+            be evaluated out-of-order.
+
+        Raises:
+            TimeoutError: If the entire result iterator could not be generated
+                before the given timeout.
+            Exception: If ``fn(*args)`` raises for any values.
+        """
         # This is a version of the CPython 3.9 `.map` implementation modified to not use `cancel`
         if timeout is not None:
             end_time = timeout + time.monotonic()
 
         # Submit the applications
-        app = self._get_app(fn)
+        app = self.get_app(fn)
         fs = [app(*args) for args in zip(*iterables)]
 
         # Yield the futures as completed

parsl/tests/sites/test_concurrent.py

@@ -1,14 +1,26 @@
 """Tests of the interfaces to Python's concurrent library"""
 from pytest import mark, raises, warns
 
-from parsl import Config, HighThroughputExecutor, load
+from parsl import Config, HighThroughputExecutor, load, python_app
 from parsl.concurrent import ParslPoolExecutor
 
 
 def f(x):
     return x + 1
 
 
+def g(x):
+    return 2 * x
+
+
+@python_app
+def is_odd(x):
+    if x % 2 == 1:
+        return 1
+    else:
+        return 0
+
+
 def make_config():
     return Config(
         executors=[
@@ -68,4 +80,17 @@ def test_with_dfk():
     with load(config) as dfk, ParslPoolExecutor(dfk=dfk, executors=['test_executor']) as exc:
         future = exc.submit(f, 1)
         assert future.result() == 2
-        assert next(iter(exc._app_cache.values())).executors == ['test_executor'], 'Executors were not passed through'
+        assert exc.get_app(f).executors == ['test_executor']
+
+
+@mark.local
+def test_chaining():
+    """Make sure the executor functions can be chained together"""
+    config = make_config()
+
+    with ParslPoolExecutor(config) as exc:
+        future_odd = exc.submit(f, 10)
+        assert is_odd(future_odd).result()
+
+        future_even = exc.submit(g, future_odd)
+        assert not is_odd(future_even).result()