Merge pull request #6910 from hjoliver/group-trigger-followup

Group trigger: re-use already-completed outputs of active group start tasks.

Author: hjoliver

changes.d/6910.fix.md

@@ -0,0 +1 @@
+Trigger command: automatically handle already-completed outputs of triggered tasks.

cylc/flow/commands.py

@@ -735,6 +735,8 @@ def _force_trigger_tasks(
 
     warnings_flow_none = []
     warnings_has_job = []
+    active_completed_outputs = {}
+
     for itask in active:
         # Find active group start tasks (parentless, or with only off-group
         # prerequisites) and set all prerequisites (to trigger them now).
@@ -764,6 +766,12 @@ def _force_trigger_tasks(
                 )
                 continue
 
+            if itask.state(*TASK_STATUSES_ACTIVE):
+                for (label, msg, completed) in itask.state.outputs:
+                    if completed:
+                        active_completed_outputs[
+                            (str(itask.point), itask.tdef.name)] = (label, msg)
+
             if itask.state(TASK_STATUS_PREPARING, *TASK_STATUSES_ACTIVE):
                 # This is a live active group start task
                 warnings_has_job.append(str(itask))
@@ -826,7 +834,8 @@ def _force_trigger_tasks(
                 set_all=True  # prerequisites
             )
         else:
-            off_flow_prereqs = {
+            # Off-flow prereqs to satisfy, for the triggered flow:
+            prereqs_to_set = {
                 PrereqTuple(str(key.point), str(key.task), key.output)
                 for pre in tdef.get_prereqs(point)
                 for key in pre.keys()
@@ -838,15 +847,24 @@ def _force_trigger_tasks(
                 for key in pre.keys()
                 if (key.task, str(key.point)) in group_ids
             )
+            # Prereqs to satisfy, from already-completed outputs of active
+            # group start tasks, for the triggered flow.
+            prereqs_to_set.update({
+                PrereqTuple(str(key.point), str(key.task), key.output)
+                for pre in tdef.get_prereqs(point)
+                for key in pre.keys()
+                if (str(key.point), key.task) in active_completed_outputs
+            })
+
             if (
-                off_flow_prereqs
+                prereqs_to_set
                 or tdef.get_xtrigs(point)
                 or tdef.external_triggers
             ):
                 # Satisfy any off-group prereqs or ext/xtriggers to spawn task.
                 jtask = schd.pool._set_prereqs_tdef(
                     point, tdef,
-                    off_flow_prereqs,
+                    prereqs_to_set,
                     {"all": True},  # xtriggers
                     flow_nums,
                     flow_wait,

cylc/flow/scripts/trigger.py

@@ -17,34 +17,20 @@
 
 """cylc trigger [OPTIONS] ARGS
 
-Manually trigger tasks, respecting dependencies among them.
-
-Triggering individual tasks:
-  * Triggering an unqueued task queues it, triggering a queued task runs it.
-    So you many need to trigger an unqueued task twice to run it immediately.
-  * Live tasks (preparing, submitted, or running) can't be triggered.
-  * Triggered tasks can run even if the workflow is paused.
-
-Triggering a group of tasks at once (e.g. members of a sub-graph):
-  Cylc will automatically:
-  * Erase the run history of members, so they can re-run in the same flow.
-  * Identify group start tasks, and trigger them to start the flow.
-  * Identify off-group dependencies, and satisfy them to avoid a stall.
-  * Leave in-group dependencies to be satisfied by the triggered flow.
-
-  If the workflow is paused, group start tasks will trigger immediately; the
-  flow will continue downstream of them when you resume the workflow.
-
-  Beware of triggering live (preparing, submitted, or running) tasks:
-  * Live in-group tasks will be killed and their run history erased, to allow
-    them to re-run in the triggered flow.
-  * Live group-start tasks are left to run; they don't need to be retriggered.
-    WARNING: if they already completed outputs that other group members depend
-    on, you must manually satisfy those prerequisites again for the triggered
-    flow (run history erasure wipes out the original satisfied prerequisites).
-
-Flow number assignment in triggered tasks:
-  Active tasks (n=0) already have flows assigned; inactive tasks (n>0) do not.
+Trigger a group of one or more tasks, respecting dependencies among them.
+
+Prerequisites on tasks outside of the group will be satisfied automatically.
+
+Tasks will be removed if necessary to allow re-run without intervention, so
+triggered tasks that are preparing, submitted, or running may be killed.
+
+Tasks that lead into a group will run immediately even if the workflow is
+paused; activity will flow on from them once the workflow is resumed.
+
+Triggering an unqueued task queues it; triggering a queued task runs it.
+
+How flow numbers are assigned to triggered tasks:
+  Active tasks (n=0) already have assigned flows; inactive tasks (n>0) do not.
   * If an interdependent group of triggered tasks includes active tasks, the
     flow will be assigned the existing flow numbers of those active tasks.
   * Otherwise the flow will be assigned all current active flow numbers.
@@ -63,16 +49,6 @@
   # rerun sub-graph "a => b & c" in the same flow, ignoring "off => b"
   $ cylc trigger test //1/a //1/b //1/c
 
- Flow numbers of triggered tasks are determined as follows:
-  Active tasks (n=0) already have existing flow numbers.
-   * default: merge active and existing flow numbers
-   * --flow=INT or "new": merge given and existing flow numbers
-   * --flow="none": ERROR (not valid for already-active tasks)
-  Inactive tasks (n>0) do not have flow numbers assigned:
-   * default: run with all active flow numbers
-   * --flow=INT or "new": run with the given flow numbers
-   * --flow="none": run as no-flow (activity will not flow on downstream)
-
 """
 
 from functools import partial

tests/integration/test_force_trigger.py

@@ -37,7 +37,10 @@
     set_prereqs_and_outputs,
 )
 from cylc.flow.cycling.integer import IntegerPoint
-from cylc.flow.task_state import TASK_STATUS_WAITING
+from cylc.flow.task_state import (
+    TASK_STATUS_WAITING,
+    TASK_STATUS_RUNNING
+)
 
 
 async def test_trigger_workflow_paused(
@@ -625,6 +628,81 @@ async def test_trigger_n0_tasks(
         }
 
 
+async def test_replay_outputs(flow, scheduler, start, complete, log_filter):
+    """Triggered group start tasks re-emit (kind of) earlier outputs.
+
+    https://github.com/cylc/cylc-flow/issues/6858
+
+    Example graph:
+        a:started => b => end
+        k:kustom => l => end
+        k:kustom => offg
+
+    If I trigger a, b, k, l AFTER a:started and k:kustom have completed:
+        cylc trigger workflow //1/a //1/b //1/k //1/l
+
+    Then I should expect outputs `k:kustom` and `a:started` to be re-used
+    to satify b and l in the triggered flow, but NOT off-group task offg.
+    """
+    msg_prereq = '[1/{}:waiting(runahead)] prerequisite force-satisfied: 1/{}'
+    msg_spawned = "[1/{}:waiting(runahead)] => waiting"
+    msg_removed = "Removed tasks: 1/{}"
+
+    wid = flow({
+        'scheduling': {
+            'graph': {
+                'R1': """
+                    a:started => b => end
+                    k:kustom => l => end
+                    k:kustom => offg
+                """
+            }
+        },
+        'runtime': {
+            'a': {},
+            'k': {
+                'outputs': {'kustom': 'custom message'}
+            }
+        }
+    })
+    schd = scheduler(wid, paused_start=True)
+    async with start(schd):
+        # Set initial tasks a and k to "running" so they are recognized as
+        # live during the forthcoming trigger operation.
+
+        # Complete the a:started and k:kustom outputs.
+        schd.pool.set_prereqs_and_outputs(
+            ['1/a'], ['started'], [], []
+        )
+        schd.pool.set_prereqs_and_outputs(
+            ['1/k'], ['kustom'], [], []
+        )
+        # It should spawn b, l, and offg.
+        for task in ['b', 'l', 'offg']:
+            assert log_filter(contains=msg_spawned.format(task))
+
+        # Set a and k as running so they're recognized as live start tasks
+        # by the trigger operation.
+        for itask in schd.pool.get_tasks():
+            itask.state_reset(TASK_STATUS_RUNNING)
+
+        # Now trigger the group.
+        await run_cmd(
+            force_trigger_tasks(schd, ['1/a', '1/b', '1/k', '1/l'], [])
+        )
+        # It should remove b and l (in-group tasks)
+        for task in ['b', 'l']:
+            assert log_filter(contains=msg_removed.format(task))
+
+        # But they will be respawned immediately by re-satisfying dependence
+        # on the earlier outputs of a and k for in-group tasks:
+        assert log_filter(contains=msg_prereq.format('b', 'a:started'))
+        assert log_filter(contains=msg_prereq.format('l', 'k:custom message'))
+        # But not for the off-group task offg:
+        assert not log_filter(
+            contains=msg_prereq.format('offg', 'k:custom message'))
+
+
 async def test_trigger_with_sequential_task(flow, scheduler, run, log_filter):
     """It should trigger a failed sequential task.