'This method schedules the tasks for a single DAG by looking at the
active DAG runs and adding task instances that should run to the
queue.'

def _process_task_instances(self, dag, queue):

    session = settings.Session()
    dag_runs = DagRun.find(dag_id=dag.dag_id, state=State.RUNNING, session=session)
    active_dag_runs = []
    for run in dag_runs:
        self.logger.info(u'Examining    DAG    run    {}'.format(run))
        if (run.execution_date > datetime.now()):
            self.logger.error(u'Execution    date    is    in    future:    {}'.format(run.execution_date))
            continue
        if (len(active_dag_runs) >= dag.max_active_runs):
            self.logger.info(u'Active    dag    runs    >    max_active_run.')
            continue
        if run.is_backfill:
            continue
        run.dag = dag
        run.verify_integrity(session=session)
        run.update_state(session=session)
        if (run.state == State.RUNNING):
            make_transient(run)
            active_dag_runs.append(run)
    for run in active_dag_runs:
        self.logger.debug(u'Examining    active    DAG    run    {}'.format(run))
        tis = run.get_task_instances(state=(State.NONE, State.UP_FOR_RETRY))
        for ti in tis:
            task = dag.get_task(ti.task_id)
            ti.task = task
            if task.adhoc:
                continue
            if ti.are_dependencies_met(dep_context=DepContext(flag_upstream_failed=True), session=session):
                self.logger.debug(u'Queuing    task:    {}'.format(ti))
                queue.append(ti.key)
    session.close()
