Repository: apache/airflow
File: airflow-core/src/airflow/jobs/scheduler_job_runner.py (Python)

Code before:

        pool_num_starving_tasks: dict[str, int] = Counter()

        for loop_count in itertools.count(start=1):
            num_starved_pools = len(starved_pools)
            num_starved_dags = len(starved_dags)
            num_starved_tasks = len(starved_tasks)
            num_starved_tasks_task_dagrun_concurrency = len(starved_tasks_task_dagrun_concurrency)

            dr_task_concurrency_subquery = _get_current_dr_task_concurrency(states=EXECUTION_STATES)

            query = (
                select(TI)
                .with_hint(TI, "USE INDEX (ti_state)", dialect_name="mysql")
                .join(TI.dag_run)
                .where(DR.state == DagRunState.RUNNING)
                .join(TI.dag_model)
                .where(~DM.is_paused)
                .where(TI.state == TaskInstanceState.SCHEDULED)
                .where(DM.bundle_name.is_not(None))
                .join(
                    dr_task_concurrency_subquery,
                    TI.run_id == dr_task_concurrency_subquery.c.run_id,
                    isouter=True,
                )
                .where(func.coalesce(dr_task_concurrency_subquery.c.dr_count, 0) < DM.max_active_tasks)
                .options(selectinload(TI.dag_model))
                .order_by(-TI.priority_weight, DR.logical_date, TI.map_index)
            )

            # Create a subquery with row numbers partitioned by run_id.
            #
            # run_id    | task_id | priority_weight | row_num
            # ----------|---------|-----------------|--------
            # dag1_dr1  | task1   | 100             | 1
            # dag1_dr1  | task22  | 90              | 2
            # dag1_dr1  | task5   | 80              | 3
            # dag1_dr1  | task13  | 70              | 4
            # dag2_dr1  | task3   | 95              | 1
            # dag2_dr1  | task1   | 85              | 2
            # dag2_dr1  | task5   | 75              | 3
            ranked_query = (
                query.add_columns(
                    func.row_number()
                    .over(
                        partition_by=TI.run_id,
                        order_by=[-TI.priority_weight, DR.logical_date, TI.map_index],
                    )
                    .label("row_num"),
                    DM.max_active_tasks.label("dr_max_active_tasks"),
                    # Create columns for the order_by checks here for sqlite.

Review comment (bug):
Hmm, `run_id` is only unique per DAG, but this subquery groups by `run_id` alone and the join uses just `TI.run_id`. When one DAG reaches `max_active_tasks`, every other DAG that happens to share the same `run_id` string (common for scheduled runs) now fails the `< DM.max_active_tasks` check and never gets considered. 

Better to group by `(dag_id, run_id)` and join on both columns instead?

https://github.com/apache/airflow/blob/085b0edfe7c7151517cd727020c13a2834409210/airflow-core/src/airflow/models/dagrun.py#L233

Code after:
        if not _is_parent_process():
            # Only the parent process should perform the debug dump.
            return

        try:
            sig_name = signal.Signals(signum).name
        except Exception:
            sig_name = str(signum)

        self.log.info("%s\n%s received, printing debug\n%s", "-" * 80, sig_name, "-" * 80)

        for executor in self.job.executors:
            self.log.info("Debug dump for the executor %s", executor)
            executor.debug_dump()
            self.log.info("-" * 80)

        id2name = {th.ident: th.name for th in threading.enumerate()}
        for threadId, stack in sys._current_frames().items():
            self.log.info("Stack Trace for Scheduler Job Runner on thread: %s", id2name[threadId])
            callstack = extract_stack(f=stack, limit=10)
            self.log.info("\n\t".join(map(repr, callstack)))
            self.log.info("-" * 80)

    def _executable_task_instances_to_queued(self, max_tis: int, session: Session) -> list[TI]:
        """
        Find TIs that are ready for execution based on conditions.

        Conditions include:
        - pool limits
        - DAG max_active_tasks
        - executor state
        - priority
        - max active tis per DAG
        - max active tis per DAG run

        :param max_tis: Maximum number of TIs to queue in this loop.
        :return: list[airflow.models.TaskInstance]
        """
        from airflow.models.pool import Pool
        from airflow.utils.db import DBLocks

        executable_tis: list[TI] = []

        if get_dialect_name(session) == "postgresql":
            # Optimization: to avoid littering the DB errors of "ERROR: canceling statement due to lock
            # timeout", try to take out a transactional advisory lock (unlocks automatically on
            # COMMIT/ROLLBACK)
            lock_acquired = session.execute(
                text("SELECT pg_try_advisory_xact_lock(:id)").bindparams(
                    id=DBLocks.SCHEDULER_CRITICAL_SECTION.value
                )