'Attempts to execute TaskInstances that should be executed by the scheduler.
There are three steps:
1. Pick TIs by priority with the constraint that they are in the expected states
and that we do exceed max_active_runs or pool limits.
2. Change the state for the TIs above atomically.
3. Enqueue the TIs in the executor.
:param simple_dag_bag: TaskInstances associated with DAGs in the
simple_dag_bag will be fetched from the DB and executed
:type simple_dag_bag: SimpleDagBag
:param states: Execute TaskInstances in these states
:type states: Tuple[State]
:return: None'

@provide_session
def _execute_task_instances(self, simple_dag_bag, states, session=None):

    executable_tis = self._find_executable_task_instances(simple_dag_bag, states, session=session)
    if (self.max_tis_per_query == 0):
        tis_with_state_changed = self._change_state_for_executable_task_instances(executable_tis, states, session=session)
        self._enqueue_task_instances_with_queued_state(simple_dag_bag, tis_with_state_changed)
        session.commit()
        return len(tis_with_state_changed)
    else:
        chunks = [executable_tis[i:(i + self.max_tis_per_query)] for i in range(0, len(executable_tis), self.max_tis_per_query)]
        total_tis_queued = 0
        for chunk in chunks:
            tis_with_state_changed = self._change_state_for_executable_task_instances(chunk, states, session=session)
            self._enqueue_task_instances_with_queued_state(simple_dag_bag, tis_with_state_changed)
            session.commit()
            total_tis_queued += len(tis_with_state_changed)
        return total_tis_queued
