Merge branch 'master' into group-trigger

Author: hjoliver

changes.d/6574.fix.md

@@ -0,0 +1 @@
+Broadcast: Report any settings that are not compatible with the scheduler Cylc version.

cylc/flow/broadcast_mgr.py

@@ -37,7 +37,7 @@
     fail_if_platform_and_host_conflict,
     PlatformLookupError,
 )
-
+from cylc.flow.util import uniq
 
 if TYPE_CHECKING:
     from cylc.flow.id import Tokens
@@ -151,7 +151,7 @@ def clear_broadcast(
             LOG.error(get_broadcast_bad_options_report(bad_options))
         if modified_settings:
             self.data_store_mgr.delta_broadcast()
-        return modified_settings, bad_options
+        return uniq(modified_settings), bad_options
 
     def expire_broadcast(self, cutoff=None, **kwargs):
         """Clear all broadcasts targeting cycle points earlier than cutoff."""
@@ -284,18 +284,24 @@ def put_broadcast(
           bad_options is as described in the docstring for self.clear().
         """
         modified_settings = []
-        bad_point_strings = []
-        bad_namespaces = []
+        bad_settings = []
+        bad_point_strings = set()
+        bad_namespaces = set()
 
         with self.lock:
             for setting in settings or []:
                 # Coerce setting to cylc runtime object,
                 # i.e. str to  DurationFloat.
                 coerced_setting = deepcopy(setting)
-                BroadcastConfigValidator().validate(
-                    coerced_setting,
-                    SPEC['runtime']['__MANY__'],
-                )
+                try:
+                    BroadcastConfigValidator().validate(
+                        coerced_setting,
+                        SPEC['runtime']['__MANY__'],
+                    )
+                except Exception as exc:
+                    LOG.error(exc)
+                    bad_settings.append(setting)
+                    continue
 
                 # Skip and warn if a run mode is broadcast to a workflow
                 # running in simulation or dummy mode.
@@ -308,22 +314,24 @@ def put_broadcast(
                         f' running in {self.workflow_run_mode.value} mode'
                         ' will have no effect, and will not be actioned.'
                     )
+                    bad_settings.append(setting)
                     continue
 
                 for point_string in point_strings or []:
                     # Standardise the point and check its validity.
                     bad_point = False
                     try:
                         point_string = standardise_point_string(point_string)
+
                     except PointParsingError:
                         if point_string != '*':
-                            bad_point_strings.append(point_string)
+                            bad_point_strings.add(point_string)
                             bad_point = True
                     if not bad_point and point_string not in self.broadcasts:
                         self.broadcasts[point_string] = {}
                     for namespace in namespaces or []:
                         if namespace not in self.linearized_ancestors:
-                            bad_namespaces.append(namespace)
+                            bad_namespaces.add(namespace)
                         elif not bad_point:
                             # Check broadcast against config and against
                             # existing broadcasts:
@@ -340,6 +348,7 @@ def put_broadcast(
                                 namespace,
                                 coerced_setting,
                             ):
+                                bad_settings.append(setting)
                                 continue
 
                             if namespace not in self.broadcasts[point_string]:
@@ -362,13 +371,15 @@ def put_broadcast(
         LOG.info(get_broadcast_change_report(modified_settings))
 
         bad_options = {}
+        if bad_settings:
+            bad_options["settings"] = uniq(bad_settings)
         if bad_point_strings:
-            bad_options["point_strings"] = bad_point_strings
+            bad_options["point_strings"] = sorted(bad_point_strings)
         if bad_namespaces:
-            bad_options["namespaces"] = bad_namespaces
+            bad_options["namespaces"] = sorted(bad_namespaces)
         if modified_settings:
             self.data_store_mgr.delta_broadcast()
-        return modified_settings, bad_options
+        return uniq(modified_settings), bad_options
 
     @staticmethod
     def bc_mixes_old_and_new_platform_settings(

cylc/flow/broadcast_report.py

@@ -26,6 +26,13 @@
 CHANGE_TITLE_CANCEL = "Broadcast cancelled:"
 CHANGE_TITLE_SET = "Broadcast set:"
 
+CLI_OPT_MAP = {
+    # broadcast field: cli option
+    'point_strings': 'point',
+    'namespaces': 'namespace',
+    'settings': 'set',
+}
+
 
 def get_broadcast_bad_options_report(bad_options, is_set=False):
     """Return a string to report bad options for broadcast cancel/clear."""
@@ -48,7 +55,11 @@ def get_broadcast_bad_options_report(bad_options, is_set=False):
                         value_str += val
             else:
                 value_str = value
-            msg += BAD_OPTIONS_FMT % (key, value_str)
+            if isinstance(value, dict):
+                value_str = ', '.join(
+                    f'{key}={val}' for key, val in value.items()
+                )
+            msg += BAD_OPTIONS_FMT % (CLI_OPT_MAP.get(key, key), value_str)
     return msg
 
 

cylc/flow/data_messages_pb2.py

@@ -304,7 +304,7 @@ class PbTrigger(_message.Message):
     def __init__(self, id: _Optional[str] = ..., label: _Optional[str] = ..., message: _Optional[str] = ..., satisfied: bool = ..., time: _Optional[float] = ...) -> None: ...
 
 class PbTaskProxy(_message.Message):
-    __slots__ = ("stamp", "id", "task", "state", "cycle_point", "depth", "job_submits", "outputs", "namespace", "prerequisites", "jobs", "first_parent", "name", "is_held", "edges", "ancestors", "flow_nums", "external_triggers", "xtriggers", "is_queued", "is_runahead", "flow_wait", "runtime", "graph_depth")
+    __slots__ = ("stamp", "id", "task", "state", "cycle_point", "depth", "job_submits", "outputs", "namespace", "prerequisites", "jobs", "first_parent", "name", "is_held", "edges", "ancestors", "flow_nums", "external_triggers", "xtriggers", "is_queued", "is_runahead", "flow_wait", "runtime", "graph_depth", "is_retry", "is_wallclock", "is_xtriggered")
     class OutputsEntry(_message.Message):
         __slots__ = ("key", "value")
         KEY_FIELD_NUMBER: _ClassVar[int]
@@ -350,6 +350,9 @@ class PbTaskProxy(_message.Message):
     FLOW_WAIT_FIELD_NUMBER: _ClassVar[int]
     RUNTIME_FIELD_NUMBER: _ClassVar[int]
     GRAPH_DEPTH_FIELD_NUMBER: _ClassVar[int]
+    IS_RETRY_FIELD_NUMBER: _ClassVar[int]
+    IS_WALLCLOCK_FIELD_NUMBER: _ClassVar[int]
+    IS_XTRIGGERED_FIELD_NUMBER: _ClassVar[int]
     stamp: str
     id: str
     task: str
@@ -374,7 +377,10 @@ class PbTaskProxy(_message.Message):
     flow_wait: bool
     runtime: PbRuntime
     graph_depth: int
-    def __init__(self, stamp: _Optional[str] = ..., id: _Optional[str] = ..., task: _Optional[str] = ..., state: _Optional[str] = ..., cycle_point: _Optional[str] = ..., depth: _Optional[int] = ..., job_submits: _Optional[int] = ..., outputs: _Optional[_Mapping[str, PbOutput]] = ..., namespace: _Optional[_Iterable[str]] = ..., prerequisites: _Optional[_Iterable[_Union[PbPrerequisite, _Mapping]]] = ..., jobs: _Optional[_Iterable[str]] = ..., first_parent: _Optional[str] = ..., name: _Optional[str] = ..., is_held: bool = ..., edges: _Optional[_Iterable[str]] = ..., ancestors: _Optional[_Iterable[str]] = ..., flow_nums: _Optional[str] = ..., external_triggers: _Optional[_Mapping[str, PbTrigger]] = ..., xtriggers: _Optional[_Mapping[str, PbTrigger]] = ..., is_queued: bool = ..., is_runahead: bool = ..., flow_wait: bool = ..., runtime: _Optional[_Union[PbRuntime, _Mapping]] = ..., graph_depth: _Optional[int] = ...) -> None: ...
+    is_retry: bool
+    is_wallclock: bool
+    is_xtriggered: bool
+    def __init__(self, stamp: _Optional[str] = ..., id: _Optional[str] = ..., task: _Optional[str] = ..., state: _Optional[str] = ..., cycle_point: _Optional[str] = ..., depth: _Optional[int] = ..., job_submits: _Optional[int] = ..., outputs: _Optional[_Mapping[str, PbOutput]] = ..., namespace: _Optional[_Iterable[str]] = ..., prerequisites: _Optional[_Iterable[_Union[PbPrerequisite, _Mapping]]] = ..., jobs: _Optional[_Iterable[str]] = ..., first_parent: _Optional[str] = ..., name: _Optional[str] = ..., is_held: bool = ..., edges: _Optional[_Iterable[str]] = ..., ancestors: _Optional[_Iterable[str]] = ..., flow_nums: _Optional[str] = ..., external_triggers: _Optional[_Mapping[str, PbTrigger]] = ..., xtriggers: _Optional[_Mapping[str, PbTrigger]] = ..., is_queued: bool = ..., is_runahead: bool = ..., flow_wait: bool = ..., runtime: _Optional[_Union[PbRuntime, _Mapping]] = ..., graph_depth: _Optional[int] = ..., is_retry: bool = ..., is_wallclock: bool = ..., is_xtriggered: bool = ...) -> None: ...
 
 class PbFamily(_message.Message):
     __slots__ = ("stamp", "id", "name", "meta", "depth", "proxies", "parents", "child_tasks", "child_families", "first_parent", "runtime", "descendants")
@@ -405,7 +411,7 @@ class PbFamily(_message.Message):
     def __init__(self, stamp: _Optional[str] = ..., id: _Optional[str] = ..., name: _Optional[str] = ..., meta: _Optional[_Union[PbMeta, _Mapping]] = ..., depth: _Optional[int] = ..., proxies: _Optional[_Iterable[str]] = ..., parents: _Optional[_Iterable[str]] = ..., child_tasks: _Optional[_Iterable[str]] = ..., child_families: _Optional[_Iterable[str]] = ..., first_parent: _Optional[str] = ..., runtime: _Optional[_Union[PbRuntime, _Mapping]] = ..., descendants: _Optional[_Iterable[str]] = ...) -> None: ...
 
 class PbFamilyProxy(_message.Message):
-    __slots__ = ("stamp", "id", "cycle_point", "name", "family", "state", "depth", "first_parent", "child_tasks", "child_families", "is_held", "ancestors", "states", "state_totals", "is_held_total", "is_queued", "is_queued_total", "is_runahead", "is_runahead_total", "runtime", "graph_depth")
+    __slots__ = ("stamp", "id", "cycle_point", "name", "family", "state", "depth", "first_parent", "child_tasks", "child_families", "is_held", "ancestors", "states", "state_totals", "is_held_total", "is_queued", "is_queued_total", "is_runahead", "is_runahead_total", "runtime", "graph_depth", "is_retry", "is_wallclock", "is_xtriggered")
     class StateTotalsEntry(_message.Message):
         __slots__ = ("key", "value")
         KEY_FIELD_NUMBER: _ClassVar[int]
@@ -434,6 +440,9 @@ class PbFamilyProxy(_message.Message):
     IS_RUNAHEAD_TOTAL_FIELD_NUMBER: _ClassVar[int]
     RUNTIME_FIELD_NUMBER: _ClassVar[int]
     GRAPH_DEPTH_FIELD_NUMBER: _ClassVar[int]
+    IS_RETRY_FIELD_NUMBER: _ClassVar[int]
+    IS_WALLCLOCK_FIELD_NUMBER: _ClassVar[int]
+    IS_XTRIGGERED_FIELD_NUMBER: _ClassVar[int]
     stamp: str
     id: str
     cycle_point: str
@@ -455,7 +464,10 @@ class PbFamilyProxy(_message.Message):
     is_runahead_total: int
     runtime: PbRuntime
     graph_depth: int
-    def __init__(self, stamp: _Optional[str] = ..., id: _Optional[str] = ..., cycle_point: _Optional[str] = ..., name: _Optional[str] = ..., family: _Optional[str] = ..., state: _Optional[str] = ..., depth: _Optional[int] = ..., first_parent: _Optional[str] = ..., child_tasks: _Optional[_Iterable[str]] = ..., child_families: _Optional[_Iterable[str]] = ..., is_held: bool = ..., ancestors: _Optional[_Iterable[str]] = ..., states: _Optional[_Iterable[str]] = ..., state_totals: _Optional[_Mapping[str, int]] = ..., is_held_total: _Optional[int] = ..., is_queued: bool = ..., is_queued_total: _Optional[int] = ..., is_runahead: bool = ..., is_runahead_total: _Optional[int] = ..., runtime: _Optional[_Union[PbRuntime, _Mapping]] = ..., graph_depth: _Optional[int] = ...) -> None: ...
+    is_retry: bool
+    is_wallclock: bool
+    is_xtriggered: bool
+    def __init__(self, stamp: _Optional[str] = ..., id: _Optional[str] = ..., cycle_point: _Optional[str] = ..., name: _Optional[str] = ..., family: _Optional[str] = ..., state: _Optional[str] = ..., depth: _Optional[int] = ..., first_parent: _Optional[str] = ..., child_tasks: _Optional[_Iterable[str]] = ..., child_families: _Optional[_Iterable[str]] = ..., is_held: bool = ..., ancestors: _Optional[_Iterable[str]] = ..., states: _Optional[_Iterable[str]] = ..., state_totals: _Optional[_Mapping[str, int]] = ..., is_held_total: _Optional[int] = ..., is_queued: bool = ..., is_queued_total: _Optional[int] = ..., is_runahead: bool = ..., is_runahead_total: _Optional[int] = ..., runtime: _Optional[_Union[PbRuntime, _Mapping]] = ..., graph_depth: _Optional[int] = ..., is_retry: bool = ..., is_wallclock: bool = ..., is_xtriggered: bool = ...) -> None: ...
 
 class PbEdge(_message.Message):
     __slots__ = ("stamp", "id", "source", "target", "suicide", "cond")

cylc/flow/data_messages_pb2.pyi

@@ -1526,6 +1526,8 @@ def _process_internal_task_proxy(
             xtrig.satisfied = satisfied
             self.xtrigger_tasks.setdefault(sig, set()).add((tproxy.id, label))
 
+        self._set_task_xtrigger_modifiers(tproxy)
+
         if tproxy.state in self.latest_state_tasks:
             tp_ref = itask.identity
             tp_queue = self.latest_state_tasks[tproxy.state]
@@ -2014,6 +2016,15 @@ def update_family_proxies(self):
         if self.updated_state_families:
             self.state_update_follow_on = True
 
+    @staticmethod
+    def from_delta_or_node(tp_delta, tp_node, label):
+        """Get an item from task proxy delta if available, falling back to
+        node otherwise."""
+        this = tp_delta
+        if this is None or not this.HasField(label):
+            this = tp_node
+        return getattr(this, label) or None
+
     def _family_ascent_point_update(self, fp_id):
         """Updates the given family and children recursively.
 
@@ -2052,6 +2063,9 @@ def _family_ascent_point_update(self, fp_id):
             is_held_total = 0
             is_queued_total = 0
             is_runahead_total = 0
+            is_retry = False
+            is_wallclock = False
+            is_xtriggered = False
             graph_depth = self.n_edge_distance
             for child_id in fam_node.child_families:
                 child_node = fp_updated.get(child_id, fp_data.get(child_id))
@@ -2071,31 +2085,28 @@ def _family_ascent_point_update(self, fp_id):
                 tp_delta = tp_updated.get(tp_id)
                 tp_node = tp_added.get(tp_id, tp_data.get(tp_id))
 
-                tp_state = tp_delta
-                if tp_state is None or not tp_state.HasField('state'):
-                    tp_state = tp_node
-                if tp_state.state:
-                    task_states.append(tp_state.state)
+                tp_state = self.from_delta_or_node(tp_delta, tp_node, 'state')
+                if tp_state:
+                    task_states.append(tp_state)
 
-                tp_held = tp_delta
-                if tp_held is None or not tp_held.HasField('is_held'):
-                    tp_held = tp_node
-                if tp_held.is_held:
+                if self.from_delta_or_node(tp_delta, tp_node, 'is_held'):
                     is_held_total += 1
 
-                tp_queued = tp_delta
-                if tp_queued is None or not tp_queued.HasField('is_queued'):
-                    tp_queued = tp_node
-                if tp_queued.is_queued:
+                if self.from_delta_or_node(tp_delta, tp_node, 'is_queued'):
                     is_queued_total += 1
 
-                tp_runahead = tp_delta
-                if (tp_runahead is None
-                        or not tp_runahead.HasField('is_runahead')):
-                    tp_runahead = tp_node
-                if tp_runahead.is_runahead:
+                if self.from_delta_or_node(tp_delta, tp_node, 'is_runahead'):
                     is_runahead_total += 1
 
+                if self.from_delta_or_node(tp_delta, tp_node, 'is_retry'):
+                    is_retry = True
+
+                if self.from_delta_or_node(tp_delta, tp_node, 'is_wallclock'):
+                    is_wallclock = True
+
+                if self.from_delta_or_node(tp_delta, tp_node, 'is_xtriggered'):
+                    is_xtriggered = True
+
                 tp_depth = tp_delta
                 if tp_depth is None or not tp_depth.HasField('graph_depth'):
                     tp_depth = tp_node
@@ -2114,7 +2125,10 @@ def _family_ascent_point_update(self, fp_id):
                 is_queued_total=is_queued_total,
                 is_runahead=(is_runahead_total > 0),
                 is_runahead_total=is_runahead_total,
-                graph_depth=graph_depth
+                is_retry=is_retry,
+                is_wallclock=is_wallclock,
+                is_xtriggered=is_xtriggered,
+                graph_depth=graph_depth,
             )
             fp_delta.states[:] = state_counter.keys()
             # Use all states to clean up pruned counts
@@ -2522,19 +2536,72 @@ def delta_task_ext_trigger(
         ext_trigger.time = update_time
         self.updates_pending = True
 
+    @staticmethod
+    def _set_task_xtrigger_modifiers(node_or_delta):
+        # update the xtrigger task modifiers
+        node_or_delta.is_retry = False
+        node_or_delta.is_wallclock = False
+        node_or_delta.is_xtriggered = False
+        for xtrigger in node_or_delta.xtriggers.values():
+            if xtrigger.satisfied:
+                continue
+            if (
+                xtrigger.label.startswith('_cylc_retry')
+                or xtrigger.label.startswith('_cylc_submit_retry')
+            ):
+                node_or_delta.is_retry = True
+            elif xtrigger.id.startswith('wall_clock'):
+                node_or_delta.is_wallclock = True
+            else:
+                node_or_delta.is_xtriggered = True
+
     def _delta_xtrigger(
-        self, tp_id: str, label: str, sig: str, satisfied: bool,
-        t_update: float
+        self,
+        tp_id: str,
+        label: str,
+        sig: str,
+        satisfied: bool,
+        update_time: float,
     ) -> None:
         """Helper for the two xtrigger delta methods."""
+        # fetch the task from the store
+        tp_id, tproxy = self.store_node_fetcher(Tokens(tp_id))
+        if not tproxy:
+            return
+
+        # create or fetch the updated delta
         tp_delta = self.updated[TASK_PROXIES].setdefault(
-            tp_id, PbTaskProxy(id=tp_id))
-        tp_delta.stamp = f'{tp_id}@{t_update}'
+            tp_id, PbTaskProxy(id=tp_id)
+        )
+        tp_delta.stamp = f'{tp_id}@{update_time}'
+
+        # populate all xtriggers on the delta if not already present
+        if not tp_delta.xtriggers:
+            # NOTE: if one xtrigger changes, we must include all in the
+            # delta, see https://github.com/cylc/cylc-flow/issues/6307
+            for _sig, xtrigger in tproxy.xtriggers.items():
+                if _sig == sig:
+                    # don't copy the xtrigger we are changing
+                    continue
+                _xtrigger = tp_delta.xtriggers[_sig]
+                _xtrigger.id = xtrigger.id
+                _xtrigger.label = xtrigger.label
+                _xtrigger.satisfied = xtrigger.satisfied
+                _xtrigger.time = xtrigger.time
+
+        # modify the xtrigger that has changed
         xtrigger = tp_delta.xtriggers[sig]
         xtrigger.id = sig
         xtrigger.label = label
         xtrigger.satisfied = satisfied
-        xtrigger.time = t_update
+        xtrigger.time = update_time
+
+        # update the xtrigger task modifiers
+        self._set_task_xtrigger_modifiers(tp_delta)
+
+        # ensure family modifier counts are updated
+        self.state_update_families.add(tproxy.first_parent)
+
         self.updates_pending = True
 
     def delta_xtrigger(self, sig: str, succeeded: bool) -> None: