Event for tasks that are late w.r.t. cycle point

Add event settings to a task to report itself as *late" with respect to
its (date-time) cycle point.

Author: matthewrmshin

doc/src/suite-design-guide/roadmap.tex

@@ -12,7 +12,7 @@ \subsection{List Item Override In Site Include-Files}
 \begin{lstlisting}
 [scheduling]
     [[special tasks]]
-        clock-triggered = get-data-a, get-data-b
+        clock-trigger = get-data-a, get-data-b
     #...
 #...
 \end{lstlisting}

lib/cylc/broadcast_mgr.py

@@ -179,7 +179,7 @@ def load_db_broadcast_states(self, row_idx, row):
 
     def match_ext_trigger(self, itask):
         """Match external triggers for a waiting task proxy."""
-        if not self.ext_triggers:
+        if not self.ext_triggers or not itask.state.external_triggers:
             return
         has_changed = False
         for trig, satisfied in itask.state.external_triggers.items():

lib/cylc/cfgspec/suite.py

@@ -397,6 +397,8 @@ def _coerce_parameter_list(value, keys, _):
                 'submission timeout': vdr(vtype='interval'),
 
                 'expired handler': vdr(vtype='string_list'),
+                'late offset': vdr(vtype='interval'),
+                'late handler': vdr(vtype='string_list'),
                 'submitted handler': vdr(vtype='string_list'),
                 'started handler': vdr(vtype='string_list'),
                 'succeeded handler': vdr(vtype='string_list'),

lib/cylc/rundb.py

@@ -187,6 +187,7 @@ class CylcSuiteDAO(object):
     TABLE_TASK_EVENTS = "task_events"
     TABLE_TASK_ACTION_TIMERS = "task_action_timers"
     TABLE_CHECKPOINT_ID = "checkpoint_id"
+    TABLE_TASK_LATE_FLAGS = "task_late_flags"
     TABLE_TASK_OUTPUTS = "task_outputs"
     TABLE_TASK_POOL = "task_pool"
     TABLE_TASK_POOL_CHECKPOINTS = "task_pool_checkpoints"
@@ -272,6 +273,11 @@ class CylcSuiteDAO(object):
             ["event"],
             ["message"],
         ],
