Deferrable Tasks do not Respect Task Pools

[ReadytoRocc]

Apache Airflow version

2.2.3 (latest released)

What happened

When using a deferrable operator, the pool slot is released after the task goes into a state of deferred.

What you expected to happen

I would expect the task to retain the pool slot for the entire execution across the worker and triggerer resources, and only release the pool once it reaches a terminal state.

How to reproduce

Create a DAG the below dag, and a Pool named async with a slot count of 1. Let this dag run, and notice that tasks first 16 tasks enter queued -> running -> deferred.

from datetime import datetime
from airflow import DAG
from airflow.sensors.date_time import DateTimeSensorAsync

with DAG(
    "async_dag",
    start_date=datetime(2022, 1, 18),
    schedule_interval="* * * * *",
    catchup=True,
) as dag:

    async_sensor = DateTimeSensorAsync(
        task_id="async_task",
        target_time="""{{ macros.datetime.utcnow() + macros.timedelta(minutes=20) }}""",
        pool="async",
    )

Operating System

Debian GNU/Linux 11 (bullseye)

Versions of Apache Airflow Providers

No response

Deployment

Other Docker-based deployment

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

[potiuk]

I think this is an intended feature, not a bug.

It's the matter of assumption/definiton of whether the deferred task takes the pool slot or not. I think (but I would love to hear what others think) this is quite a reasonable assumption that deferred task does not take a slot.

Taking into account that the deferred task - by definition takes almost no resources and any external communication is asynchronous and pool feature is designed to limit resources an potentially decrease the loaad/throttle external, mostly synchronous services, I think not taking a pool slot by deferred task is quite reasonable.

Does the task take the slot back when it is back from defferred state @ReadytoRocc ?

[ReadytoRocc]

Does the task take the slot back when it is back from defferred state?

@potiuk yes. Once the trigger is finished, the task will be marked as SCHEDULED. The task will then re-queue once a pool slot is available.

[potiuk]

So I'd say it's pretty expected behaviour. I wonder what others think.

[potiuk]

@andrewgodwin @uranusjr - WDYT ? I think the behaviour is intended and maybe just need a bit clarification in the docs.

[andrewgodwin]

This was indeed originally by design, as I saw pools as a "worker/active task slot" system, designed to concurrency-limit things that were running - and of course, deferred tasks are not really running.

I think either option here is reasonable, and we could change it to have deferred tasks use up a pool slot if that's the consensus, but I went with this behaviour in the initial feature as it seemed most "sensible". It's probably worth documenting it explicitly, though?

[potiuk]

I agree with @andrewgodwin @ReadytoRocc - is that explanation/behaviour plausible for you ?

[potiuk]

In short:

[ReadytoRocc]

@andrewgodwin & @potiuk thank you for the detailed responses and for confirming this behavior is expected. Are we able to convert this issue into a discussion? I believe there are some use cases for using pools with deferrable operators.

An example would be if you want to limit the use of an external service that Airflow is orchestrating. Another would be to limit certain sets of tasks from using all of the triggerer’s resources.

Please let me know your thoughts, thanks!

[ReadytoRocc]

@potiuk - I have worked on some ideas based on this discussion, and would like to continue the work so I can contribute them. Do you recommend opening a new issue, or can we turn this discussion into a feature request?

[potiuk]

Ideally - turn it into PR and lead it to completion.

Feature requests have very little value here. They are often simply closed and generally there are no more than "wishlist". And if no-one volunteers to do it, it will be there, open and nothing will happen to it. The exsting if "feature" request does not make it any closer to implementation - but implementation in PR does.

If you want to make a proposal that you would like to get comments, devlist thread is much better, especially if you want to propose a significant change in how airflow works. This is IMHO borderline - adding existing pool feature to deferred tasks/triggerrer makes sense and whilte not terribly necessary - seems like a natural progression of the "defferable" operators (also @andrewgodwin commented in the same tone). So I think it does not need neither AIP nor discussion in the devlist.

So then, the next big think is make a PR to implement it. We do not need "issue" to create a PR. We are very light-weight and as-low-beauracracy as possible, so if you want to contribute a change .... just contribute it :).

The discussions in PR - even if it won't be yet complete but will show actual code, is almost always better than creating an issue describing what kind of code you are going to contribute. Worst thing that can happen - we will close the PR, best thing - it will get through reviews and get merged :).

[ReadytoRocc]

Thanks @potiuk for the guidance. I will create a Feature Pull Request 👍.

[dstandish]

While I understand the motivation, I don't really think we should add another state for this.
Look at how many states we have:

It's kindof extreme already. How is the airflow user supposed to remember which colors refer to what? It takes up too much horizontal space on the UI as it is.
I think states need a refactor -- execution state vs resolution etc.
and, there's already a PR (#26249) ready for merge that gives visibility into what's happening with your deferred task

BUT... just one person's opinion 🤷

[dstandish]

maybe we benefit from something like the "pod phase" of kubernetes.... a high level summary of task state.... not intended to be a fully comprehensive state machine.... then we can go wild with detailed statuses to freely support the workings of the scheduler etc with worrying about kludging up the UI and UX...

[notatallshaw-gts]

FYI I am relying on the behavior that given an async operator the trigger task doesn't take up a pool slot but the worker processes does.

We have occasional performance issue with lots of processes starting at once (e.g. say we want to start 50 FileSensorAsyncs), but we don't have any performance issues with the triggerer running many trigger tasks. So we put these async operators under a pool and it limits too many starting or stopping all at once.

Given the different runtime context of trigger tasks vs. worker processes I feel that if pools applied to both I would want a way to distinguish between them and get different behavior (like we do now)

[potiuk]

Any proposals - PR that can be disucssed in detail on code proposal, are most welcome.

[Usiel]

I added a PR at #32709 that works for me in some basic testing. It introduces a new flag on Pool that allows the admin to decide whether to include deferred tasks in the occupied pool slots count.