DAG Runs from a single DAG can prevent scheduler from seeing other DAG's runs

[collinmcnulty]

Apache Airflow version

2.10.5

If "Other Airflow 2 version" selected, which one?

No response

What happened?

Two DAGs each receive a large batch of DAG Runs. The number of runs for each DAG exceeds max_dagruns_per_loop_to_schedule. Each DAG run is very short, shorter than the heartrate of this Airflow deployment. Both DAGs have a max_active_runs that is far less than dagruns_per_loop.

So: max_active_runs < max_dagruns_per_loop_to_schedule < number of queued DAG runs.

Each scheduler loop, there are a very small number of DAG Run "slots" for the first DAG, so the check coalesce(running_drs.c.num_running, text("0")) < coalesce(Backfill.max_active_runs, DagModel.max_active_runs), does not apply. But then all the DAG runs that are considered are from the first DAG. So Second DAG effectively has to wait for nearly all of First DAG's runs to complete before any of its runs are moved from queued to running.

What you think should happen instead?

I think the "most correct" thing to do is to change the global yes/no for a DAG being included in the check on the basis of max_active_runs to some kind of limit on the number for that DAG that can be included. I can't see a good way to do this in SQL but others may have insight.

Alternatively, because this is predominantly a problem when a single DAG dominates the scheduler's attention, we could add an explicit check to see if the result of the DAG run query contains only the a single DAG, and if so re-run the query with that DAG excluded.

How to reproduce

1. Create two DAGs with a single, simple task.
2. Set max_active_runs=100
3. Set max_dagruns_per_loop_to_schedule=2000
4. Start 5000 Runs of the first DAG
5. Start 5000 Runs of the second DAG
6. Hard to reproduce: keep the heartrate of the scheduler low enough that Runs complete within one scheduler loop.

Operating System

Debian GNU/Linux 12 (bookworm)

Versions of Apache Airflow Providers

No response

Deployment

Astronomer

Deployment details

No response

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[collinmcnulty]

The workaround we're using is to set max_dagruns_per_loop_to_schedule to a value greater than the largest number of Runs we'll get for any individual DAG.

[uranusjr]

The weird thing to me is we seems to have a mechanism for this

airflow/airflow-core/src/airflow/models/dagrun.py

Lines 522 to 539 in e4957ff

	query = (
	select(cls)
	.with_hint(cls, "USE INDEX (idx_dag_run_running_dags)", dialect_name="mysql")
	.where(cls.state == DagRunState.RUNNING)
	.join(DagModel, DagModel.dag_id == cls.dag_id)
	.join(BackfillDagRun, BackfillDagRun.dag_run_id == DagRun.id, isouter=True)
	.where(
	DagModel.is_paused == false(),
	DagModel.is_stale == false(),
	)
	.options(joinedload(cls.task_instances))
	.order_by(
	nulls_first(BackfillDagRun.sort_ordinal, session=session),
	nulls_first(cls.last_scheduling_decision, session=session),
	cls.run_after,
	)
	.limit(cls.DEFAULT_DAGRUNS_TO_EXAMINE)
	)

This should prioritize runs we have not touched (nulls first), or touched the least recently. So even if a run is not seen in an iteration, runs before it (and thus touched) should be deprioritized until all other runs were also touched, and so on, so each run should get a chance at some point. Maybe the last_scheduling_decision timestamp is not updated correctly?

[uranusjr]

I tried to reduce this to a reasonable minimal scenario. This is against main since we did not notice any significant logic changes in the area.

• Two dags 1 and 2 both have one single task. Both dags set max_active_runs=1. The task simply does sleep(5).
• Set AIRFLOW__SCHEDULER__MAX_DAGRUNS_PER_LOOP_TO_SCHEDULE to 1.
• Trigger 10 runs for 1, and then 10 for 2, both via API (I used a Python script with requests so the ordering is deterministic).

After triggering, I can observe the scheduler actually alternates between the two dags. It picks up a run from the first dag, and then a run from the second, the first, the second, and so on. Interetingly, I did not see any logs on max_active_runs at all; the scheduler seems to just do it out of the box.

So I think we’re missing something. Not sure what…

DAG code

from airflow.providers.standard.operators.python import PythonOperator
from airflow.sdk import DAG

def wait():
    import time
    time.sleep(5)

with DAG("1", is_paused_upon_creation=False, max_active_runs=1):
    PythonOperator(task_id="t", python_callable=wait)

with DAG("2", is_paused_upon_creation=False, max_active_runs=1):
    PythonOperator(task_id="t", python_callable=wait)

Triggering script

import datetime
import sys

import requests

def main(*dag_ids):
    r = requests.post(
        "http://localhost:8080/auth/token",
        json={"username": "admin", "password": "admin"},
    )
    r.raise_for_status()
    token = r.json()["access_token"]
    for dag_id in dag_ids:
        for i in range(10):
            r = requests.post(
                f"http://localhost:8080/api/v2/dags/{dag_id}/dagRuns",
                json={
                    "dag_run_id": f"{dag_id}_{i}",
                    "logical_date": datetime.datetime.now(datetime.timezone.utc).isoformat(),
                },
                headers={"Authorization": f"Bearer {token}"},
            )
            print(r.json())
            r.raise_for_status()


if __name__ == "__main__":
    _, *dag_ids = sys.argv
    main(*dag_ids)

# Run like this: python script.py 1 2

scheduler-output.txt

[eladkal]

It seems that starvation can happen. Also reported in #45636

Given 2 pools default_pool and special_pool with 2 DAGs:

from datetime import datetime
from airflow import DAG
from airflow.decorators import task
import time

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2023, 2, 1)
}

