'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
:param session: the current session object
:type session: Session
:return: the list of execution_dates for the finished dag runs
:rtype: list'

@provide_session
def _process_backfill_task_instances(self, ti_status, executor, pickle_id, start_date=None, session=None):

    executed_run_dates = []
    while (((len(ti_status.to_run) > 0) or (len(ti_status.started) > 0)) and (len(ti_status.deadlocked) == 0)):
        self.logger.debug(u'***    Clearing    out    not_ready    list    ***')
        ti_status.not_ready.clear()
        for task in self.dag.topological_sort():
            for (key, ti) in list(ti_status.to_run.items()):
                if (task.task_id != ti.task_id):
                    continue
                ti.refresh_from_db()
                task = self.dag.get_task(ti.task_id)
                ti.task = task
                ignore_depends_on_past = (self.ignore_first_depends_on_past and (ti.execution_date == (start_date or ti.start_date)))
                self.logger.debug(u'Task    instance    to    run    {}    state    {}'.format(ti, ti.state))
                if (ti.state == State.NONE):
                    self.logger.warning(u'FIXME:    task    instance    {}    state    was    set    to    None    externally.    This    should    not    happen')
                    ti.set_state(State.SCHEDULED, session=session)
                if (ti.state == State.SUCCESS):
                    ti_status.succeeded.add(key)
                    self.logger.debug(u"Task    instance    {}    succeeded.    Don't    rerun.".format(ti))
                    ti_status.to_run.pop(key)
                    if (key in ti_status.started):
                        ti_status.started.pop(key)
                    continue
                elif (ti.state == State.SKIPPED):
                    ti_status.skipped.add(key)
                    self.logger.debug(u"Task    instance    {}    skipped.    Don't    rerun.".format(ti))
                    ti_status.to_run.pop(key)
                    if (key in ti_status.started):
                        ti_status.started.pop(key)
                    continue
                elif (ti.state == State.FAILED):
                    self.logger.error(u'Task    instance    {}    failed'.format(ti))
                    ti_status.failed.add(key)
                    ti_status.to_run.pop(key)
                    if (key in ti_status.started):
                        ti_status.started.pop(key)
                    continue
                elif (ti.state == State.UPSTREAM_FAILED):
                    self.logger.error(u'Task    instance    {}    upstream    failed'.format(ti))
                    ti_status.failed.add(key)
                    ti_status.to_run.pop(key)
                    if (key in ti_status.started):
                        ti_status.started.pop(key)
                    continue
                backfill_context = DepContext(deps=RUN_DEPS, ignore_depends_on_past=ignore_depends_on_past, ignore_task_deps=self.ignore_task_deps, flag_upstream_failed=True)
                if ti.are_dependencies_met(dep_context=backfill_context, session=session, verbose=True):
                    ti.refresh_from_db(lock_for_update=True, session=session)
                    if ((ti.state == State.SCHEDULED) or (ti.state == State.UP_FOR_RETRY)):
                        if executor.has_task(ti):
                            self.logger.debug(u'Task    Instance    {}    already    in    executor    waiting    for    queue    to    clear'.format(ti))
                        else:
                            self.logger.debug(u'Sending    {}    to    executor'.format(ti))
                            ti.state = State.QUEUED
                            session.merge(ti)
                            executor.queue_task_instance(ti, mark_success=self.mark_success, pickle_id=pickle_id, ignore_task_deps=self.ignore_task_deps, ignore_depends_on_past=ignore_depends_on_past, pool=self.pool)
                            ti_status.started[key] = ti
                            ti_status.to_run.pop(key)
                    session.commit()
                    continue
                if (ti.state == State.UPSTREAM_FAILED):
                    self.logger.error(u'Task    instance    {}    upstream    failed'.format(ti))
                    ti_status.failed.add(key)
                    ti_status.to_run.pop(key)
                    if (key in ti_status.started):
                        ti_status.started.pop(key)
                    continue
                if (ti.state == State.UP_FOR_RETRY):
                    self.logger.debug(u'Task    instance    {}    retry    period    not    expired    yet'.format(ti))
                    if (key in ti_status.started):
                        ti_status.started.pop(key)
                    ti_status.to_run[key] = ti
                    continue
                self.logger.debug(u'Adding    {}    to    not_ready'.format(ti))
                ti_status.not_ready.add(key)
        self.heartbeat()
        executor.heartbeat()
        if (ti_status.not_ready and (ti_status.not_ready == set(ti_status.to_run)) and (len(ti_status.started) == 0)):
            self.logger.warning(u'Deadlock    discovered    for    ti_status.to_run={}'.format(ti_status.to_run.values()))
            ti_status.deadlocked.update(ti_status.to_run.values())
            ti_status.to_run.clear()
        self._manage_executor_state(ti_status.started)
        self._update_counters(ti_status=ti_status)
        _dag_runs = ti_status.active_runs[:]
        for run in _dag_runs:
            run.update_state(session=session)
            if (run.state in State.finished()):
                ti_status.finished_runs += 1
                ti_status.active_runs.remove(run)
                executed_run_dates.append(run.execution_date)
            if run.dag.is_paused:
                models.DagStat.update([run.dag_id], session=session)
        self._log_progress(ti_status)
    return executed_run_dates
