Repository: apache/airflow
File: airflow-core/src/airflow/executors/base_executor.py (Python)

Code before:
        )

    def order_queued_tasks_by_priority(self) -> list[tuple[TaskInstanceKey, workloads.ExecuteTask]]:
        """
        Orders the queued tasks by priority.

        :return: List of workloads from the queued_tasks according to the priority.
        """
        if not self.queued_tasks:
            return []

        # V3 + new executor that supports workloads
        return sorted(
            self.queued_tasks.items(),
            key=lambda x: x[1].ti.priority_weight,
            reverse=True,
        )

    @add_debug_span
    def trigger_tasks(self, open_slots: int) -> None:
        """
        Initiate async execution of the queued tasks, up to the number of available slots.

        :param open_slots: Number of open slots
        """
        sorted_queue = deque(self.order_queued_tasks_by_priority())
        workload_list = []

        for _ in range(min((open_slots, len(self.queued_tasks)))):
            key, item = sorted_queue.popleft()

            # If a task makes it here but is still understood by the executor
            # to be running, it generally means that the task has been killed
            # externally and not yet been marked as failed.
            #
            # However, when a task is deferred, there is also a possibility of
            # a race condition where a task might be scheduled again during
            # trigger processing, even before we are able to register that the
            # deferred task has completed. In this case and for this reason,
            # we make a small number of attempts to see if the task has been
            # removed from the running set in the meantime.
            if key in self.attempts:
                del self.attempts[key]

            if isinstance(item, workloads.ExecuteTask) and hasattr(item, "ti"):
                ti = item.ti

                # If it's None, then the span for the current id hasn't been started.
                if self.active_spans is not None and self.active_spans.get("ti:" + str(ti.id)) is None:
                    if isinstance(ti, workloads.TaskInstance):
                        parent_context = Trace.extract(ti.parent_context_carrier)

Review comment (suggestion):
I believe this should be solved differently, as the reverse order doesn't make sense, we can sort the tasks in ascending order and pop from the end of the list here (which is O(1))

Code after:
        )

    def order_queued_tasks_by_priority(self) -> list[tuple[TaskInstanceKey, workloads.ExecuteTask]]:
        """
        Orders the queued tasks by priority.

        :return: List of workloads from the queued_tasks according to the priority.
        """
        if not self.queued_tasks:
            return []

        # V3 + new executor that supports workloads
        return sorted(
            self.queued_tasks.items(),
            key=lambda x: x[1].ti.priority_weight,
            reverse=False,
        )

    @add_debug_span
    def trigger_tasks(self, open_slots: int) -> None:
        """
        Initiate async execution of the queued tasks, up to the number of available slots.

        :param open_slots: Number of open slots
        """
        sorted_queue = self.order_queued_tasks_by_priority()
        workload_list = []

        for _ in range(min((open_slots, len(self.queued_tasks)))):
            key, item = sorted_queue.pop()

            # If a task makes it here but is still understood by the executor
            # to be running, it generally means that the task has been killed
            # externally and not yet been marked as failed.
            #
            # However, when a task is deferred, there is also a possibility of
            # a race condition where a task might be scheduled again during
            # trigger processing, even before we are able to register that the
            # deferred task has completed. In this case and for this reason,
            # we make a small number of attempts to see if the task has been
            # removed from the running set in the meantime.
            if key in self.attempts:
                del self.attempts[key]

            if isinstance(item, workloads.ExecuteTask) and hasattr(item, "ti"):
                ti = item.ti

                # If it's None, then the span for the current id hasn't been started.
                if self.active_spans is not None and self.active_spans.get("ti:" + str(ti.id)) is None:
                    if isinstance(ti, workloads.TaskInstance):
                        parent_context = Trace.extract(ti.parent_context_carrier)