+        TABLE_TASK_LATE_FLAGS: [
+            ["cycle", {"is_primary_key": True}],
+            ["name", {"is_primary_key": True}],
+            ["value", {"datatype": "INTEGER"}],
+        ],
         TABLE_TASK_OUTPUTS: [
             ["cycle", {"is_primary_key": True}],
             ["name", {"is_primary_key": True}],
@@ -664,8 +670,8 @@ def select_task_pool_for_restart(self, callback, id_key=None):
         """Select from task_pool+task_states+task_jobs for restart.
 
         Invoke callback(row_idx, row) on each row, where each row contains:
-            [cycle, name, spawned, status, hold_swap, submit_num, try_num,
-             user_at_host, time_submit, time_run, timeout]
+            [cycle, name, spawned, is_late, status, hold_swap, submit_num,
+             try_num, user_at_host, time_submit, time_run, timeout, outputs]
 
         If id_key is specified,
         select from task_pool table if id_key == CHECKPOINT_LATEST_ID.
@@ -676,6 +682,7 @@ def select_task_pool_for_restart(self, callback, id_key=None):
                 %(task_pool)s.cycle,
                 %(task_pool)s.name,
                 %(task_pool)s.spawned,
+                %(task_late_flags)s.value,
                 %(task_pool)s.status,
                 %(task_pool)s.hold_swap,
                 %(task_states)s.submit_num,
@@ -691,6 +698,10 @@ def select_task_pool_for_restart(self, callback, id_key=None):
                 %(task_states)s
             ON  %(task_pool)s.cycle == %(task_states)s.cycle AND
                 %(task_pool)s.name == %(task_states)s.name
+            LEFT OUTER JOIN
+                %(task_late_flags)s
+            ON  %(task_pool)s.cycle == %(task_late_flags)s.cycle AND
+                %(task_pool)s.name == %(task_late_flags)s.name
             LEFT OUTER JOIN
                 %(task_jobs)s
             ON  %(task_pool)s.cycle == %(task_jobs)s.cycle AND
@@ -708,6 +719,7 @@ def select_task_pool_for_restart(self, callback, id_key=None):
         form_data = {
             "task_pool": self.TABLE_TASK_POOL,
             "task_states": self.TABLE_TASK_STATES,
+            "task_late_flags": self.TABLE_TASK_LATE_FLAGS,
             "task_timeout_timers": self.TABLE_TASK_TIMEOUT_TIMERS,
             "task_jobs": self.TABLE_TASK_JOBS,
             "task_outputs": self.TABLE_TASK_OUTPUTS,

lib/cylc/scheduler.py

@@ -59,11 +59,13 @@
 from cylc.task_job_mgr import TaskJobManager
 from cylc.task_pool import TaskPool
 from cylc.task_proxy import TaskProxy, TaskProxySequenceBoundsError
-from cylc.task_state import TASK_STATUSES_ACTIVE, TASK_STATUS_FAILED
+from cylc.task_state import (
+    TASK_STATUSES_ACTIVE, TASK_STATUSES_NEVER_ACTIVE, TASK_STATUS_FAILED)
 from cylc.templatevars import load_template_vars
 from cylc.version import CYLC_VERSION
 from cylc.wallclock import (
-    get_current_time_string, get_seconds_as_interval_string)
+    get_current_time_string, get_seconds_as_interval_string,
+    get_time_string_from_unix_time as time2str)
 from cylc.profiler import Profiler
 
 
@@ -133,15 +135,15 @@ def __init__(self, is_restart, options, args):
         self.config = None
 
         self.is_restart = is_restart
-        self._cli_initial_point_string = None
-        self._cli_start_point_string = None
+        self.cli_initial_point_string = None
+        self.cli_start_point_string = None
         start_point_str = None
         if len(args) > 1:
             start_point_str = args[1]
         if getattr(self.options, 'warm', None):
-            self._cli_start_point_string = start_point_str
+            self.cli_start_point_string = start_point_str
         else:
-            self._cli_initial_point_string = start_point_str
+            self.cli_initial_point_string = start_point_str
         self.template_vars = load_template_vars(
             self.options.templatevars, self.options.templatevars_file)
 
@@ -382,16 +384,7 @@ def configure(self):
             self.load_tasks_for_run()
         self.profiler.log_memory("scheduler.py: after load_tasks")
 
-        self.suite_db_mgr.put_suite_params(
-            CYLC_VERSION,
-            self.task_job_mgr.task_remote_mgr.uuid_str,
-            self.run_mode,
-            str(cylc.flags.utc),
-            self.initial_point,
-            self.final_point,
-            self.pool.is_held,
-            self.config.cfg['cylc']['cycle point format'],
-            self._cli_start_point_string)
+        self.suite_db_mgr.put_suite_params(self)
         self.suite_db_mgr.put_suite_template_vars(self.template_vars)
         self.suite_db_mgr.put_runtime_inheritance(self.config)
         self.configure_suite_environment()
@@ -446,8 +439,8 @@ def load_tasks_for_restart(self):
         """Load tasks for restart."""
         self.suite_db_mgr.pri_dao.select_suite_params(
             self._load_suite_params_2, self.options.checkpoint)
-        if self._cli_start_point_string:
-            self.start_point = self._cli_start_point_string
+        if self.cli_start_point_string:
+            self.start_point = self.cli_start_point_string
         self.suite_db_mgr.pri_dao.select_broadcast_states(
             self.task_events_mgr.broadcast_mgr.load_db_broadcast_states,
             self.options.checkpoint)
@@ -901,15 +894,7 @@ def command_reload_suite(self):
         self.configure_suite_environment()
         if self.options.genref or self.options.reftest:
             self.configure_reftest(recon=True)
-        self.suite_db_mgr.put_suite_params(
-            CYLC_VERSION,
-            self.task_job_mgr.task_remote_mgr.uuid_str,
-            self.run_mode,
-            str(cylc.flags.utc),
-            self.initial_point,
-            self.final_point,
-            self.pool.is_held,
-            self.config.cfg['cylc']['cycle point format'])
+        self.suite_db_mgr.put_suite_params(self)
         cylc.flags.iflag = True
 
     def set_suite_timer(self):
@@ -993,8 +978,8 @@ def load_suiterc(self, is_reload=False):
         self.config = SuiteConfig(
             self.suite, self.suiterc, self.template_vars,
             run_mode=self.run_mode,
-            cli_initial_point_string=self._cli_initial_point_string,
-            cli_start_point_string=self._cli_start_point_string,
+            cli_initial_point_string=self.cli_initial_point_string,
+            cli_start_point_string=self.cli_start_point_string,
             cli_final_point_string=self.options.final_point_string,
             is_reload=is_reload,
             mem_log_func=self.profiler.log_memory,
@@ -1044,17 +1029,17 @@ def load_suiterc(self, is_reload=False):
             self.run_mode = self.config.run_mode
 
     def _load_suite_params_1(self, _, row):
-        """Load previous initial/start cycle point.
+        """Load previous initial cycle point or (warm) start cycle point.
 
-        For restart, it may be missing from "suite.rc", but was specified as a
-        command line argument on cold/warm start.
+        For restart, these may be missing from "suite.rc", but was specified as
+        a command line argument on cold/warm start.
         """
         key, value = row
-        if key == 'initial_point':
-            self._cli_initial_point_string = value
+        if key == "initial_point":
+            self.cli_initial_point_string = value
             self.task_events_mgr.pflag = True
-        elif key == 'warm_point':
-            self._cli_start_point_string = value
+        elif key in ["start_point", "warm_point"]:
+            self.cli_start_point_string = value
             self.task_events_mgr.pflag = True
         elif key == 'uuid_str':
             self.task_job_mgr.task_remote_mgr.uuid_str = str(value)
@@ -1245,6 +1230,22 @@ def database_health_check(self):
             # Something has to be very wrong here, so stop the suite
             raise SchedulerError(str(exc))
 
+    def late_tasks_check(self):
+        """Report tasks that are late for their clock triggers."""
+        now = time()
+        for itask in self.pool.get_tasks():
+            if (not itask.is_late and itask.get_late_time() and
+                    itask.state.status in TASK_STATUSES_NEVER_ACTIVE and
+                    now > itask.get_late_time()):
+                msg = '%s (late-time=%s)' % (
+                    self.task_events_mgr.EVENT_LATE,
+                    time2str(itask.get_late_time()))
+                itask.is_late = True
+                LOG.warning(msg, itask=itask)
+                self.task_events_mgr.setup_event_handlers(
+                    itask, self.task_events_mgr.EVENT_LATE, msg)
+                self.suite_db_mgr.put_insert_task_late_flags(itask)
+
     def timeout_check(self):
         """Check suite and task timers."""
         self.check_suite_timer()
@@ -1373,8 +1374,9 @@ def run(self):
 
             # PROCESS ALL TASKS whenever something has changed that might
             # require renegotiation of dependencies, etc.
-            if self.process_tasks():
+            if self.should_process_tasks():
                 self.process_task_pool()
+            self.late_tasks_check()
 
             self.process_queued_task_messages()
             self.process_command_queue()
@@ -1481,22 +1483,13 @@ def check_suite_stalled(self):
             if self._get_events_conf(self.EVENT_TIMEOUT):
                 self.set_suite_timer()
 
-    def process_tasks(self):
+    def should_process_tasks(self):
         """Return True if waiting tasks are ready."""
         # do we need to do a pass through the main task processing loop?
         process = False
 
-        # External triggers must be matched now. If any are matched pflag
-        # is set to tell process_tasks() that task processing is required.
-        broadcast_mgr = self.task_events_mgr.broadcast_mgr
-        broadcast_mgr.add_ext_triggers(self.ext_trigger_queue)
-        for itask in self.pool.get_tasks():
-            if (itask.state.external_triggers and
-                    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
+            # This flag is turned on by commands that change task state
             process = True
             self.task_events_mgr.pflag = False  # reset
 
@@ -1505,10 +1498,18 @@ def process_tasks(self):
             process = True
             self.task_job_mgr.task_remote_mgr.ready = False  # reset
 
-        self.pool.set_expired_tasks()
-        if self.pool.waiting_tasks_ready():
-            process = True
-
+        broadcast_mgr = self.task_events_mgr.broadcast_mgr
+        broadcast_mgr.add_ext_triggers(self.ext_trigger_queue)
+        now = time()
+        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, now):
+                process = True
+            if itask.is_ready(now):
+                process = True
         if self.run_mode == 'simulation' and self.pool.sim_time_check(
                 self.message_queue):
             process = True

lib/cylc/suite_db_mgr.py

@@ -32,8 +32,10 @@
 from tempfile import mkstemp
 
 from cylc.broadcast_report import get_broadcast_change_iter
+import cylc.flags
 from cylc.rundb import CylcSuiteDAO
 from cylc.suite_logging import ERR, LOG
+from cylc.version import CYLC_VERSION
 from cylc.wallclock import get_current_time_string
 
 
@@ -251,44 +253,33 @@ def put_runtime_inheritance(self, config):
                 "namespace": namespace,
                 "inheritance": json.dumps(value)})
 
