Persist satisfied xtriggers to the DB.

lib/cylc/rundb.py

@@ -191,6 +191,7 @@ class CylcSuiteDAO(object):
     TABLE_TASK_POOL_CHECKPOINTS = "task_pool_checkpoints"
     TABLE_TASK_STATES = "task_states"
     TABLE_TASK_TIMEOUT_TIMERS = "task_timeout_timers"
+    TABLE_XTRIGGERS = "xtriggers"
 
     TABLES_ATTRS = {
         TABLE_BROADCAST_EVENTS: [
@@ -283,6 +284,10 @@ class CylcSuiteDAO(object):
             ["status"],
             ["hold_swap"],
         ],
+        TABLE_XTRIGGERS: [
+            ["cycle"],
+            ["signature", {"is_primary_key": True}],
+        ],
         TABLE_TASK_POOL_CHECKPOINTS: [
             ["id", {"datatype": "INTEGER", "is_primary_key": True}],
             ["cycle", {"is_primary_key": True}],
@@ -628,6 +633,11 @@ def select_submit_nums_for_insert(self, task_ids):
             ret[(name, cycle)] = submit_num
         return ret
 
+    def select_xtriggers_for_restart(self, callback):
+        stmt = r"SELECT cycle,signature FROM %s" % self.TABLE_XTRIGGERS
+        for row_idx, row in enumerate(self.connect().execute(stmt, [])):
+            callback(row_idx, list(row))
+
     def select_task_pool(self, callback, id_key=None):
         """Select from task_pool or task_pool_checkpoints.
 

lib/cylc/scheduler.py

@@ -449,6 +449,9 @@ def load_tasks_for_restart(self):
             self.pool.load_db_task_pool_for_restart, self.options.checkpoint)
         self.suite_db_mgr.pri_dao.select_task_action_timers(
             self.pool.load_db_task_action_timers)
+        self.suite_db_mgr.pri_dao.select_xtriggers_for_restart(
+            self.xtrigger_mgr.load_xtrigger_for_restart)
+
         # Re-initialise run directory for user@host for each submitted and
         # running tasks.
         # Note: tasks should all be in the runahead pool at this point.
@@ -1151,6 +1154,7 @@ def process_task_pool(self):
         min_point = self.pool.get_min_point()
         self.broadcast_mgr.expire_broadcast(min_point)
         self.xtrigger_mgr.housekeep(min_point)
+        self.suite_db_mgr.put_xtriggers(self.xtrigger_mgr.xtrig_satisfied)
         if cylc.flags.debug:
             LOG.debug("END TASK PROCESSING (took %s seconds)" %
                       (time() - time0))

lib/cylc/suite_db_mgr.py

@@ -51,6 +51,7 @@ class SuiteDatabaseManager(object):
     TABLE_TASK_OUTPUTS = CylcSuiteDAO.TABLE_TASK_OUTPUTS
     TABLE_TASK_STATES = CylcSuiteDAO.TABLE_TASK_STATES
     TABLE_TASK_TIMEOUT_TIMERS = CylcSuiteDAO.TABLE_TASK_TIMEOUT_TIMERS
+    TABLE_XTRIGGERS = CylcSuiteDAO.TABLE_XTRIGGERS
 
     def __init__(self, pri_d=None, pub_d=None):
         self.pri_path = None
@@ -68,7 +69,8 @@ def __init__(self, pri_d=None, pub_d=None):
             self.TABLE_TASK_POOL: [],
             self.TABLE_TASK_ACTION_TIMERS: [],
             self.TABLE_TASK_OUTPUTS: [],
-            self.TABLE_TASK_TIMEOUT_TIMERS: []}
+            self.TABLE_TASK_TIMEOUT_TIMERS: [],
+            self.TABLE_XTRIGGERS: []}
         self.db_inserts_map = {
             self.TABLE_BROADCAST_EVENTS: [],
             self.TABLE_BROADCAST_STATES: [],
@@ -79,7 +81,8 @@ def __init__(self, pri_d=None, pub_d=None):
             self.TABLE_TASK_POOL: [],
             self.TABLE_TASK_ACTION_TIMERS: [],
             self.TABLE_TASK_OUTPUTS: [],
-            self.TABLE_TASK_TIMEOUT_TIMERS: []}
+            self.TABLE_TASK_TIMEOUT_TIMERS: [],
+            self.TABLE_XTRIGGERS: []}
         self.db_updates_map = {}
 
     def checkpoint(self, name):
@@ -285,6 +288,13 @@ def put_task_event_timers(self, task_events_mgr):
                     "delay": timer.delay,
                     "timeout": timer.timeout})
 
+    def put_xtriggers(self, xtrig_satisfied):
+        self.db_deletes_map[self.TABLE_XTRIGGERS].append({})
+        for cycle, signature in xtrig_satisfied:
+            self.db_inserts_map[self.TABLE_XTRIGGERS].append({
+                "cycle": str(cycle),
+                "signature": signature})
+
     def put_task_pool(self, pool):
         """Put statements to update the task_pool table in runtime database.
 

lib/cylc/xtrigger_mgr.py

@@ -19,8 +19,8 @@
 import re
 from time import time
 from copy import deepcopy
-from cylc.suite_logging import LOG
-
+from cylc.suite_logging import LOG, OUT
+from cylc.cycling.loader import get_point
 
 # Templates for string replacement in funtion call arg values.
 TMPL_POINT = 'cycle_point'
@@ -101,6 +101,12 @@ def add_trig(self, key, val):
                     raise ValueError(
                         "Illegal template in xtrigger %s: %s" % (key, match))
 
+    def load_xtrigger_for_restart(self, row_idx, row):
+        if row_idx == 0:
+            OUT.info("LOADING satisfied xtriggers")
+        cycle_str, signature = row 
+        self.xtrig_satisfied.append((get_point(cycle_str), signature))
+
     def housekeep(self, point):
         """Delete satisfied xtriggers and xclocks older than point."""
         rem = []

tests/database/00-simple/schema.out

@@ -14,3 +14,4 @@ CREATE TABLE task_pool(cycle TEXT, name TEXT, spawned INTEGER, status TEXT, hold
 CREATE TABLE task_pool_checkpoints(id INTEGER, cycle TEXT, name TEXT, spawned INTEGER, status TEXT, hold_swap TEXT, PRIMARY KEY(id, cycle, name));
 CREATE TABLE task_states(name TEXT, cycle TEXT, time_created TEXT, time_updated TEXT, submit_num INTEGER, status TEXT, PRIMARY KEY(name, cycle));
 CREATE TABLE task_timeout_timers(cycle TEXT, name TEXT, timeout REAL, PRIMARY KEY(cycle, name));
+CREATE TABLE xtriggers(cycle TEXT, signature TEXT, PRIMARY KEY(signature));