.. _label-joinapp:

Join Apps
=========

Join apps, defined with the ``@join_app`` decorator, are a form of app that can
launch other pieces of a workflow: for example a Parsl sub-workflow, or a task
that runs in some other system.

Parsl sub-workflows
-------------------

One reason for launching Parsl apps from inside a join app, rather than
directly in the main workflow code, is because the definitions of those tasks
are not known well enough at the start of the workflow.

For example, a workflow might run an expensive step to detect some objects
in an image, and then on each object, run a further expensive step. Because
the number of objects is not known at the start of the workflow, but instead
only after an expensive step has completed, the subsequent tasks cannot be
defined until after that step has completed.

In simple cases, the main workflow script can be stopped using
``Future.result()`` and join apps are not necessary, but in more complicated
cases, that approach can severely limit concurrency.

Join apps allow more naunced dependencies to be expressed that can help with:

* increased concurrency - helping with strong scaling
* more focused error propagation - allowing more of an ultimately failing workflow to complete
* more useful monitoring information

Using Futures from other components
-----------------------------------

Sometimes, a workflow might need to incorporate tasks from other systems that
run asynchronously but do not need a Parsl worker allocated for their entire
run. An example of this is delegating some work into Globus Compute: work can
be given to Globus Compute, but Parsl does not need to keep a worker allocated
to that task while it runs. Instead, Parsl can be told to wait for the ``Future``
returned by Globus Compute to complete.

Usage
-----

A `join_app` looks quite like a `python_app`, but should return one or more
``Future`` objects, rather than a value. Once the Python code has run, the
app will wait for those Futures to complete without occuping a Parsl worker,
and when those Futures complete, their contents will be the return value
of the `join_app`.

For example:

.. code-block:: python

  @python_app
  def some_app():
    return 3

  @join_app
  def example():
    x: Future = some_app()
    return x  # note that x is a Future, not a value

  assert example.result() == 3

Example of a Parsl sub-workflow
-------------------------------

This example workflow shows a preprocessing step, followed by
a middle stage that is chosen by the result of the pre-processing step
(either option 1 or option 2) followed by a know post-processing step.

.. code-block:: python

  @python_app
  def pre_process():
    return 3

  @python_app
  def option_one(x):
    return x*2

  @python_app
  def option_two(x):
    return (-x) * 2

  @join_app
  def process(x):
    if x > 0:
      return option_one(x)
    else:
      return option_two(x)

  @python_app
  def post_process(x):
    return str(x)

  assert post_process(process(pre_process()))).result() == "6"

* Why can't process be a regular python function?

``process`` needs to inspect the value of ``x`` to make a decision about
what app to launch. So it needs to defer execution until after the
pre-processing stage has completed. In Parsl, the way to defer that is
using apps: even though ``process`` is invoked at the start of the workflow,
it will execute later on, when the Future returned by ``pre_process`` has a
value.

* Why can't process be a @python_app?

A Python app, if run in a `parsl.executors.ThreadPoolExecutor`, can launch
more parsl apps; so a ``python_app`` implementation of process() would be able
to inspect x and choose and invoke the appropriate ``option_{one, two}``.

From launching the ``option_{one, two}`` app, the app body python code would
get a ``Future[int]`` - a ``Future`` that will eventually contain ``int``.

But, we want to invoke ``post_process`` at submission time near the start of
workflow so that Parsl knows about as many tasks as possible. But we don't
want it to execute until the value of the chosen ``option_{one, two}`` app
is known.

If we don't have join apps, how can we do this?

We could make process wait for ``option_{one, two}`` to complete, before
returning, like this:

.. code-block:: python

  @python_app
  def process(x):
    if x > 0:
      f = option_one(x)
    else:
      f = option_two(x)
    return f.result()

but this will block the worker running ``process`` until ``option_{one, two}``
has completed. If there aren't enough workers to run ``option_{one, two}`` this
can even deadlock. (principle: apps should not wait on completion of other 
apps and should always allow parsl to handle this through dependencies)

We could make process return the ``Future`` to the main workflow thread:

.. code-block:: python

  @python_app
  def process(x):
    if x > 0:
      f = option_one(x)
    else:
      f = option_two(x)
    return f  # f is a Future[int]

  # process(3) is a Future[Future[int]]


What comes out of invoking ``process(x)`` now is a nested ``Future[Future[int]]``
- it's a promise that eventually process will give you a promise (from
``option_one, two}``) that will eventually give you an int.

We can't pass that future into post_process... because post_process wants the
final int, and that future will complete before the int is ready, and that
(outer) future will have as its value the inner future (which won't be complete yet).

So we could wait for the result in the main workflow thread:

.. code-block:: python

  f_outer = process(pre_process())  # Future[Future[int]]
  f_inner = f_outer.result  # Future[int]
  result = post_process(f_inner)
  # result == "6"

But this now blocks the main workflow thread. If we really only need to run
these three lines, that's fine, but what about if we are in a for loop that
sets up 1000 parametrised iterations:

.. code-block:: python

  for x in [1..1000]:
    f_outer = process(pre_process(x))  # Future[Future[int]]
    f_inner = f_outer.result()  # Future[int]
    result = post_process(f_inner)

The ``for`` loop can only iterate after pre_processing is done for each
iteration - it is unnecessarily serialised by the ``.result()`` call, 
so that pre_processing cannot run in parallel.

So, the rule about not calling ``.result()`` applies in the main workflow thread
too.

What join apps add is the ability for parsl to unwrap that Future[Future[int]] into a
Future[int] in a "sensible" way (eg it doesn't need to block a worker).


.. _label-join-globus-compute:

Example of invoking a Futures-driven task from another system
-------------------------------------------------------------


This example shows launching some activity in another system, without
occupying a Parsl worker while that activity happens: in this example, work is
delegated to Globus Compute, which performs the work elsewhere. When the work
is completed, Globus Compute will put the result into the future that it
returns, and then (because the Parsl app is a ``@join_app``), that result will
be used as the result of the Parsl app.

As above, the motivation for doing this inside an app, rather than in the
top level is that sufficient information to launch the Globus Compute task
might not be available at start of the workflow.

This workflow will run a first stage, ``const_five``, on a Parsl worker,
then using the result of that stage, pass the result as a parameter to a
Globus Compute task, getting a ``Future`` from that submission. Then, the
results of the Globus Compute task will be passed onto a second Parsl
local task, ``times_two``.

.. code-block:: python

  import parsl
  from globus_compute_sdk import Executor

  tutorial_endpoint_uuid = '4b116d3c-1703-4f8f-9f6f-39921e5864df'
  gce = Executor(endpoint_id=tutorial_endpoint_uuid)

  def increment_in_funcx(n):
      return n+1

  @parsl.join_app
  def increment_in_parsl(n):
      future = gce.submit(increment_in_funcx, n)
      return future

  @parsl.python_app
  def times_two(n):
      return n*2

  @parsl.python_app
  def const_five():
      return 5

  parsl.load()

  workflow = times_two(increment_in_parsl(const_five()))

  r = workflow.result()

  assert r == (5+1)*2


Terminology
-----------

The term "join" comes from use of monads in functional programming, especially Haskell.