-    def put_suite_params(
-            self, cylc_version, uuid_str, run_mode, UTC_mode, initial_point,
-            final_point, is_held, cycle_point_format=None, warm_point=None):
-        """Put run mode, cylc version, UTC mode & initial & final cycle
-        points in runtime database.
+    def put_suite_params(self, schd):
+        """Put various suite parameters from schd in runtime database.
 
         This method queues the relevant insert statements.
 
         Arguments:
-            cylc_version (str): Cylc version of the running suite.
-            uuid_str (str): UUID of the suite that persists for restarts.
-            run_mode (str): "live", "dummy", "simulation", etc.
-            UTC_mode (boolean): Is UTC mode?
-            initial_point (str): Initial cycle point.
-            final_point (str): Final cycle point.
-            is_held (boolean): Is suite held?
-            cycle_point_format (str): Format string for cycle points.
-            warm_point (str): Warn start cycle point.
+            schd (cylc.scheduler.Scheduler): scheduler object.
         """
         self.db_inserts_map[self.TABLE_SUITE_PARAMS].extend([
-            {"key": "uuid_str", "value": uuid_str},
-            {"key": "run_mode", "value": run_mode},
-            {"key": "cylc_version", "value": cylc_version},
-            {"key": "UTC_mode", "value": UTC_mode},
-            {"key": "initial_point", "value": str(initial_point)},
-            {"key": "final_point", "value": str(final_point)},
+            {"key": "uuid_str",
+             "value": schd.task_job_mgr.task_remote_mgr.uuid_str},
+            {"key": "run_mode", "value": schd.run_mode},
+            {"key": "cylc_version", "value": CYLC_VERSION},
+            {"key": "UTC_mode", "value": cylc.flags.utc},
+            {"key": "initial_point", "value": str(schd.initial_point)},
+            {"key": "final_point", "value": str(schd.final_point)},
         ])
-        if cycle_point_format:
-            self.db_inserts_map[self.TABLE_SUITE_PARAMS].append(
-                {"key": "cycle_point_format", "value": str(cycle_point_format)}
-            )
-        if is_held:
-            self.db_inserts_map[self.TABLE_SUITE_PARAMS].append(
-                {"key": "is_held", "value": 1})
-        if warm_point:
-            self.db_inserts_map[self.TABLE_SUITE_PARAMS].append(
-                {"key": "warm_point", "value": str(warm_point)}
-            )
+        if schd.config.cfg['cylc']['cycle point format']:
+            self.db_inserts_map[self.TABLE_SUITE_PARAMS].append({
+                "key": "cycle_point_format",
+                "value": schd.config.cfg['cylc']['cycle point format']})
+        if schd.pool.is_held:
+            self.db_inserts_map[self.TABLE_SUITE_PARAMS].append({
+                "key": "is_held", "value": 1})
+        if schd.cli_start_point_string:
+            self.db_inserts_map[self.TABLE_SUITE_PARAMS].append({
+                "key": "start_point", "value": schd.cli_start_point_string})
 
     def put_suite_template_vars(self, template_vars):
         """Put template_vars in runtime database.
