Skip Mode

changes.d/6039.feat.md

@@ -0,0 +1 @@
+Added a new task run mode "skip", which allows skipping of tasks in a running workflow as well as creation of dummy tasks as part of workflow design.

cylc/flow/cfgspec/workflow.py

@@ -56,7 +56,10 @@
 from cylc.flow.platforms import (
     fail_if_platform_and_host_conflict, get_platform_deprecated_settings,
     is_platform_definition_subshell)
+from cylc.flow.run_modes import RunMode
 from cylc.flow.task_events_mgr import EventData
+from cylc.flow.run_modes import TASK_CONFIG_RUN_MODES
+
 
 # Regex to check whether a string is a command
 REC_COMMAND = re.compile(r'(`|\$\()\s*(.*)\s*([`)])$')
@@ -1334,6 +1337,26 @@ def get_script_common_text(this: str, example: Optional[str] = None):
                     "[platforms][<platform name>]submission retry delays"
                 )
             )
+            Conf(
+                'run mode', VDR.V_STRING,
+                options=list(TASK_CONFIG_RUN_MODES),
+                default=RunMode.LIVE.value,
+                desc=f'''
+                    Run the task in one of the following modes:
+
+                    {RunMode.LIVE.value}:
+                        {RunMode.LIVE.describe()}
+                    {RunMode.SKIP.value}:
+                        {RunMode.SKIP.describe()}
+
+
+                    .. seealso::
+
+                       :ref:`task-run-modes`
+
+                    .. versionadded:: 8.4.0
+
+            ''')
             with Conf('meta', desc=r'''
                 Metadata for the task or task family.
 
@@ -1406,7 +1429,43 @@ def get_script_common_text(this: str, example: Optional[str] = None):
                     determine how an event handler responds to task failure
                     events.
                 ''')
+            with Conf('skip', desc='''
+                Task configuration for task :ref:`task-run-modes.skip`.
 
+                For a full description of skip run mode see
+                :ref:`task-run-modes.skip`.
+
+                .. versionadded:: 8.4.0
+            '''):
+                Conf(
+                    'outputs',
+                    VDR.V_STRING_LIST,
+                    desc='''
+                        Outputs to be emitted by a task in skip mode.
+
+                        * By default, all required outputs will be generated
+                          plus succeeded if success is optional.
+                        * If skip-mode outputs is specified and does not
+                          include either succeeded or failed then succeeded
+                          will be produced.
+                        * The outputs submitted and started are always
+                          produced and do not need to be defined in outputs.
+
+                        .. versionadded:: 8.4.0
+                    '''
+                )
+                Conf(
+                    'disable task event handlers',
+                    VDR.V_BOOLEAN,
+                    default=True,
+                    desc='''
+                        Task event handlers are turned off by default for
+                        skip mode tasks. Changing this setting to ``False``
+                        will re-enable task event handlers.
+
+                        .. versionadded:: 8.4.0
+                    '''
+                )
             with Conf('simulation', desc='''
                 Task configuration for workflow *simulation* and *dummy* run
                 modes.

cylc/flow/commands.py

@@ -76,9 +76,11 @@
 from cylc.flow.log_level import log_level_to_verbosity
 from cylc.flow.network.schema import WorkflowStopMode
 from cylc.flow.parsec.exceptions import ParsecError
+from cylc.flow.run_modes import RunMode
 from cylc.flow.task_id import TaskID
-from cylc.flow.task_state import TASK_STATUSES_ACTIVE, TASK_STATUS_FAILED
-from cylc.flow.workflow_status import RunMode, StopMode
+from cylc.flow.task_state import (
+    TASK_STATUSES_ACTIVE, TASK_STATUS_FAILED)
+from cylc.flow.workflow_status import StopMode
 
 from metomi.isodatetime.parsers import TimePointParser
 

cylc/flow/config.py

@@ -82,7 +82,8 @@
 )
 from cylc.flow.print_tree import print_tree
 from cylc.flow.task_qualifiers import ALT_QUALIFIERS
-from cylc.flow.simulation import configure_sim_modes
+from cylc.flow.run_modes.simulation import configure_sim_mode
+from cylc.flow.run_modes.skip import skip_mode_validate
 from cylc.flow.subprocctx import SubFuncContext
 from cylc.flow.task_events_mgr import (
     EventData,
@@ -99,6 +100,7 @@
     get_trigger_completion_variable_maps,
     trigger_to_completion_variable,
 )
+from cylc.flow.run_modes import RunMode
 from cylc.flow.task_trigger import TaskTrigger, Dependency
 from cylc.flow.taskdef import TaskDef
 from cylc.flow.unicode_rules import (
@@ -114,7 +116,6 @@
     WorkflowFiles,
     check_deprecation,
 )
-from cylc.flow.workflow_status import RunMode
 from cylc.flow.xtrigger_mgr import XtriggerCollator
 
 if TYPE_CHECKING:
@@ -513,9 +514,10 @@ def __init__(
 
         self.process_runahead_limit()
 
-        run_mode = self.run_mode()
+        run_mode = RunMode.get(self.options)
         if run_mode in {RunMode.SIMULATION, RunMode.DUMMY}:
-            configure_sim_modes(self.taskdefs.values(), run_mode)
+            for taskdef in self.taskdefs.values():
+                configure_sim_mode(taskdef.rtconfig, None, False)
 
         self.configure_workflow_state_polling_tasks()
 
@@ -567,6 +569,8 @@ def __init__(
 
         self.mem_log("config.py: end init config")
 
+        skip_mode_validate(self.taskdefs)
+
     @staticmethod
     def _warn_if_queues_have_implicit_tasks(
         config, taskdefs, max_warning_lines
@@ -1741,10 +1745,6 @@ def process_config_env(self):
                 ]
             )
 
-    def run_mode(self) -> str:
-        """Return the run mode."""
-        return RunMode.get(self.options)
-
     def _check_task_event_handlers(self):
         """Check custom event handler templates can be expanded.
 
@@ -2496,7 +2496,10 @@ def _get_taskdef(self, name: str) -> TaskDef:
 
         # Get the taskdef object for generating the task proxy class
         taskd = TaskDef(
-            name, rtcfg, self.run_mode(), self.start_point,
+            name,
+            rtcfg,
+            RunMode.get(self.options),
+            self.start_point,
             self.initial_point)
 
         # TODO - put all taskd.foo items in a single config dict

cylc/flow/data_messages.proto

@@ -128,6 +128,7 @@ message PbRuntime {
     optional string environment = 16;
     optional string outputs = 17;
     optional string completion = 18;
+    optional string run_mode = 19;
 }