https://github.com/apache/airflow/blob/main/airflow/dag_processing/manager.py#L773
all_files_processed = all(self.get_last_finish_time(x) is not None for x in self.file_paths)
            max_runs_reached = self.max_runs_reached()
            try:
                if self._direct_scheduler_conn:
                    self._direct_scheduler_conn.send(
                        DagParsingStat(
                            max_runs_reached,
                            all_files_processed,
                        )
                    )
            except BlockingIOError:
                
                
                
                
                
                self.log.debug("BlockingIOError received trying to send DagParsingStat, ignoring")
            if max_runs_reached:
                self.log.info(
                    "Exiting dag parsing loop as all files have been processed %s times", self._max_runs
                )
                break
            if self._async_mode:
                loop_duration = time.monotonic() - loop_start_time
                if loop_duration < 1:
                    poll_time = 1 - loop_duration
                else:
                    poll_time = 0.0
    @provide_session
    def _fetch_callbacks(self, max_callbacks: int, session: Session = NEW_SESSION):