@@ -377,6 +368,12 @@ def put_insert_task_events(self, itask, args):
         """Put INSERT statement for task_events table."""
         self._put_insert_task_x(CylcSuiteDAO.TABLE_TASK_EVENTS, itask, args)
 
+    def put_insert_task_late_flags(self, itask):
+        """If itask is late, put INSERT statement to task_late_flags table."""
+        if itask.is_late:
+            self._put_insert_task_x(
+                CylcSuiteDAO.TABLE_TASK_LATE_FLAGS, itask, {"value": True})
+
     def put_insert_task_jobs(self, itask, args):
         """Put INSERT statement for task_jobs table."""
         self._put_insert_task_x(CylcSuiteDAO.TABLE_TASK_JOBS, itask, args)
@@ -413,11 +410,6 @@ def put_update_task_outputs(self, itask):
             CylcSuiteDAO.TABLE_TASK_OUTPUTS,
             itask, {"outputs": json.dumps(items)})
 
-    def put_update_task_states(self, itask, set_args):
-        """Put UPDATE statement for task_states table."""
-        self._put_update_task_x(
-            CylcSuiteDAO.TABLE_TASK_STATES, itask, set_args)
-
     def _put_update_task_x(self, table_name, itask, set_args):
         """Put UPDATE statement for a task_* table."""
         where_args = {