Task proxy spawn on demand.

CHANGES.md

@@ -33,6 +33,13 @@ namespace, under `cylc.jinja.filters`.
 
 Cylc Review was also removed in this version.
 
+-------------------------------------------------------------------------------
+<<<<<<< HEAD
+## __cylc-8.0a3 (2020-Q3?)__
+
+[#3515](https://github.com/cylc/cylc-flow/pull/3515) - spawn-on-demand: a more
+efficient way of the evolving the workflow via the graph.
+
 -------------------------------------------------------------------------------
 ## __cylc-8.0a2 (2020-07-03)__
 

cylc/flow/cfgspec/globalcfg.py

@@ -126,7 +126,8 @@
         with Conf('main loop', desc='''
             Configuration of the Cylc Scheduler's main loop.
         '''):
-            Conf('plugins', VDR.V_STRING_LIST, ['health check'], desc='''
+            Conf('plugins', VDR.V_STRING_LIST,
+                 ['health check', 'prune flow labels'], desc='''
                 Configure the default main loop plugins to use when
                 starting up new suites.
             ''')
@@ -146,6 +147,13 @@
                     The interval with which this plugin is run.
                 ''')
 
+            with Conf('prune flow labels', meta=MainLoopPlugin, desc='''
+                Prune redundant flow labels.
+            '''):
+                Conf('interval', VDR.V_INTERVAL, DurationFloat(600), desc='''
+                    The interval with which this plugin is run.
+                ''')
+
     # suite
     with Conf('suite logging', desc='''
         The suite event log, held under the suite run directory, is maintained

cylc/flow/cfgspec/suite.py

@@ -269,24 +269,6 @@
             consecutive cycle points to be active at any time, adjusted up if
             necessary for any future triggering.
         ''')
-        Conf('spawn to max active cycle points', VDR.V_BOOLEAN, desc='''
-            Allows tasks to spawn out to
-            :cylc:conf:`[..]max active cycle points`,
-            removing restriction that a task
-            has to have submitted before its successor can be spawned.
-
-            .. warning::
-               This should be used with care given the potential impact of
-               additional task proxies in terms of memory and cpu for the
-               cylc server program. Also, use of the setting may highlight
-               any issues with suite design relying on the default behaviour
-               where downstream tasks would otherwise be waiting on ones
-               upstream submitting and the suite would have stalled e.g. a
-               housekeeping task at a later cycle deleting an earlier cycle's
-               data before that cycle has had chance to run where previously
-               the task would not have been spawned until its predecessor had
-               been submitted.
-        ''')
 
         with Conf('queues', desc='''
             Configuration of internal queues, by which the number of

cylc/flow/config.py

@@ -1553,7 +1553,7 @@ def generate_taskdefs(self, orig_expr, left_nodes, right, seq):
                 # if right is None, lefts are lone nodes
                 # for which we still define the taskdefs
                 continue
-            name, offset_is_from_icp, _, offset, _ = (
+            name, offset, _, offset_is_from_icp, _, _ = (
                 GraphNodeParser.get_inst().parse(node))
 
             if name not in self.cfg['runtime']:
@@ -1620,30 +1620,9 @@ def generate_triggers(self, lexpression, left_nodes, right, seq,
                 xtrig_labels.add(left[1:])
                 continue
             # (GraphParseError checked above)
-            name, offset_is_from_icp, offset_is_irregular, offset, output = (
+            (name, offset, output, offset_is_from_icp,
+             offset_is_irregular, offset_is_absolute) = (
                 GraphNodeParser.get_inst().parse(left))
-            ltaskdef = self.taskdefs[name]
-
-            # Determine intercycle offsets.
-            abs_cycle_point = None
-            cycle_point_offset = None
-            if offset_is_from_icp:
-                first_point = get_point_relative(offset, self.initial_point)
-                last_point = seq.get_stop_point()
-                abs_cycle_point = first_point
-                if last_point is None:
-                    # This dependency persists for the whole suite run.
-                    ltaskdef.intercycle_offsets.add((None, seq))
-                else:
-                    ltaskdef.intercycle_offsets.add(
-                        (str(-(last_point - first_point)), seq))
-            elif offset:
-                if offset_is_irregular:
-                    offset_tuple = (offset, seq)
-                else:
-                    offset_tuple = (offset, None)
-                ltaskdef.intercycle_offsets.add(offset_tuple)
-                cycle_point_offset = offset
 
             # Qualifier.
             outputs = self.cfg['runtime'][name]['outputs']
@@ -1658,7 +1637,9 @@ def generate_triggers(self, lexpression, left_nodes, right, seq,
                 qualifier = TASK_OUTPUT_SUCCEEDED
 
             # Generate TaskTrigger if not already done.
-            key = (name, abs_cycle_point, cycle_point_offset, qualifier)
+            key = (name, offset, qualifier,
+                   offset_is_irregular, offset_is_absolute,
+                   offset_is_from_icp, self.initial_point)
             try:
                 task_trigger = task_triggers[key]
             except KeyError:
@@ -1667,6 +1648,11 @@ def generate_triggers(self, lexpression, left_nodes, right, seq,
 
             triggers[left] = task_trigger
 
+            # (name is left name)
+            self.taskdefs[name].add_graph_child(task_trigger, right, seq)
+            # graph_parents not currently used but might be needed soon:
+            # self.taskdefs[right].add_graph_parent(task_trigger, name, seq)
+
         # Walk down "expr_list" depth first, and replace any items matching a
         # key in "triggers" ("left" values) with the trigger.
         stack = [expr_list]
@@ -1849,7 +1835,7 @@ def get_graph_raw(self, start_point_string, stop_point_string,
                         offset_is_from_icp = False
                         offset = None
                     else:
-                        name, offset_is_from_icp, _, offset, _ = (
+                        name, offset, _, offset_is_from_icp, _, _ = (
                             GraphNodeParser.get_inst().parse(left))
                     if offset:
                         if offset_is_from_icp:
@@ -1963,7 +1949,7 @@ def get_graph_edges(self, start_point, stop_point):
                         offset_is_from_icp = False
                         offset = None
                     else:
-                        name, offset_is_from_icp, _, offset, _ = (
+                        name, offset, _, offset_is_from_icp, _, _ = (
                             GraphNodeParser.get_inst().parse(left))
                     if offset:
                         if offset_is_from_icp:
@@ -2203,8 +2189,7 @@ def _get_taskdef(self, name):
 
         # Get the taskdef object for generating the task proxy class
         taskd = TaskDef(
-            name, rtcfg, self.run_mode(), self.start_point,
-            self.cfg['scheduling']['spawn to max active cycle points'])
+            name, rtcfg, self.run_mode(), self.start_point)
 
         # TODO - put all taskd.foo items in a single config dict
 

cylc/flow/data_messages.proto

@@ -162,21 +162,20 @@ message PbTaskProxy {
     optional string task = 3;
     optional string state = 4;
     optional string cycle_point = 5;
-    optional bool spawned = 6;
-    optional int32 depth = 7;
-    optional int32 job_submits = 8;
-    optional string latest_message = 9;
-    repeated string outputs = 10;
-    repeated string broadcasts = 11;
-    repeated string namespace = 12;
-    repeated PbPrerequisite prerequisites = 13;
-    repeated string jobs = 14;
-    repeated string parents = 15;
-    optional string first_parent = 16;
-    optional string name = 17; /* filter item */
-    optional bool is_held = 18;
-    repeated string edges = 19;
-    repeated string ancestors = 20;
+    optional int32 depth = 6;
+    optional int32 job_submits = 7;
+    optional string latest_message = 8;
+    repeated string outputs = 9;
+    repeated string broadcasts = 10;
+    repeated string namespace = 11;
+    repeated PbPrerequisite prerequisites = 12;
+    repeated string jobs = 13;
+    repeated string parents = 14;
+    optional string first_parent = 15;
+    optional string name = 16; /* filter item */
+    optional bool is_held = 17;
+    repeated string edges = 18;
+    repeated string ancestors = 19;
 }
 
 message PbFamily {