@provide_session<EOL><INDENT>def _process_backfill_task_instances(self,<EOL>ti_status,<EOL>executor,<EOL>pickle_id,<EOL>start_date=None, session=None):<DEDENT>
executed_run_dates = []<EOL>while ((len(ti_status.to_run) > <NUM_LIT:0> or len(ti_status.running) > <NUM_LIT:0>) and<EOL>len(ti_status.deadlocked) == <NUM_LIT:0>):<EOL><INDENT>self.log.debug("<STR_LIT>")<EOL>ti_status.not_ready.clear()<EOL>@provide_session<EOL>def _per_task_process(task, key, ti, session=None):<EOL><INDENT>ti.refresh_from_db()<EOL>task = self.dag.get_task(ti.task_id)<EOL>ti.task = task<EOL>ignore_depends_on_past = (<EOL>self.ignore_first_depends_on_past and<EOL>ti.execution_date == (start_date or ti.start_date))<EOL>self.log.debug(<EOL>"<STR_LIT>", ti, ti.state)<EOL>if ti.state == State.SUCCESS:<EOL><INDENT>ti_status.succeeded.add(key)<EOL>self.log.debug("<STR_LIT>", ti)<EOL>ti_status.to_run.pop(key)<EOL>if key in ti_status.running:<EOL><INDENT>ti_status.running.pop(key)<EOL><DEDENT>return<EOL><DEDENT>elif ti.state == State.SKIPPED:<EOL><INDENT>ti_status.skipped.add(key)<EOL>self.log.debug("<STR_LIT>", ti)<EOL>ti_status.to_run.pop(key)<EOL>if key in ti_status.running:<EOL><INDENT>ti_status.running.pop(key)<EOL><DEDENT>return<EOL><DEDENT>elif ti.state == State.NONE:<EOL><INDENT>self.log.warning(<EOL>"<STR_LIT>"<EOL>"<STR_LIT>"<EOL>)<EOL>ti.set_state(State.SCHEDULED, session=session)<EOL><DEDENT>if self.rerun_failed_tasks:<EOL><INDENT>if ti.state in (State.FAILED, State.UPSTREAM_FAILED):<EOL><INDENT>self.log.error("<STR_LIT>"<EOL>"<STR_LIT>".format(ti=ti,<EOL>state=ti.state))<EOL>if key in ti_status.running:<EOL><INDENT>ti_status.running.pop(key)<EOL><DEDENT>ti.set_state(State.SCHEDULED, session=session)<EOL><DEDENT><DEDENT>else:<EOL><INDENT>if ti.state in (State.FAILED, State.UPSTREAM_FAILED):<EOL><INDENT>self.log.error("<STR_LIT>"<EOL>"<STR_LIT>".format(ti=ti,<EOL>state=ti.state))<EOL>ti_status.failed.add(key)<EOL>ti_status.to_run.pop(key)<EOL>if key in ti_status.running:<EOL><INDENT>ti_status.running.pop(key)<EOL><DEDENT>return<EOL><DEDENT><DEDENT>backfill_context = DepContext(<EOL>deps=RUN_DEPS,<EOL>ignore_depends_on_past=ignore_depends_on_past,<EOL>ignore_task_deps=self.ignore_task_deps,<EOL>flag_upstream_failed=True)<EOL>if ti.are_dependencies_met(<EOL>dep_context=backfill_context,<EOL>session=session,<EOL>verbose=self.verbose):<EOL><INDENT>ti.refresh_from_db(lock_for_update=True, session=session)<EOL>if ti.state in (State.SCHEDULED, State.UP_FOR_RETRY, State.UP_FOR_RESCHEDULE):<EOL><INDENT>if executor.has_task(ti):<EOL><INDENT>self.log.debug(<EOL>"<STR_LIT>"<EOL>"<STR_LIT>",<EOL>ti<EOL>)<EOL><DEDENT>else:<EOL><INDENT>self.log.debug('<STR_LIT>', ti)<EOL>ti.state = State.QUEUED<EOL>ti.queued_dttm = timezone.utcnow() if not ti.queued_dttm else ti.queued_dttm<EOL>session.merge(ti)<EOL>cfg_path = None<EOL>if executor.__class__ in (executors.LocalExecutor,<EOL>executors.SequentialExecutor):<EOL><INDENT>cfg_path = tmp_configuration_copy()<EOL><DEDENT>executor.queue_task_instance(<EOL>ti,<EOL>mark_success=self.mark_success,<EOL>pickle_id=pickle_id,<EOL>ignore_task_deps=self.ignore_task_deps,<EOL>ignore_depends_on_past=ignore_depends_on_past,<EOL>pool=self.pool,<EOL>cfg_path=cfg_path)<EOL>ti_status.running[key] = ti<EOL>ti_status.to_run.pop(key)<EOL><DEDENT><DEDENT>session.commit()<EOL>return<EOL><DEDENT>if ti.state == State.UPSTREAM_FAILED:<EOL><INDENT>self.log.error("<STR_LIT>", ti)<EOL>ti_status.failed.add(key)<EOL>ti_status.to_run.pop(key)<EOL>if key in ti_status.running:<EOL><INDENT>ti_status.running.pop(key)<EOL><DEDENT>return<EOL><DEDENT>if ti.state == State.UP_FOR_RETRY:<EOL><INDENT>self.log.debug(<EOL>"<STR_LIT>"<EOL>"<STR_LIT>", ti)<EOL>if key in ti_status.running:<EOL><INDENT>ti_status.running.pop(key)<EOL><DEDENT>ti_status.to_run[key] = ti<EOL>return<EOL><DEDENT>if ti.state == State.UP_FOR_RESCHEDULE:<EOL><INDENT>self.log.debug(<EOL>"<STR_LIT>"<EOL>"<STR_LIT>", ti)<EOL>if key in ti_status.running:<EOL><INDENT>ti_status.running.pop(key)<EOL><DEDENT>ti_status.to_run[key] = ti<EOL>return<EOL><DEDENT>self.log.debug('<STR_LIT>', ti)<EOL>ti_status.not_ready.add(key)<EOL><DEDENT>non_pool_slots = conf.getint('<STR_LIT>', '<STR_LIT>')<EOL>try:<EOL><INDENT>for task in self.dag.topological_sort():<EOL><INDENT>for key, ti in list(ti_status.to_run.items()):<EOL><INDENT>if task.task_id != ti.task_id:<EOL><INDENT>continue<EOL><DEDENT>if task.pool:<EOL><INDENT>pool = session.query(models.Pool).filter(models.Pool.pool == task.pool).first()<EOL>if not pool:<EOL><INDENT>raise PoolNotFound('<STR_LIT>'.format(task.pool))<EOL><DEDENT>open_slots = pool.open_slots(session=session)<EOL>if open_slots <= <NUM_LIT:0>:<EOL><INDENT>raise NoAvailablePoolSlot(<EOL>"<STR_LIT>"<EOL>"<STR_LIT>".format(<EOL>open_slots, task.pool))<EOL><DEDENT><DEDENT>else:<EOL><INDENT>if non_pool_slots <= <NUM_LIT:0>:<EOL><INDENT>raise NoAvailablePoolSlot(<EOL>"<STR_LIT>"<EOL>"<STR_LIT>")<EOL><DEDENT>non_pool_slots -= <NUM_LIT:1><EOL><DEDENT>num_running_tasks = DAG.get_num_task_instances(<EOL>self.dag_id,<EOL>states=(State.QUEUED, State.RUNNING))<EOL>if num_running_tasks >= self.dag.concurrency:<EOL><INDENT>raise DagConcurrencyLimitReached(<EOL>"<STR_LIT>"<EOL>"<STR_LIT>"<EOL>)<EOL><DEDENT>_per_task_process(task, key, ti)<EOL><DEDENT><DEDENT><DEDENT>except (NoAvailablePoolSlot, DagConcurrencyLimitReached) as e:<EOL><INDENT>self.log.debug(e)<EOL><DEDENT>self.heartbeat()<EOL>executor.heartbeat()<EOL>if (ti_status.not_ready and<EOL>ti_status.not_ready == set(ti_status.to_run) and<EOL>len(ti_status.running) == <NUM_LIT:0>):<EOL><INDENT>self.log.warning(<EOL>"<STR_LIT>",<EOL>ti_status.to_run.values()<EOL>)<EOL>ti_status.deadlocked.update(ti_status.to_run.values())<EOL>ti_status.to_run.clear()<EOL><DEDENT>self._manage_executor_state(ti_status.running)<EOL>self._update_counters(ti_status=ti_status)<EOL>_dag_runs = ti_status.active_runs[:]<EOL>for run in _dag_runs:<EOL><INDENT>run.update_state(session=session)<EOL>if run.state in State.finished():<EOL><INDENT>ti_status.finished_runs += <NUM_LIT:1><EOL>ti_status.active_runs.remove(run)<EOL>executed_run_dates.append(run.execution_date)<EOL><DEDENT><DEDENT>self._log_progress(ti_status)<EOL><DEDENT>return executed_run_dates<EOL>
Process a set of task instances from a set of dag runs. Special handling is done
to account for different task instance states that could be present when running
them in a backfill process.

:param ti_status: the internal status of the job
:type ti_status: BackfillJob._DagRunTaskStatus
:param executor: the executor to run the task instances
:type executor: BaseExecutor
:param pickle_id: the pickle_id if dag is pickled, None otherwise
:type pickle_id: int
:param start_date: the start date of the backfill job
:type start_date: datetime.datetime
:param session: the current session object
:type session: sqlalchemy.orm.session.Session
:return: the list of execution_dates for the finished dag runs
:rtype: list