Tasks are starved due to incorrect scheduler prioritization

[nevcohen]

Apache Airflow version

2.10.4

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

2.9.3

What happened?

We have a problem with our airflow cluster.
When the scheduler extracts and filters the tasks for scheduling even though the pools are not in a state of starvation then certain tasks in certain pools get stuck in scheduled.

This happens due to the following situation:

There are two pools, each with 20 slots, on the first there are 1000 tasks ready for scheduling and each of them has a priority of 2, on the second pool there are 100 tasks ready for scheduling and each of them with a priority of 1.
The first pool all the slots are currently occupied except for 1 and the second, all slots are free.

What happens is, the scheduler sorts the 1100 tasks that are waiting in scheduled and each time pulls 32 (we set max_ti=32), but all the first 32 that the scheduler actually pulled after filtering belong to the first pool (it is not starving because some of its tasks are short and therefore always has a few single slots free) so it turns out that the tasks of the second pool are stuck in scheduled without being able to move forward, and in the first pool he gets 32 tasks each time, but in practice he can run a few individual tasks (2/3 tasks).

The same can be said about dags with lots of tasks but with very low concurrency.

What you think should happen instead?

Don't filter the tasks according to the pools/dags/tasks that are starving, but check how many free slots there are in each and filter (starved pool/dag/task) the amount of tasks according to each of them, before query.limit(max_tis)

How to reproduce

Pool PA: 3 slots

Pool PB: 20 slots

Dag DA: with 1000 shorts tasks, each task with priority 2 on pool PA.

Dag DB: with 100 shorts tasks, each task with priority 1 on pool PB.

Max active tasks = 32

Operating System

Debian GNU/Linux 12 (bookworn)

Versions of Apache Airflow Providers

apache-airflow-providers-cncf-kubernetes==8.4.2

Deployment

Official Apache Airflow Helm Chart

Deployment details

Python 3.8

executor: "KubernetesExecutor"

scheduler:

• max_tis_per_query: 32
• schedule_after_task_execution: 'False'
• task_queued_timeout_chheck_interval: 600

kubernetes_executor:

• worker_pods_ceartion_batch_size: 16

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

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[potiuk]

Indeed. looks plausible. While this is an edge case, I think yes, the logic we have for pool filtering works as described. I vaguely recall we've been discussing about similar scenario in the past. Maybe someone could take a look how to optimise it for Airflow 3.

@ashb -> maybe there wrere some other ideas for this pool filtering ?

[nevcohen]

Something we thought about here and maybe it could solve the problem.

We'll add the following variable below the line here:

max_tis_loop = max_tis

And this for will become:

for loop_count in range(1, max_tis_loop):
    ...

And the is_done variable will also be fixed:

max_tis -= len(executable_tis)
is_done = len(executable_tis) == max_tis_loop or len(task_instances_to_examine) < max_tis_loop

Hope that covers all use cases...

[eladkal]

@nevcohen will you submit a PR?

[nevcohen]

No change is planned in the above logic in airflow 3?

Like @potiuk said above:

Indeed. looks plausible. While this is an edge case, I think yes, the logic we have for pool filtering works as described. I vaguely recall we've been discussing about similar scenario in the past. Maybe someone could take a look how to optimise it for Airflow 3.

@ashb -> maybe there wrere some other ideas for this pool filtering ?

[Nataneljpwd]

After looking a little deeper, it looks like the solution recommended above would not solve this problem, all that is done there is to stay longer in the scheduler loop, it does not look as if it locks any additional tasks so each run mostly the same tasks will be returned from the query, and if it does lock, then the time spent in the critical section and amount of locked tasks increase, which both are things that we try to minimize, time in the critical section and amount of locked tasks.

The problem does not seem to lie in the scheduler loop but in the prioritization and the way airflow chooses which tasks to run first, so maybe in addition to the window solution, the way prioritization works in airflow should change

I think it might be a deeper problem and some interesting solutions could arise from a discussion

[IndiTheo]

We started a discussion on this issue to find a better strategy for the scheduler, so tasks won't be dropped like in the example provided here:

#49160

[eladkal]

Closing to keep the discussion in one place.
Please continue in #49508