Fix duplicate task triggers

Author: MetRonnie

changes.d/6625.fix.md

@@ -0,0 +1 @@
+Efficiency improvement: avoid storing duplicate information on graph triggers.

cylc/flow/task_trigger.py

@@ -14,7 +14,11 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-from typing import TYPE_CHECKING, Tuple
+from typing import (
+    TYPE_CHECKING,
+    Optional,
+    Tuple,
+)
 
 from cylc.flow.cycling.loader import (
     get_interval,
@@ -50,9 +54,16 @@ class TaskTrigger:
                  'offset_is_irregular', 'offset_is_absolute',
                  'offset_is_from_icp', 'initial_point']
 
-    def __init__(self, task_name, cycle_point_offset, output,
-                 offset_is_irregular=False, offset_is_absolute=False,
-                 offset_is_from_icp=False, initial_point=None):
+    def __init__(
+        self,
+        task_name: str,
+        cycle_point_offset: str,
+        output: str,
+        offset_is_irregular: bool = False,
+        offset_is_absolute: bool = False,
+        offset_is_from_icp: bool = False,
+        initial_point: 'Optional[PointBase]' = None,
+    ):
         self.task_name = task_name
         self.cycle_point_offset = cycle_point_offset
         self.output = output
@@ -147,7 +158,24 @@ def __str__(self):
         else:
             return '%s:%s' % (self.task_name, self.output)
 
-    __repr__ = __str__
+    def __repr__(self) -> str:
+        return f"<{type(self).__name__} {self}>"
+
+    def __hash__(self) -> int:
+        return hash((
+            self.task_name,
+            self.cycle_point_offset,
+            self.output,
+            self.offset_is_irregular,
+            self.offset_is_from_icp,
+            self.offset_is_absolute,
+            self.initial_point,
+        ))
+
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, TaskTrigger):
+            return NotImplemented
+        return hash(self) == hash(other)
 
     @staticmethod
     def standardise_name(name):

cylc/flow/taskdef.py

@@ -183,7 +183,7 @@ def __init__(self, name, rtcfg, start_point, initial_point):
         self.dependencies: Dict[SequenceBase, List[Dependency]] = {}
         self.outputs = {}  # {output: (message, is_required)}
         self.graph_children: Dict[
-            SequenceBase, Dict[str, List[Tuple[str, TaskTrigger]]]
+            SequenceBase, Dict[str, Set[Tuple[str, TaskTrigger]]]
         ] = {}
         self.graph_parents: Dict[
             SequenceBase, Set[Tuple[str, TaskTrigger]]
@@ -250,8 +250,8 @@ def add_graph_child(
         self.graph_children.setdefault(
             sequence, {}
         ).setdefault(
-            trigger.output, []
-        ).append((taskname, trigger))
+            trigger.output, set()
+        ).add((taskname, trigger))
 
     def add_graph_parent(
         self, trigger: 'TaskTrigger', parent: str, sequence: 'SequenceBase'