Flow-specific hold/release

changes.d/5698.feat.md

@@ -0,0 +1 @@
+Flow-specific task hold and release.

cylc/flow/data_store_mgr.py

@@ -1012,10 +1012,7 @@ def generate_ghost_task(
             id=tp_id,
             task=t_id,
             cycle_point=point_string,
-            is_held=(
-                (name, point)
-                in self.schd.pool.tasks_to_hold
-            ),
+            is_held=self.schd.pool.hold_mgr.is_held(name, point),
             depth=task_def.depth,
             name=name,
         )

cylc/flow/flow_mgr.py

@@ -20,6 +20,7 @@
 import datetime
 
 from cylc.flow import LOG
+from cylc.flow.exceptions import InputError
 from cylc.flow.workflow_db_mgr import WorkflowDatabaseManager
 
 
@@ -30,6 +31,15 @@
 FLOW_NONE = "none"
 
 
+def validate_flow_opt(val):
+    """Validate command line --flow opions."""
+    if val is not None:
+        try:
+            int(val)
+        except ValueError:
+            raise InputError(f"--flow={val}: value must be integer.")
+
+
 class FlowMgr:
     """Logic to manage flow counter and flow metadata."""
 

cylc/flow/network/schema.py

@@ -1735,7 +1735,9 @@ class Arguments:
             description='Hold all tasks after the specified cycle point.',
             required=True
         )
-
+        flow_num = Int(
+            description='Number of flow to hold.'
+        )
     result = GenericScalar()
 
 
@@ -2035,6 +2037,11 @@ class Meta:
         ''')
         resolver = mutator
 
+    class Arguments(TaskMutation.Arguments):
+        flow_num = Int(
+            description='Number of flow to hold.'
+        )
+
 
 class Release(Mutation, TaskMutation):
     class Meta:
@@ -2047,6 +2054,11 @@ class Meta:
         ''')
         resolver = mutator
 
+    class Arguments(TaskMutation.Arguments):
+        flow_num = Int(
+            description='Number of flow to release.'
+        )
+
 
 class Kill(Mutation, TaskMutation):
     # TODO: This should be a job mutation?

cylc/flow/rundb.py

@@ -319,6 +319,7 @@ class CylcWorkflowDAO:
         TABLE_TASKS_TO_HOLD: [
             ["name"],
             ["cycle"],
+            ["flow"],
         ],
     }
 
