Dont create duplicate dag file processors

[pingzh]

Context: when a dag file is under processing and multiple callbacks are
created either via zombies or executor events, the dag file is added to
the _file_path_queue and the manager will launch a new process to
process it, which it should not since the dag file is currently under
processing. This will bypass the _parallelism eventually especially when
it takes long time to process some dag files. We have seen ~200 dag
processors on the scheduler even we set the _parallelism as 60. More dag
file processors make CPU spike and in turn it makes the dag file
processing even slower. In the end, the scheduler is taken down.

---

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code change, Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in UPDATING.md.

[ashb]

This is too tightly coupled to the way this is implemented, rather than the goal.

Also the fact that you can't esaily test this probably means you need to re-think the a.pproach

[pingzh]

if the manager loop calls prepare_file_path_queue() in every loop, we don't need this change (passing files_paths_to_exclude_in_this_loop to the start_new_processes.

It is only called when if not self._file_path_queue. If we remove this, we will also need to to some dupe work in the prepare_file_path_queue

[pingzh]

i chose to have files_paths_to_exclude_in_this_loop inside the start_new_processes as i was thinking that is single place we do a pop(0) and start new process, so adding a check will guarantee it won't double process dag files.

[ashb]

Are you sure this is still a problem on mainline. Looking at start_new_processes:

    def start_new_processes(self):
        """Start more processors if we have enough slots and files to process"""
        while self._parallelism - len(self._processors) > 0 and self._file_path_queue:
            file_path = self._file_path_queue.pop(0)
            callback_to_execute_for_file = self._callback_to_execute[file_path]
            processor = self._processor_factory(
                file_path,
                callback_to_execute_for_file,
                self._dag_ids,
                self._pickle_dags)

            del self._callback_to_execute[file_path]
            Stats.incr('dag_processing.processes')

            processor.start()
            self.log.debug(
                "Started a process (PID: %s) to generate tasks for %s",
                processor.pid, file_path
            )
            self._processors[file_path] = processor
            self.waitables[processor.waitable_handle] = processor

I can't see at first glance how self._parallelism - len(self._processors) > 0 would ever lead to too many processes.

[pingzh]

Are you sure this is still a problem on mainline. Looking at start_new_processes:

    def start_new_processes(self):
        """Start more processors if we have enough slots and files to process"""
        while self._parallelism - len(self._processors) > 0 and self._file_path_queue:
            file_path = self._file_path_queue.pop(0)
            callback_to_execute_for_file = self._callback_to_execute[file_path]
            processor = self._processor_factory(
                file_path,
                callback_to_execute_for_file,
                self._dag_ids,
                self._pickle_dags)

            del self._callback_to_execute[file_path]
            Stats.incr('dag_processing.processes')

            processor.start()
            self.log.debug(
                "Started a process (PID: %s) to generate tasks for %s",
                processor.pid, file_path
            )
            self._processors[file_path] = processor
            self.waitables[processor.waitable_handle] = processor

I can't see at first glance how self._parallelism - len(self._processors) > 0 would ever lead to too many processes.

Yes, it still an issue, the main logic to launch new dag file processes does not change much between 1.10.4 and the master branch. We also cherry-picked this PR #7597 to our 1.10.4 version.

The issue does not happen often.

This is the incident leading us to find this issue. As you can see, the same dag file is processing many times (the dag process for this dag file usually takes more than 15 min)

when the manager adds the callback to the _file_path_queue, it does not care whether this dag file is currently under processing or in the cool down period, which leads to multiple dag processes processing the same dag file.

As for the exceed of the _parallelism, I have lost some context about how exactly it got into that state :(

[ashb]

when the manager adds the callback to the _file_path_queue, it does not care whether this dag file is currently under processing or in the cool down period,

Yes, this is "by design" -- if there's a callback we need to execute it "now" to run it.

https://github.com/apache/airflow/blob/master/airflow/utils/dag_processing.py#L713-L718

Given we remove it form the list if it's already there, and the existing checks I'm not sure the two "issues" are related (1- jumping up the queue/ignoring cool down period, and 2- going beyond concurrency limits)

[pingzh]

when the manager adds the callback to the _file_path_queue, it does not care whether this dag file is currently under processing or in the cool down period,

Yes, this is "by design" -- if there's a callback we need to execute it "now" to run it.

https://github.com/apache/airflow/blob/master/airflow/utils/dag_processing.py#L713-L718

Given we remove it form the list if it's already there, and the existing checks I'm not sure the two "issues" are related (1- jumping up the queue/ignoring cool down period, and 2- going beyond concurrency limits)

However, the original design has a risk of letting a dag file with a long processing time to take over all dag processors and also introduce a race condition

Race condition case:

1. a task of a dag is treated as zombie
2. it is put to the event queue and then being processed by the DagProcesssor. The dag file takes very long time to process dag file and runs the callback.
3. the next loop in the dag manager sees the same task as zombie and puts it tot he event queue
4. a new DagProcessor is launched to process the same dag file and also the same zombie event

[pingzh]

@kaxil and @ashb friendly remind of this PR. thanks

[potiuk]

Hey @kaxil @ashb - do your want to make it part of 2.0.0rc1 ? O should we change the milestone?

[ashb]

I'm not yet convinced that this is a) actually a bug, or b) the right solution.

Changing the milestone for now, and it can come in 2.0.x or 2.1