with DAG('scheduler1', schedule="@daily", catchup=True, default_args=default_args):

    @task()
    def sleep_it1():
        time.sleep(3600) # 1 hour
    sleep_it1()

with DAG('scheduler2', schedule=None, catchup=False, default_args=default_args):
    @task(pool="special_pool")
    def sleep_it2():
        time.sleep(60) # 1 min

    sleep_it2()

The scheduler1 dag creates many dagruns as it has catchup. Once those runs are created there is large "backlog" of tasks to complete, create a run in scheduler2. The task sleep_it2 will not be executed till all runs of sleep_it1 are finished. That is though in each loop of the scheduler it can not really schedule tasks of sleep_it1 but can schedule sleep_it2.
I waited for more than 30 min and sleep_it2 did not get scheduled though it can be.

This happens both in Airflow 2 and Airflow 3.

[collinmcnulty]

Aha, yes. That must be the missing ingredient that made it "hard" to reproduce. Still don't fully understand why each DAG Run in DAG 1 don't get marked with scheduling_decision and then DAG 2 is checked. It seems like the scheduler should be able to check all runs faster than 30 minutes.

[eladkal]

Aha, yes. That must be the missing ingredient that made it "hard" to reproduce. Still don't fully understand why each DAG Run in DAG 1 don't get marked with scheduling_decision and then DAG 2 is checked. It seems like the scheduler should be able to check all runs faster than 30 minutes.

Because when we read records from the database in the scheduler critical section we read as bulk. We don't have extra logic as we want this part to be as fast as possible. The logic comes after and you apply it on the bulk records you read (AIRFLOW__SCHEDULER__MAX_TIS_PER_QUERY)

[jscheffl]

We also had such issue (had a talk about this in https://airflowsummit.org/sessions/2024/how-we-tuned-our-airflow-to-make-1-2-million-dag-runs-per-day/) and I think my colleguage @AutomationDev85 supplied a fix to Airflow 2.6 or 2.7... we our problem in this area were gone.

But one important remark: We ensured that the amount of parallel DAG runs is limited. We triggered new DAG runs in state "queued" and not "running", then the scheduler saves efforts in scheduling "not ready" runs. And was only adding "running" state at a point when sufficient DAGs have been worked-off. Can you try this?

[eladkal]

Can you try this?

Your approach definitely will work. In practice you bypassed the scheduler :)

But I can't really use it. The environment is shared by many dag authors. I can't predict when bottle necks will occur and I don't want to act as a "police" over the cluster. My goal is to find a native solution, I think by now we know this is more common than just a unique edge case so we need to find a proper solutions without workarounds.

[Asquator]

Looks convincing. According to OP, the second DAG's runs are stuck at queued and not getting scheduled, so it shouldn't be related to get_running_dag_runs_to_examine or pools or any issues with the critical section, because queued -> running transition for DAG runs happens much earlier than we get to any scheduling decisions for tasks (like creating TIs or scheduling them in CS). The whole logic for transitioning DAG runs to running is in SchedulerJobRunner._start_queued_dagruns and DagRun.get_queued_dag_runs_to_set_running, and the described behaviour is a consequence of the optimistic scheduling strategy used across Airflow not being able to handle a variety of edge cases.

We first pull at most DEFAULT_DAGRUNS_TO_EXAMINE DAG runs whose DAGs are not yet maxed out:

airflow/airflow-core/src/airflow/models/dagrun.py

Lines 602 to 611 in e4957ff

	.where(
	# there are two levels of checks for num_running
	# the one done in this query verifies that the dag is not maxed out
	# it could return many more dag runs than runnable if there is even
	# capacity for 1. this could be improved.
	coalesce(running_drs.c.num_running, text("0"))
	< coalesce(Backfill.max_active_runs, DagModel.max_active_runs),
	# don't set paused dag runs as running
	not_(coalesce(Backfill.is_paused, False)),
	)

And then go over each one of them increasing the currently scheduled DAGrun count and dropping once we can't stuff any more runs:

airflow/airflow-core/src/airflow/jobs/scheduler_job_runner.py

Line 1740 in 65be581

for dag_run in dag_runs:

airflow/airflow-core/src/airflow/jobs/scheduler_job_runner.py

Lines 1762 to 1773 in 65be581

	elif dag.max_active_runs:
	if active_runs >= dag.max_active_runs:
	# todo: delete all candidate dag runs for this dag from list right now
	self.log.info(
	"dag cannot be started due to dag max_active_runs constraint; "
	"active_runs=%s max_active_runs=%s dag_id=%s run_id=%s",
	active_runs,
	dag_run.max_active_runs,
	dag_run.dag_id,
	dag_run.run_id,
	)
	continue

It's indeed the same as #45636, but now with DAG runs instead of tasks. If the original issue can be efficiently solved with windowing, I guess this one can also benefit from it, like:

1. Window partitioned by DAG
2. For each DAG compute the number of runs
3. Stuff new runs as far as there are free slots (per DAG)
4. Put a limit on the result (which can be far less restrictive, as we delegate more work to SQL and the cost of a query won't go up as fast as throughput will).

As windowing may introduce additional overhead which sometimes won't result in higher throughput, assuming there aren't so many DAG runs in the system, I'm looking for a compromise between the optimistic and the windowing (pessimistic) approaches. It's possible to have it configurable, though it will make the system harder and harder to tweak to everyone's needs, which is not desirable. I hope the windowing can be done in such a way that will be negligible for the query time, by the help of correct indexing, or completely changing some core concurrency parameters, like introducing DEFAULT_DAGS_TO_EXAMINE in favor of DEFAULT_DAGRUNS_TO_EXAMINE, where the new parameter limits the number of windows in the query. Some research is needed to find out.