@@ -939,11 +940,11 @@ def select_task_prerequisites(
         stmt_args = [cycle, name, flow_nums]
         return list(self.connect().execute(stmt, stmt_args))
 
-    def select_tasks_to_hold(self) -> List[Tuple[str, str]]:
+    def select_tasks_to_hold(self) -> List[Tuple[str, str, str]]:
         """Return all tasks to hold stored in the DB."""
         stmt = rf'''
             SELECT
-                name, cycle
+                name, cycle, flow
             FROM
                 {self.TABLE_TASKS_TO_HOLD}
         '''  # nosec (table name is code constant)

cylc/flow/scheduler.py

@@ -905,7 +905,8 @@
     def queue_command(self, command: str, kwargs: dict) -> None:
         self.command_queue.put((
             command,
-            tuple(kwargs.values()), {}
+            (),
+            kwargs,
         ))
 
     async def process_command_queue(self) -> None:
@@ -1011,9 +1012,13 @@
         self.stop_mode = stop_mode
         self.update_data_store()
 
-    def command_release(self, task_globs: Iterable[str]) -> int:
+    def command_release(
+        self,
+        tasks: Iterable[str],
+        flow_num: Optional[int] = None
+    ) -> int:
         """Release held tasks."""
-        return self.pool.release_held_tasks(task_globs)
+        return self.pool.release_held_tasks(tasks, flow_num)
 
     def command_release_hold_point(self) -> None:
         """Release all held tasks and unset workflow hold after cycle point,
@@ -1025,57 +1030,68 @@
         """Resume paused workflow."""
         self.resume_workflow()
 
-    def command_poll_tasks(self, items: List[str]) -> int:
+    def command_poll_tasks(self, tasks: List[str]) -> int:
         """Poll pollable tasks or a task or family if options are provided."""
         if self.config.run_mode('simulation'):
             return 0
-        itasks, _, bad_items = self.pool.filter_task_proxies(items)
+        itasks, _, bad_items = self.pool.filter_task_proxies(tasks)
         self.task_job_mgr.poll_task_jobs(self.workflow, itasks)
         return len(bad_items)
 
-    def command_kill_tasks(self, items: List[str]) -> int:
+    def command_kill_tasks(self, tasks: List[str]) -> int:
         """Kill all tasks or a task/family if options are provided."""
-        itasks, _, bad_items = self.pool.filter_task_proxies(items)
+        itasks, _, bad_items = self.pool.filter_task_proxies(tasks)
         if self.config.run_mode('simulation'):
             for itask in itasks:
                 if itask.state(*TASK_STATUSES_ACTIVE):
                     itask.state_reset(TASK_STATUS_FAILED)
                     self.data_store_mgr.delta_task_state(itask)
             return len(bad_items)
-        self.task_job_mgr.kill_task_jobs(self.workflow, itasks)
+        to_kill = self.task_job_mgr.kill_task_jobs(self.workflow, itasks)
+        # Hold killed tasks to prevent automatic retry.
+        for itask in to_kill:
+            self.pool.hold_mgr.hold_active_task(itask)
         return len(bad_items)
 
-    def command_hold(self, task_globs: Iterable[str]) -> int:
+    def command_hold(
+        self,
+        tasks: Iterable[str],
+        flow_num: Optional[int] = None
+    ) -> int:
         """Hold specified tasks."""
-        return self.pool.hold_tasks(task_globs)
+        return self.pool.hold_tasks(tasks, flow_num)
 
-    def command_set_hold_point(self, point: str) -> None:
+    def command_set_hold_point(
+        self,
+        point: str,
+        flow_num: Optional[int] = None
+    ) -> None:
         """Hold all tasks after the specified cycle point."""
         cycle_point = TaskID.get_standardised_point(point)
         if cycle_point is None:
             raise CyclingError("Cannot set hold point to None")
         LOG.info(
             f"Setting hold cycle point: {cycle_point}\n"
             "All tasks after this point will be held.")
-        self.pool.set_hold_point(cycle_point)
+        self.pool.set_hold_point(cycle_point, flow_num)
 
     def command_pause(self) -> None:
         """Pause the workflow."""
         self.pause_workflow()
 
     @staticmethod
-    def command_set_verbosity(lvl: Union[int, str]) -> None:
+    def command_set_verbosity(level: Union[int, str]) -> None:
         """Set workflow verbosity."""
         try:
-            lvl = int(lvl)
-            LOG.setLevel(lvl)
+            level = int(level)
+            LOG.setLevel(level)
         except (TypeError, ValueError) as exc:
             raise CommandFailedError(exc)
-        cylc.flow.flags.verbosity = log_level_to_verbosity(lvl)
+        cylc.flow.flags.verbosity = log_level_to_verbosity(level)
 
-    def command_remove_tasks(self, items) -> int:
+    def command_remove_tasks(self, tasks) -> int:
         """Remove tasks."""
-        return self.pool.remove_tasks(items)
+        return self.pool.remove_tasks(tasks)
 
     async def command_reload_workflow(self) -> None:
         """Reload workflow configuration."""
@@ -1329,6 +1345,8 @@
         * Original workflow run time zone.
         """
         LOG.info('LOADING workflow parameters')
+        self.options.holdcp_flow = None  # (not CLI but needed on restart)
+
         for key, value in params:
             if value is None:
                 continue
@@ -1370,6 +1388,12 @@
             ):
                 self.options.holdcp = value
                 LOG.info(f"+ hold point = {value}")
+            elif (
+                key == self.workflow_db_mgr.KEY_HOLD_CYCLE_POINT_FLOW
+                and self.options.holdcp_flow is None
+            ):
+                self.options.holdcp_flow = value
+                LOG.info(f"+ hold point flow = {value}")
             elif key == self.workflow_db_mgr.KEY_STOP_CLOCK_TIME:
                 int_val = int(value)
                 msg = f"stop clock time = {int_val} ({time2str(int_val)})"

cylc/flow/scripts/hold.py

@@ -59,6 +59,7 @@
 from typing import TYPE_CHECKING
 
 from cylc.flow.exceptions import InputError
