fix: reduce number of edge cases where lazy variadic components wait for inputs that can't arrive anymore (#8907)

* wip

* fix: running order with lazy variadic components

* fix: tests

* format

* comment

* fix: alternative approach to fixing running order

* unused imports

* revert fix

* remove unneeded return

* remove data based approach to tie breaking

* release note

* trailing spaces

* newline eof

* unused import

* add more explanations to release note

Author: mathislucka

haystack/core/pipeline/async_pipeline.py

@@ -144,6 +144,7 @@ async def process_results():
         ordered_names = sorted(self.graph.nodes.keys())
         cached_receivers = {n: self._find_receivers_from(n) for n in ordered_names}
         component_visits = {component_name: 0 for component_name in ordered_names}
+        cached_topological_sort = None
 
         # We fill the queue once and raise if all components are BLOCKED
         self.validate_pipeline(self._fill_queue(ordered_names, inputs_state, component_visits))
@@ -385,6 +386,15 @@ async def _wait_for_all_tasks_to_complete() -> AsyncIterator[Dict[str, Any]]:
 
                 # We only schedule components with priority DEFER or DEFER_LAST when no other tasks are running
                 elif priority in (ComponentPriority.DEFER, ComponentPriority.DEFER_LAST) and not running_tasks:
+                    if len(priority_queue) > 0:
+                        component_name, topological_sort = self._tiebreak_waiting_components(
+                            component_name=component_name,
+                            priority=priority,
+                            priority_queue=priority_queue,
+                            topological_sort=cached_topological_sort,
+                        )
+                        cached_topological_sort = topological_sort
+
                     await _schedule_task(component_name)
 
                 # To make progress, we wait for one task to complete before re-starting the loop

haystack/core/pipeline/base.py

@@ -8,7 +8,7 @@
 from datetime import datetime
 from enum import IntEnum
 from pathlib import Path
-from typing import Any, Dict, Iterator, List, Optional, TextIO, Tuple, Type, TypeVar, Union
+from typing import Any, Dict, Iterator, List, Optional, Set, TextIO, Tuple, Type, TypeVar, Union
 
 import networkx  # type:ignore
 
