Repository: apache/airflow
File: airflow-core/src/airflow/dag_processing/manager.py (Python)

Code before:
    def _add_files_to_queue(
        self,
        files: list[DagFileInfo],
        *,
        mode: Literal["front", "back", "frontprio"],
    ):
        """Add stuff to the back or front of the file queue, unless it's already present."""
        if mode == "frontprio":
            for file in files:
                # Try removing the file if already present
                with contextlib.suppress(ValueError):
                    self._file_queue.remove(file)
                # enqueue file to the start of the queue.
                self._file_queue.appendleft(file)
        elif mode == "front":
            new_files = list(f for f in files if f not in self._file_queue)
            self._file_queue.extendleft(new_files)
        elif mode == "back":
            new_files = list(f for f in files if f not in self._file_queue)
            self._file_queue.extend(new_files)
        else:
            assert_never(mode)

        # If we've just added files to the queue for the first time since metrics were last emitted, reset the
        # parse time counter.
        if not self._parsing_start_time and self._file_queue:
            self._parsing_start_time = time.perf_counter()

        Stats.gauge("dag_processing.file_path_queue_size", len(self._file_queue))

    def max_runs_reached(self):
        """:return: whether all file paths have been processed max_runs times."""
        if self.max_runs == -1:  # Unlimited runs.
            return False
        if self._num_run < self.max_runs:
            return False
        return all(stat.run_count >= self.max_runs for stat in self._file_stats.values())

    def terminate(self):
        """Stop all running processors."""
        for file, processor in self._processors.items():
            # todo: AIP-66 what to do about file_path tag? replace with bundle name and rel path?
            Stats.decr(
                "dag_processing.processes", tags={"file_path": str(file.rel_path), "action": "terminate"}
            )
            # SIGTERM, wait 5s, SIGKILL if still alive
            processor.kill(signal.SIGTERM, escalation_delay=5.0)

    def end(self):
        """Kill all child processes on exit since we don't want to leave them as orphaned."""
        pids_to_kill = [p.pid for p in self._processors.values()]

Review comment (suggestion):
```suggestion
        if self._parsing_start_time is not None and self._file_queue:
```

Code after:
        mode: Literal["front", "back", "frontprio"],
    ):
        """Add stuff to the back or front of the file queue, unless it's already present."""
        if mode == "frontprio":
            for file in files:
                # Try removing the file if already present
                with contextlib.suppress(ValueError):
                    self._file_queue.remove(file)
                # enqueue file to the start of the queue.
                self._file_queue.appendleft(file)
        elif mode == "front":
            new_files = list(f for f in files if f not in self._file_queue)
            self._file_queue.extendleft(new_files)
        elif mode == "back":
            new_files = list(f for f in files if f not in self._file_queue)
            self._file_queue.extend(new_files)
        else:
            assert_never(mode)

        # If we've just added files to the queue for the first time since metrics were last emitted, reset the
        # parse time counter.
        if self._parsing_start_time is None and self._file_queue:
            self._parsing_start_time = time.perf_counter()

        Stats.gauge("dag_processing.file_path_queue_size", len(self._file_queue))

    def max_runs_reached(self):
        """:return: whether all file paths have been processed max_runs times."""
        if self.max_runs == -1:  # Unlimited runs.
            return False
        if self._num_run < self.max_runs:
            return False
        return all(stat.run_count >= self.max_runs for stat in self._file_stats.values())

    def terminate(self):
        """Stop all running processors."""
        for file, processor in self._processors.items():
            # todo: AIP-66 what to do about file_path tag? replace with bundle name and rel path?
            Stats.decr(
                "dag_processing.processes", tags={"file_path": str(file.rel_path), "action": "terminate"}
            )
            # SIGTERM, wait 5s, SIGKILL if still alive
            processor.kill(signal.SIGTERM, escalation_delay=5.0)

    def end(self):
        """Kill all child processes on exit since we don't want to leave them as orphaned."""
        pids_to_kill = [p.pid for p in self._processors.values()]
        if pids_to_kill:
            kill_child_processes_by_pids(pids_to_kill)