@provide_session<EOL><INDENT>def _find_executable_task_instances(self, simple_dag_bag, states, session=None):<DEDENT>
executable_tis = []<EOL>TI = models.TaskInstance<EOL>DR = models.DagRun<EOL>DM = models.DagModel<EOL>ti_query = (<EOL>session<EOL>.query(TI)<EOL>.filter(TI.dag_id.in_(simple_dag_bag.dag_ids))<EOL>.outerjoin(<EOL>DR,<EOL>and_(DR.dag_id == TI.dag_id, DR.execution_date == TI.execution_date)<EOL>)<EOL>.filter(or_(DR.run_id == None,  <EOL>not_(DR.run_id.like(BackfillJob.ID_PREFIX + '<STR_LIT:%>'))))<EOL>.outerjoin(DM, DM.dag_id == TI.dag_id)<EOL>.filter(or_(DM.dag_id == None,  <EOL>not_(DM.is_paused)))<EOL>)<EOL>if None in states:<EOL><INDENT>ti_query = ti_query.filter(<EOL>or_(TI.state == None, TI.state.in_(states))  <EOL>)<EOL><DEDENT>else:<EOL><INDENT>ti_query = ti_query.filter(TI.state.in_(states))<EOL><DEDENT>task_instances_to_examine = ti_query.all()<EOL>if len(task_instances_to_examine) == <NUM_LIT:0>:<EOL><INDENT>self.log.debug("<STR_LIT>")<EOL>return executable_tis<EOL><DEDENT>task_instance_str = "<STR_LIT>".join(<EOL>[repr(x) for x in task_instances_to_examine])<EOL>self.log.info(<EOL>"<STR_LIT>", len(task_instances_to_examine),<EOL>task_instance_str<EOL>)<EOL>pools = {p.pool: p for p in session.query(models.Pool).all()}<EOL>pool_to_task_instances = defaultdict(list)<EOL>for task_instance in task_instances_to_examine:<EOL><INDENT>pool_to_task_instances[task_instance.pool].append(task_instance)<EOL><DEDENT>states_to_count_as_running = [State.RUNNING, State.QUEUED]<EOL>dag_concurrency_map, task_concurrency_map = self.__get_concurrency_maps(<EOL>states=states_to_count_as_running, session=session)<EOL>for pool, task_instances in pool_to_task_instances.items():<EOL><INDENT>pool_name = pool<EOL>if not pool:<EOL><INDENT>open_slots = models.Pool.default_pool_open_slots()<EOL>pool_name = models.Pool.default_pool_name<EOL><DEDENT>else:<EOL><INDENT>if pool not in pools:<EOL><INDENT>self.log.warning(<EOL>"<STR_LIT>",<EOL>pool<EOL>)<EOL>open_slots = <NUM_LIT:0><EOL><DEDENT>else:<EOL><INDENT>open_slots = pools[pool].open_slots(session=session)<EOL><DEDENT><DEDENT>num_ready = len(task_instances)<EOL>self.log.info(<EOL>"<STR_LIT>"<EOL>"<STR_LIT>",<EOL>pool, open_slots, num_ready<EOL>)<EOL>priority_sorted_task_instances = sorted(<EOL>task_instances, key=lambda ti: (-ti.priority_weight, ti.execution_date))<EOL>num_starving_tasks = <NUM_LIT:0><EOL>for current_index, task_instance in enumerate(priority_sorted_task_instances):<EOL><INDENT>if open_slots <= <NUM_LIT:0>:<EOL><INDENT>self.log.info(<EOL>"<STR_LIT>",<EOL>open_slots, pool<EOL>)<EOL>num_starving_tasks = len(priority_sorted_task_instances) - current_index<EOL>break<EOL><DEDENT>dag_id = task_instance.dag_id<EOL>simple_dag = simple_dag_bag.get_dag(dag_id)<EOL>current_dag_concurrency = dag_concurrency_map[dag_id]<EOL>dag_concurrency_limit = simple_dag_bag.get_dag(dag_id).concurrency<EOL>self.log.info(<EOL>"<STR_LIT>",<EOL>dag_id, current_dag_concurrency, dag_concurrency_limit<EOL>)<EOL>if current_dag_concurrency >= dag_concurrency_limit:<EOL><INDENT>self.log.info(<EOL>"<STR_LIT>"<EOL>"<STR_LIT>",<EOL>task_instance, dag_id, dag_concurrency_limit<EOL>)<EOL>continue<EOL><DEDENT>task_concurrency_limit = simple_dag.get_task_special_arg(<EOL>task_instance.task_id,<EOL>'<STR_LIT>')<EOL>if task_concurrency_limit is not None:<EOL><INDENT>current_task_concurrency = task_concurrency_map[<EOL>(task_instance.dag_id, task_instance.task_id)<EOL>]<EOL>if current_task_concurrency >= task_concurrency_limit:<EOL><INDENT>self.log.info("<STR_LIT>"<EOL>"<STR_LIT>", task_instance)<EOL>continue<EOL><DEDENT><DEDENT>if self.executor.has_task(task_instance):<EOL><INDENT>self.log.debug(<EOL>"<STR_LIT>",<EOL>task_instance.key<EOL>)<EOL>continue<EOL><DEDENT>executable_tis.append(task_instance)<EOL>open_slots -= <NUM_LIT:1><EOL>dag_concurrency_map[dag_id] += <NUM_LIT:1><EOL>task_concurrency_map[(task_instance.dag_id, task_instance.task_id)] += <NUM_LIT:1><EOL><DEDENT>Stats.gauge('<STR_LIT>'.format(pool_name=pool_name),<EOL>num_starving_tasks)<EOL><DEDENT>task_instance_str = "<STR_LIT>".join(<EOL>[repr(x) for x in executable_tis])<EOL>self.log.info(<EOL>"<STR_LIT>", task_instance_str)<EOL>for ti in executable_tis:<EOL><INDENT>copy_dag_id = ti.dag_id<EOL>copy_execution_date = ti.execution_date<EOL>copy_task_id = ti.task_id<EOL>make_transient(ti)<EOL>ti.dag_id = copy_dag_id<EOL>ti.execution_date = copy_execution_date<EOL>ti.task_id = copy_task_id<EOL><DEDENT>return executable_tis<EOL>
Finds TIs that are ready for execution with respect to pool limits,
dag concurrency, executor state, and priority.

: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: airflow.utils.dag_processing.SimpleDagBag
:param executor: the executor that runs task instances
:type executor: BaseExecutor
:param states: Execute TaskInstances in these states
:type states: tuple[airflow.utils.state.State]
:return: list[airflow.models.TaskInstance]