@@ -1042,24 +1042,75 @@ def _add_missing_input_defaults(component_inputs: Dict[str, Any], component_inpu
 
         return component_inputs
 
+    def _tiebreak_waiting_components(
+        self,
+        component_name: str,
+        priority: ComponentPriority,
+        priority_queue: FIFOPriorityQueue,
+        topological_sort: Union[Dict[str, int], None],
+    ):
+        """
+        Decides which component to run when multiple components are waiting for inputs with the same priority.
+
+        :param component_name: The name of the component.
+        :param priority: Priority of the component.
+        :param priority_queue: Priority queue of component names.
+        :param topological_sort: Cached topological sort of all components in the pipeline.
+        """
+        components_with_same_priority = [component_name]
+
+        while len(priority_queue) > 0:
+            next_priority, next_component_name = priority_queue.peek()
+            if next_priority == priority:
+                priority_queue.pop()  # actually remove the component
+                components_with_same_priority.append(next_component_name)
+            else:
+                break
+
+        if len(components_with_same_priority) > 1:
+            if topological_sort is None:
+                if networkx.is_directed_acyclic_graph(self.graph):
+                    topological_sort = networkx.lexicographical_topological_sort(self.graph)
+                    topological_sort = {node: idx for idx, node in enumerate(topological_sort)}
+                else:
+                    condensed = networkx.condensation(self.graph)
+                    condensed_sorted = {node: idx for idx, node in enumerate(networkx.topological_sort(condensed))}
+                    topological_sort = {
+                        component_name: condensed_sorted[node]
+                        for component_name, node in condensed.graph["mapping"].items()
+                    }
+
+            components_with_same_priority = sorted(
+                components_with_same_priority, key=lambda comp_name: (topological_sort[comp_name], comp_name.lower())
+            )
+
+            component_name = components_with_same_priority[0]
+
+        return component_name, topological_sort
+
     @staticmethod
     def _write_component_outputs(
-        component_name, component_outputs, inputs, receivers, include_outputs_from
+        component_name: str,
+        component_outputs: Dict[str, Any],
+        inputs: Dict[str, Any],
+        receivers: List[Tuple],
+        include_outputs_from: Set[str],
     ) -> Dict[str, Any]:
         """
         Distributes the outputs of a component to the input sockets that it is connected to.
 
         :param component_name: The name of the component.
         :param component_outputs: The outputs of the component.
         :param inputs: The current global input state.
-        :param receivers: List of receiver_name, sender_socket, receiver_socket for connected components.
+        :param receivers: List of components that receive inputs from the component.
         :param include_outputs_from: List of component names that should always return an output from the pipeline.
         """
         for receiver_name, sender_socket, receiver_socket in receivers:
             # We either get the value that was produced by the actor or we use the _NO_OUTPUT_PRODUCED class to indicate
             # that the sender did not produce an output for this socket.
             # This allows us to track if a pre-decessor already ran but did not produce an output.
             value = component_outputs.get(sender_socket.name, _NO_OUTPUT_PRODUCED)
+
             if receiver_name not in inputs:
                 inputs[receiver_name] = {}
 

haystack/core/pipeline/pipeline.py

@@ -2,7 +2,6 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
-import warnings
 from copy import deepcopy
 from typing import Any, Dict, Mapping, Optional, Set, cast
 
@@ -209,6 +208,8 @@ def run(  # noqa: PLR0915, PLR0912
         # We store them here for easy access.
         cached_receivers = {name: self._find_receivers_from(name) for name in ordered_component_names}
 
+        cached_topological_sort = None
+
         pipeline_outputs: Dict[str, Any] = {}
         with tracing.tracer.trace(
             "haystack.pipeline.run",
@@ -231,24 +232,23 @@ def run(  # noqa: PLR0915, PLR0912
                     break
 
                 priority, component_name, component = candidate
-                if len(priority_queue) > 0:
-                    next_priority, next_name = priority_queue.peek()
-
-                    if (
-                        priority in [ComponentPriority.DEFER, ComponentPriority.DEFER_LAST]
-                        and next_priority == priority
-                    ):
-                        msg = (
-                            f"Components '{component_name}' and '{next_name}' are waiting for "
-                            f"optional inputs at the same time. The pipeline will execute '{component_name}' "
-                            f"first based on lexicographical ordering."
-                        )
-                        warnings.warn(msg)
+                if len(priority_queue) > 0 and priority in [ComponentPriority.DEFER, ComponentPriority.DEFER_LAST]:
+                    component_name, topological_sort = self._tiebreak_waiting_components(
+                        component_name=component_name,
+                        priority=priority,
+                        priority_queue=priority_queue,
+                        topological_sort=cached_topological_sort,
+                    )
+                    cached_topological_sort = topological_sort
+                    component = self._get_component_with_graph_metadata_and_visits(
+                        component_name, component_visits[component_name]
+                    )
 
                 component_outputs = self._run_component(component, inputs, component_visits, parent_span=span)
 
                 # Updates global input state with component outputs and returns outputs that should go to
                 # pipeline outputs.
+
                 component_pipeline_outputs = self._write_component_outputs(
                     component_name=component_name,
                     component_outputs=component_outputs,

releasenotes/notes/fix-pipeline-components-waiting-edge-cases-51a3054e3f1adb6e.yaml

@@ -0,0 +1,8 @@
+---
+fixes:
+  - |
+    Added a fix to the pipeline's component scheduling alogrithm to reduce edge cases where the execution order of
+    components that are simultaneously waiting for inputs has an impact on a pipeline's output. We look at topolgical
+    order first to see which of the waiting components should run first and fall back to lexicographical order when both
+    components are on the same topology-level. In cyclic pipelines, if the waiting components are in the same cycle,
+    we fall back to lexicographical order immediately.

test/core/pipeline/features/pipeline_run.feature

@@ -52,6 +52,8 @@ Feature: Pipeline running
         | with a component that has dynamic default inputs |
         | with a component that has variadic dynamic default inputs |
         | that is a file conversion pipeline with two joiners |
+        | that is a file conversion pipeline with three joiners |
+        | that is a file conversion pipeline with three joiners and a loop |
         | that has components returning dataframes |
 
     Scenario Outline: Running a bad Pipeline