This is a work in progress document to have some ideas around event based scheduling

Introduction

The current scheduler polls every dagrun for runnable tasks. If a task has not run yet the scheduler will ask the task if it is runnable (ti.is_runnable) which calls ti.are_dependencies_met. For a single task this costs around 0.05-015s. Thus for many tasks in a dag this grows quite quickly and this is the reason why you see increasing scheduling times ("Loop time took xxxx"). The code for are_dependencies_met looks currently like this:

@provide_session
def are_dependencies_met(
        self,
        session=None,
        flag_upstream_failed=False,
        ignore_depends_on_past=False,
        verbose=False):
    """
    Returns a boolean on whether the upstream tasks are in a SUCCESS state
    and considers depends_on_past and the previous run's state.

    :param flag_upstream_failed: This is a hack to generate
        the upstream_failed state creation while checking to see
        whether the task instance is runnable. It was the shortest
        path to add the feature
    :type flag_upstream_failed: boolean
    :param ignore_depends_on_past: if True, ignores depends_on_past
        dependencies. Defaults to False.
    :type ignore_depends_on_past: boolean
    :param verbose: verbose provides more logging in the case where the
        task instance is evaluated as a check right before being executed.
        In the case of the scheduler evaluating the dependencies, this
        logging would be way too verbose.
    :type verbose: boolean
    """
    TI = TaskInstance
    TR = TriggerRule

    task = self.task

    # Checking that the depends_on_past is fulfilled
    if (task.depends_on_past and not ignore_depends_on_past and
            not self.execution_date == task.start_date):
        previous_ti = session.query(TI).filter(
            TI.dag_id == self.dag_id,
            TI.task_id == task.task_id,
            TI.execution_date ==
                self.task.dag.previous_schedule(self.execution_date),
            TI.state.in_({State.SUCCESS, State.SKIPPED}),
        ).first()
        if not previous_ti:
            if verbose:
                logging.warning("depends_on_past not satisfied")
            return False

        # Applying wait_for_downstream
        previous_ti.task = self.task
        if task.wait_for_downstream and not \
                previous_ti.are_dependents_done(session=session):
            if verbose:
                logging.warning("wait_for_downstream not satisfied")
            return False

    # Checking that all upstream dependencies have succeeded
    if not task.upstream_list or task.trigger_rule == TR.DUMMY:
        return True

    # todo: this query becomes quite expensive with dags that have
    # many tasks. It should be refactored to let the task report
    # to the dag run and get the aggregates from there
    qry = (
        session
        .query(
            func.coalesce(func.sum(
                case([(TI.state == State.SUCCESS, 1)], else_=0)), 0),
            func.coalesce(func.sum(
                case([(TI.state == State.SKIPPED, 1)], else_=0)), 0),
            func.coalesce(func.sum(
                case([(TI.state == State.FAILED, 1)], else_=0)), 0),
            func.coalesce(func.sum(
                case([(TI.state == State.UPSTREAM_FAILED, 1)], else_=0)), 0),
            func.count(TI.task_id),
        )
        .filter(
            TI.dag_id == self.dag_id,
            TI.task_id.in_(task.upstream_task_ids),
            TI.execution_date == self.execution_date,
            TI.state.in_([
                State.SUCCESS, State.FAILED,
                State.UPSTREAM_FAILED, State.SKIPPED]),
        )
    )

    successes, skipped, failed, upstream_failed, done = qry.first()

    satisfied = self.evaluate_trigger_rule(
        session=session, successes=successes, skipped=skipped,
        failed=failed, upstream_failed=upstream_failed, done=done,
        flag_upstream_failed=flag_upstream_failed)

    if verbose and not satisfied:
        logging.warning("Trigger rule `{}` not satisfied".format(task.trigger_rule))
    return satisfied

 

The part that takes most of the time are the agg

Scheduling times can be reduced by implementing two changes. Firstly, in update_state a dagrun is checked if it is deadlocked or finished. It also call ti.are_dependencies_met for every task. However if a task reports to its downstream siblings its state changes