wip

Author: wxtim

cylc/flow/__init__.py

@@ -53,7 +53,7 @@ def environ_init():
 
 environ_init()
 
-__version__ = '8.5.0.dev'
+__version__ = '8.5.0.dev836'
 
 
 def iter_entry_points(entry_point_name):

cylc/flow/commands.py

@@ -462,17 +462,11 @@ async def force_trigger_tasks(
     flow: List[str],
     flow_wait: bool = False,
     flow_descr: Optional[str] = None,
-    on_resume: bool = False
 ):
     """Manual task trigger."""
     validate.is_tasks(tasks)
     validate.flow_opts(flow, flow_wait)
-    if on_resume:
-        LOG.warning(
-            "The --on-resume option is deprecated and will be removed "
-            "at Cylc 8.5."
-        )
     yield
     yield schd.pool.force_trigger_tasks(
-        tasks, flow, flow_wait, flow_descr, on_resume
+        tasks, flow, flow_wait, flow_descr
     )

cylc/flow/network/schema.py

@@ -2238,15 +2238,6 @@ class Meta:
         ''')
         resolver = partial(mutator, command='force_trigger_tasks')
 
-    class Arguments(TaskMutation.Arguments, FlowMutationArguments):
-        on_resume = Boolean(
-            default_value=False,
-            description=sstrip('''
-                If the workflow is paused, wait until it is resumed before
-                running the triggered task(s).
-            ''')
-        )
-
 
 def _mut_field(cls):
     """Convert a mutation class into a field.

cylc/flow/scheduler.py

@@ -1480,10 +1480,6 @@ def release_tasks_to_run(self) -> bool:
             if not self.is_paused:
                 # release queued tasks
                 pre_prep_tasks.update(self.pool.release_queued_tasks())
-                if self.pool.tasks_to_trigger_on_resume:
-                    # and manually triggered tasks to run once workflow resumed
-                    pre_prep_tasks.update(self.pool.tasks_to_trigger_on_resume)
-                    self.pool.tasks_to_trigger_on_resume = set()
 
         elif (
             (

cylc/flow/scripts/trigger.py

@@ -108,17 +108,6 @@ def get_option_parser() -> COP:
 
     add_flow_opts(parser)
 
-    parser.add_option(
-        "--on-resume",
-        help=(
-            "If the workflow is paused, wait until it is resumed before "
-            "running the triggered task(s). DEPRECATED - this will be "
-            "removed at Cylc 8.5."
-        ),
-        action="store_true",
-        default=False,
-        dest="on_resume"
-    )
     return parser
 
 
@@ -136,7 +125,6 @@ async def run(options: 'Values', workflow_id: str, *tokens_list):
             'flow': options.flow,
             'flowWait': options.flow_wait,
             'flowDescr': options.flow_descr,
-            'onResume': options.on_resume,
         }
     }
     return await pclient.async_request('graphql', mutation_kwargs)

cylc/flow/task_pool.py

@@ -216,7 +216,6 @@ def __init__(
 
         self.tasks_to_hold: Set[Tuple[str, 'PointBase']] = set()
         self.tasks_to_trigger_now: Set['TaskProxy'] = set()
-        self.tasks_to_trigger_on_resume: Set['TaskProxy'] = set()
 
     def set_stop_task(self, task_id):
         """Set stop after a task."""
@@ -2299,7 +2298,7 @@ def _get_flow_nums(
             for n in flow
         }
 
-    def _force_trigger(self, itask: 'TaskProxy', on_resume: bool = False):
+    def _force_trigger(self, itask: 'TaskProxy'):
         """Process a manually triggered task, ready for job submission.
 
         Assumes the task is in the pool.
@@ -2311,10 +2310,6 @@ def _force_trigger(self, itask: 'TaskProxy', on_resume: bool = False):
           - queue it, if the queue is limiting activity
           - run it, if the queue is not limiting activity
 
-        After state reset and queue handling:
-        - if on_resume is False, add the task to tasks_to_trigger_now
-        - if on_resume is True, add the task to tasks_to_trigger_on_resume
-
         The scheduler will release tasks from the tasks_to_trigger sets.
 
         """
@@ -2350,15 +2345,7 @@ def _force_trigger(self, itask: 'TaskProxy', on_resume: bool = False):
             # If not queued now, record the task as ready to run.
             itask.waiting_on_job_prep = True
 
-            if on_resume:
-                self.tasks_to_trigger_on_resume.add(itask)
-                # In case previously triggered without --on-resume.
-                # (It should have run already, but just in case).
-                self.tasks_to_trigger_now.discard(itask)
-            else:
-                self.tasks_to_trigger_now.add(itask)
-                # In case previously triggered with --on-resume.
-                self.tasks_to_trigger_on_resume.discard(itask)
+            self.tasks_to_trigger_now.add(itask)
 
         # Task may be set running before xtrigger is satisfied,
         # if so check/spawn if xtrigger sequential.
@@ -2370,7 +2357,6 @@ def force_trigger_tasks(
         flow: List[str],
         flow_wait: bool = False,
         flow_descr: Optional[str] = None,
-        on_resume: bool = False
     ):
         """Manually trigger tasks.
 
@@ -2406,7 +2392,7 @@ def force_trigger_tasks(
                 LOG.error(f"[{itask}] ignoring trigger - already active")
                 continue
             self.merge_flows(itask, flow_nums)
-            self._force_trigger(itask, on_resume)
+            self._force_trigger(itask)
 
         # Spawn and trigger inactive tasks.
         if not flow:
@@ -2441,7 +2427,7 @@ def force_trigger_tasks(
 
             # run it (or run it again for incomplete flow-wait)
             self.add_to_pool(itask)
-            self._force_trigger(itask, on_resume)
+            self._force_trigger(itask)
 
     def spawn_parentless_sequential_xtriggers(self):
         """Spawn successor(s) of parentless wall clock satisfied tasks."""

tests/integration/test_force_trigger.py

@@ -88,64 +88,3 @@ async def test_trigger_workflow_paused(
             level=logging.ERROR,
             contains="ignoring trigger - already active"
         )
-
-
-async def test_trigger_on_resume(
-    flow: 'Fixture',
-    scheduler: 'Fixture',
-    start: 'Fixture',
-    capture_submission: 'Fixture',
-    log_filter: Callable
-):
-    """
-    Test manual triggering on-resume option when the workflow is paused.
-
-    https://github.com/cylc/cylc-flow/issues/6192
-
-    """
-    id_ = flow({
-        'scheduling': {
-            'queues': {
-                'default': {
-                    'limit': 1,
-                },
-            },
-            'graph': {
-                'R1': '''
-                    a => x & y & z
-                ''',
-            },
-        },
-    })
-    schd = scheduler(id_, paused_start=True)
-
-    # start the scheduler (but don't set the main loop running)
-    async with start(schd):
-
-        # capture task submissions (prevents real submissions)
-        submitted_tasks = capture_submission(schd)
-
-        # paused at start-up so no tasks should be submitted
-        assert len(submitted_tasks) == 0
-
-        # manually trigger 1/x - it not should be submitted
-        schd.pool.force_trigger_tasks(['1/x'], [1], on_resume=True)
-        schd.release_tasks_to_run()
-        assert len(submitted_tasks) == 0
-
-        # manually trigger 1/y - it should not be submitted
-        # (queue limit reached)
-        schd.pool.force_trigger_tasks(['1/y'], [1], on_resume=True)
-        schd.release_tasks_to_run()
-        assert len(submitted_tasks) == 0
-
-        # manually trigger 1/y again - it should not be submitted
-        # (triggering a queued task runs it)
-        schd.pool.force_trigger_tasks(['1/y'], [1], on_resume=True)
-        schd.release_tasks_to_run()
-        assert len(submitted_tasks) == 0
-
-        # resume the workflow, both tasks should trigger now.
-        schd.resume_workflow()
-        schd.release_tasks_to_run()
-        assert len(submitted_tasks) == 2