'For all DAG IDs in the SimpleDagBag, look for task instances in the
old_states and set them to new_state if the corresponding DagRun
exists but is not in the running state. This normally should not
happen, but it can if the state of DagRuns are changed manually.
:param old_states: examine TaskInstances in this state
:type old_state: list[State]
:param new_state: set TaskInstances to this state
:type new_state: State
:param simple_dag_bag: TaskInstances associated with DAGs in the
simple_dag_bag and with states in the old_state will be examined
:type simple_dag_bag: SimpleDagBag'

@provide_session
def _change_state_for_tis_without_dagrun(self, simple_dag_bag, old_states, new_state, session=None):

    tis_changed = 0
    if self.using_sqlite:
        tis_to_change = session.query(models.TaskInstance).filter(models.TaskInstance.dag_id.in_(simple_dag_bag.dag_ids)).filter(models.TaskInstance.state.in_(old_states)).filter(and_((models.DagRun.dag_id == models.TaskInstance.dag_id), (models.DagRun.execution_date == models.TaskInstance.execution_date), (models.DagRun.state != State.RUNNING))).with_for_update().all()
        for ti in tis_to_change:
            ti.set_state(new_state, session=session)
            tis_changed += 1
    else:
        tis_changed = session.query(models.TaskInstance).filter(models.TaskInstance.dag_id.in_(simple_dag_bag.dag_ids)).filter(models.TaskInstance.state.in_(old_states)).filter(and_((models.DagRun.dag_id == models.TaskInstance.dag_id), (models.DagRun.execution_date == models.TaskInstance.execution_date), (models.DagRun.state != State.RUNNING))).update({models.TaskInstance.state: new_state}, synchronize_session=False)
        session.commit()
    if (tis_changed > 0):
        self.logger.warning(u'Set    {}    task    instances    to    state={}    as    their    associated    DagRun    was    not    in    RUNNING    state'.format(tis_changed, new_state))
