cylc pause

[MetRonnie]

These changes close #3897

• Implement cylc pause
• cylc play resumes a paused workflow
• Decouple cylc hold/release from workflow state, so they only hold/release tasks (or set/remove the hold cycle point for all tasks*)

*The [scheduling]hold after cycle point, cylc hold --after and cylc play --hold-after have remained pretty much the way they were, contrary to some of the discussion in #3897. However, they no longer affect the status of the task pool or scheduler, it just sets a point and if a spawned task is beyond that point it will be held

Requirements check-list

• I have read CONTRIBUTING.md and added my name as a Code Contributor.
• Contains logically grouped changes (else tidy your branch by rebase).
• Does not contain off-topic changes (use other PRs for other changes).
• Appropriate tests are included (unit and/or functional).
• Appropriate change log entry included.
• (master branch) I have opened a documentation PR at Cylc play/pause cylc-doc#204
• No dependency changes.

[MetRonnie]

Before I go through any more tests, can I ask @oliver-sanders and/or @hjoliver to review the changes to scheduler.py so far? 5a71c4d#diff-90711b005fe27b9f894e6cd09f08816d847dcf853518894c6b7e2139784cbcb8

[oliver-sanders]

Looks sensible to me, nice.

The only thing I'm unsure about is whether is_paused should cause should_process_tasks to return False or whether it should block the task processing itself in process_task_pool itself as there are other things going on in that method.

[MetRonnie]

@hjoliver What do you say?

cylc-flow/cylc/flow/scheduler.py

Lines 1549 to 1596 in 5a71c4d

	def should_process_tasks(self) -> bool:
	"""Return True if waiting tasks are ready."""
	# do we need to do a pass through the main task processing loop?
	process = False
	if self.is_paused:
	return False
	# New-style xtriggers.
	self.xtrigger_mgr.check_xtriggers(self.pool.get_tasks())
	if self.xtrigger_mgr.pflag:
	process = True
	self.xtrigger_mgr.pflag = False # reset
	# Old-style external triggers.
	self.broadcast_mgr.add_ext_triggers(self.ext_trigger_queue)
	for itask in self.pool.get_tasks():
	if (itask.state.external_triggers and
	self.broadcast_mgr.match_ext_trigger(itask)):
	process = True
	if self.task_events_mgr.pflag:
	# This flag is turned on by commands that change task state
	process = True
	self.task_events_mgr.pflag = False # reset
	if self.task_job_mgr.task_remote_mgr.ready:
	# This flag is turned on when a host init/select command completes
	process = True
	self.task_job_mgr.task_remote_mgr.ready = False # reset
	broadcast_mgr = self.task_events_mgr.broadcast_mgr
	broadcast_mgr.add_ext_triggers(self.ext_trigger_queue)
	for itask in self.pool.get_tasks():
	# External trigger matching and task expiry must be done
	# regardless, so they need to be in separate "if ..." blocks.
	if broadcast_mgr.match_ext_trigger(itask):
	process = True
	if self.pool.set_expired_task(itask, time()):
	process = True
	if all(itask.is_ready()):
	process = True
	if (
	self.config.run_mode('simulation') and
	self.pool.sim_time_check(self.message_queue)
	):
	process = True
	return process

I'm not sure what the side effects of this method are

[hjoliver]

(I still need to think about that question, ran out of time today sorry)

[oliver-sanders]

Cross-posting from element, here's the process_task_pool method:

cylc-flow/cylc/flow/scheduler.py

Lines 1210 to 1237 in 5a71c4d

	def process_task_pool(self):
	"""Process ALL TASKS whenever something has changed that might
	require renegotiation of dependencies, etc"""
	LOG.debug("BEGIN TASK PROCESSING")
	time0 = time()
	if self._get_events_conf(self.EVENT_INACTIVITY_TIMEOUT):
	self.set_suite_inactivity_timer()
	if self.stop_mode is None and self.auto_restart_time is None:
	itasks = self.pool.get_ready_tasks()
	if itasks:
	self.is_updated = True
	self.task_job_mgr.task_remote_mgr.rsync_includes = (
	self.config.get_validated_rsync_includes())
	for itask in self.task_job_mgr.submit_task_jobs(
	self.suite,
	itasks,
	self.curve_auth,
	self.client_pub_key_dir,
	self.config.run_mode('simulation')
	):
	# TODO log itask.flow_label here (beware effect on ref tests)
	LOG.info('[%s] -triggered off %s',
	itask, itask.state.get_resolved_dependencies())
	self.broadcast_mgr.expire_broadcast(self.pool.get_min_point())
	self.xtrigger_mgr.housekeep()
	self.suite_db_mgr.put_xtriggers(self.xtrigger_mgr.sat_xtrig)
	LOG.debug("END TASK PROCESSING (took %s seconds)" % (time() - time0))