+from cylc.flow.flow_mgr import validate_flow_opt
 from cylc.flow.network.client_factory import get_client
 from cylc.flow.option_parsers import (
     FULL_ID_MULTI_ARG_DOC,
@@ -67,18 +68,21 @@
 from cylc.flow.terminal import cli_function
 from cylc.flow.network.multi import call_multi
 
+
 if TYPE_CHECKING:
     from optparse import Values
 
 
 HOLD_MUTATION = '''
 mutation (
   $wFlows: [WorkflowID]!,
-  $tasks: [NamespaceIDGlob]!
+  $tasks: [NamespaceIDGlob]!,
+  $flowNum: Int
 ) {
   hold (
     workflows: $wFlows,
-    tasks: $tasks
+    tasks: $tasks,
+    flowNum: $flowNum
   ) {
     result
   }
@@ -88,11 +92,13 @@
 SET_HOLD_POINT_MUTATION = '''
 mutation (
   $wFlows: [WorkflowID]!,
-  $point: CyclePoint!
+  $point: CyclePoint!,
+  $flowNum: Int
 ) {
   setHoldPoint (
     workflows: $wFlows,
-    point: $point
+    point: $point,
+    flowNum: $flowNum
   ) {
     result
   }
@@ -114,6 +120,11 @@ def get_option_parser() -> COP:
         help="Hold all tasks after this cycle point.",
         metavar="CYCLE_POINT", action="store", dest="hold_point_string")
 
+    parser.add_option(
+        "--flow",
+        help="Hold tasks that belong to a specific flow.",
+        metavar="INT", action="store", dest="flow_num")
+
     return parser
 
 
@@ -123,12 +134,14 @@ def _validate(options: 'Values', *task_globs: str) -> None:
         if task_globs:
             raise InputError(
                 "Cannot combine --after with Cylc/Task IDs.\n"
-                "`cylc hold --after` holds all tasks after the given "
-                "cycle point.")
+                "`cylc hold --after` holds ALL tasks after the given "
+                "cycle point. Can be used with `--flow`.")
     elif not task_globs:
         raise InputError(
             "Must define Cycles/Tasks. See `cylc hold --help`.")
 
+    validate_flow_opt(options.flow_num)
+
 
 async def run(options, workflow_id, *tokens_list):
     _validate(options, *tokens_list)
@@ -137,14 +150,18 @@ async def run(options, workflow_id, *tokens_list):
 
     if options.hold_point_string:
         mutation = SET_HOLD_POINT_MUTATION
-        args = {'point': options.hold_point_string}
+        args = {
+            'point': options.hold_point_string,
+            'flowNum': options.flow_num
+        }
     else:
         mutation = HOLD_MUTATION
         args = {
             'tasks': [
                 id_.relative_id_with_selectors
                 for id_ in tokens_list
-            ]
+            ],
+            'flowNum': options.flow_num
         }
 
     mutation_kwargs = {

cylc/flow/scripts/release.py

@@ -42,6 +42,7 @@
 from typing import TYPE_CHECKING
 
 from cylc.flow.exceptions import InputError
+from cylc.flow.flow_mgr import validate_flow_opt
 from cylc.flow.network.client_factory import get_client
 from cylc.flow.network.multi import call_multi
 from cylc.flow.option_parsers import (
@@ -57,11 +58,13 @@
 RELEASE_MUTATION = '''
 mutation (
   $wFlows: [WorkflowID]!,
-  $tasks: [NamespaceIDGlob]!
+  $tasks: [NamespaceIDGlob]!,
+  $flowNum: Int
 ) {
   release (
     workflows: $wFlows,
     tasks: $tasks,
+    flowNum: $flowNum
   ) {
     result
   }
@@ -97,6 +100,11 @@ def get_option_parser() -> COP:
             "if set."),
         action="store_true", dest="release_all")
 
+    parser.add_option(
+        "--flow",
+        help="Release tasks that belong to a specific flow.",
+        metavar="INT", action="store", dest="flow_num")
+
     return parser
 
 
@@ -111,6 +119,8 @@ def _validate(options: 'Values', *tokens_list: str) -> None:
                 "Must define Cycles/Tasks. See `cylc release --help`."
             )
 
+    validate_flow_opt(options.flow_num)
+
 
 async def run(options: 'Values', workflow_id, *tokens_list):
     _validate(options, *tokens_list)
@@ -126,7 +136,8 @@ async def run(options: 'Values', workflow_id, *tokens_list):
             'tasks': [
                 tokens.relative_id_with_selectors
                 for tokens in tokens_list
-            ]
+            ],
+            'flowNum': options.flow_num
         }
 
     mutation_kwargs = {