Repository: dask/dask
File: dask/order.py (Python)

Code before:
            % "\n  -> ".join(str(x) for x in cycle)
        )

    leaf_nodes = {k for k, v in dependents.items() if not v}
    root_nodes = {k for k, v in dependencies.items() if not v}

    result: dict[Key, Order | int] = {}

    # Normalize the graph by removing leaf nodes that are not actual tasks, see
    # for instance da.store where the task is merely an alias
    # to multiple keys, i.e. [key1, key2, ...,]
    all_tasks = False
    j = 0
    while not all_tasks:
        all_tasks = True
        for leaf in list(leaf_nodes):
            if (
                not istask(dsk[leaf])
                # Having a linear chain is fine
                and len(dependencies[leaf]) > 1
            ):
                all_tasks = False
                if return_stats:
                    result[leaf] = Order(len(dsk) - 1 - j, -1)
                else:
                    result[leaf] = len(dsk) - 1 - j
                j += 1
                leaf_nodes.remove(leaf)
                del dsk[leaf]
                del dependents[leaf]
                for dep in dependencies[leaf]:
                    dependents[dep].remove(leaf)
                    if not dependents[dep]:
                        leaf_nodes.add(dep)

    assert dependencies is not None
    roots_connected, max_dependents = _connecting_to_roots(dependencies, dependents)
    leafs_connected, _ = _connecting_to_roots(dependents, dependencies)
    i = 0

    runnable_hull = set()
    reachable_hull = set()

    runnable = []

    known_runnable_paths: dict[Key, list[list[Key]]] = {}
    known_runnable_paths_pop = known_runnable_paths.pop

    crit_path_counter = 0
    scrit_path: set[Key] = set()
    _crit_path_counter_offset: int | float = 0

Review comment (suggestion):
```suggestion
                # Put non-tasks after all tasks
                if return_stats:
```

Code after:
            % "\n  -> ".join(str(x) for x in cycle)
        )

    leaf_nodes = {k for k, v in dependents.items() if not v}
    root_nodes = {k for k, v in dependencies.items() if not v}

    result: dict[Key, Order | int] = {}

    # Normalize the graph by removing leaf nodes that are not actual tasks, see
    # for instance da.store where the task is merely an alias
    # to multiple keys, i.e. [key1, key2, ...,]
    all_tasks = False
    n_removed_leaves = 0
    while not all_tasks:
        all_tasks = True
        for leaf in list(leaf_nodes):
            if (
                not istask(dsk[leaf])
                # Having a linear chain is fine
                and len(dependencies[leaf]) > 1
            ):
                all_tasks = False
                # Put non-tasks at the very end since they are merely aliases
                # and have no impact on performance at all
                prio = len(dsk) - 1 - n_removed_leaves
                if return_stats:
                    result[leaf] = Order(prio, -1)
                else:
                    result[leaf] = prio
                n_removed_leaves += 1
                leaf_nodes.remove(leaf)
                del dsk[leaf]
                del dependents[leaf]
                for dep in dependencies[leaf]:
                    dependents[dep].remove(leaf)
                    if not dependents[dep]:
                        leaf_nodes.add(dep)

    assert dependencies is not None
    roots_connected, max_dependents = _connecting_to_roots(dependencies, dependents)
    leafs_connected, _ = _connecting_to_roots(dependents, dependencies)
    i = 0

    runnable_hull = set()
    reachable_hull = set()

    runnable = []

    known_runnable_paths: dict[Key, list[list[Key]]] = {}
    known_runnable_paths_pop = known_runnable_paths.pop