Here's my suggestion:

    def process_task_pool(self):
        """Process ALL TASKS whenever something has changed that might
        require renegotiation of dependencies, etc"""
        LOG.debug("BEGIN TASK PROCESSING")
        time0 = time()
        if self._get_events_conf(self.EVENT_INACTIVITY_TIMEOUT):
            self.set_suite_inactivity_timer()
        if not self.is_paused and self.stop_mode is None and self.auto_restart_time is None:
            # ^ skip all this stuff if paused
            # ...
        self.broadcast_mgr.expire_broadcast(self.pool.get_min_point())
        self.xtrigger_mgr.housekeep()
        self.suite_db_mgr.put_xtriggers(self.xtrigger_mgr.sat_xtrig)
        LOG.debug("END TASK PROCESSING (took %s seconds)" % (time() - time0))

This would allow the other things in that function to work as I think they currently would for a "held" workflow.

[hjoliver]

Yep, I agree with @oliver-sanders 👍

[MetRonnie]

In this

cylc-flow/cylc/flow/scheduler.py

Lines 157 to 165 in c6fd91b

	# Dependency negotiation etc. will run after these commands
	PROC_CMDS = (
	'release_suite',
	'release_tasks',
	'kill_tasks',
	'force_spawn_children',
	'force_trigger_tasks',
	'reload_suite'
	)

release_suite and release_tasks are not used as command names, even on master, see

cylc-flow/cylc/flow/network/resolvers.py

Lines 650 to 659 in a582f22

	def release(self, tasks=None):
	"""Release (un-hold) the workflow."""
	self.schd.command_queue.put((
	"release",
	(),
	filter_none({
	'ids': tasks
	})
	))
	return (True, 'Command queued')

Should I change them to

diff --git a/cylc/flow/scheduler.py b/cylc/flow/scheduler.py
index 4bb79bfa0..027a2374c 100644
--- a/cylc/flow/scheduler.py
+++ b/cylc/flow/scheduler.py
@@ -156,8 +156,7 @@ class Scheduler:
     # Dependency negotiation etc. will run after these commands
     PROC_CMDS = (
-        'release_suite',
-        'release_tasks',
+        'release',
         'kill_tasks',
         'force_spawn_children',

[hjoliver]

Should I change them to ...

I think so, yes, but want to look at the associated code tomorrow...

[hjoliver]

Approving after a somewhat rushed review, but LGTM 🎉

[oliver-sanders]

Code review good, going to do some end-to-end testing, not expecting any issues.

[oliver-sanders]

All good on the end-to-end testing 👍.

One small conflict and MyPy is a bit angry since recent changes?

[MetRonnie]

Yes, the attribute declarations like pool: Optional[TaskPool] = None are not playing well. I'm trying to figure out the best solution

[MetRonnie]

If making these attributes without defaults, they have to be before attributes with defaults in an @dataclass class

[MetRonnie]

Strange that tests/f/platforms/07-localhost-set-if-not-in-globalcfg is failing on MacOS consistently while passing on Linux.

ok 1 - 07-localhost-set-if-not-in-globalcfg-validate
not ok 2 - 07-localhost-set-if-not-in-globalcfg-run
ok 3 - job-grep-ok

Suite log:

2021-03-18T22:42:10 DEBUG - PT5M suite inactivity timer starts NOW: 2021-03-18T22:42:10
2021-03-18T22:42:10 DEBUG - END TASK PROCESSING (took 0.00095367431640625 seconds)
2021-03-18T22:47:11 WARNING - suite timed out after inactivity for PT5M
2021-03-18T22:47:11 ERROR - Suite shutting down - Abort on suite inactivity is set
2021-03-18T22:47:11 WARNING - Orphaned task jobs:
	* foo.1 (submitted)

There is no job.err for foo.

Whereas on my local VLD I get

DEBUG - PT5M suite inactivity timer starts NOW: 2021-03-18T17:46:47Z
DEBUG - END TASK PROCESSING (took 0.0009641647338867188 seconds)
DEBUG - version: b'1.0', request_id: b'1', domain: '', address: 'xx.xxx.xx.xxx', identity: b'', mechanism: b'CURVE'
DEBUG - ALLOWED (CURVE) domain=* client_key=b'XXXXXXX'
DEBUG - ZAP reply code=b'200' text=b'OK'
INFO - [foo.1] status=submitted: (received)started at 2021-03-18T17:46:50Z  for job(01) flow(D)

@oliver-sanders can you take a look?

[oliver-sanders]

Sure.

[oliver-sanders]

That's because it sets job runner = at which we wouldn't expect to work on MacOS (the at daemon needs to be authenticated and enabled before it can work). Now why did this test not fail on master, or perhaps it did but no one noticed?

This should do it - https://github.com/cylc/cylc-flow/pull/4136/files

[MetRonnie]

Shouldn't be any need to re-run the tests

[oliver-sanders]

Fair'nough