https://github.com/apache/airflow/blob/main/airflow/dag_processing/manager.py#L773
"""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.popleft()
            
            if file_path in self._processors.keys():
                continue
            callback_to_execute_for_file = self._callback_to_execute[file_path]
            processor = self._create_process(
                file_path,
                self._pickle_dags,
                self._dag_ids,
                self.get_dag_directory(),
                callback_to_execute_for_file,
            )
            del self._callback_to_execute[file_path]
            Stats.incr("dag_processing.processes", tags={"file_path": file_path, "action": "start"})
            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
    def add_new_file_path_to_queue(self):
        for file_path in self.file_paths:
            if file_path not in self._file_stats:
                
                self.log.info("Adding new file %s to parsing queue", file_path)
                self._file_stats[file_path] = DagFileStat(
                    num_dags=0, import_errors=0, last_finish_time=None, last_duration=None, run_count=0
                )
                self._file_path_queue.appendleft(file_path)
    def prepare_file_path_queue(self):