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

Code before:

    dsk = dict(dsk)
    expected_len = len(dsk)

    if dependencies is None:
        dependencies = {k: get_dependencies(dsk, k) for k in dsk}

    dependents = reverse_dict(dependencies)
    num_needed, total_dependencies = ndependencies(dependencies, dependents)
    if len(total_dependencies) != len(dsk):
        cycle = getcycle(dsk, None)
        raise RuntimeError(
            "Cycle detected between the following keys:\n  -> %s"
            % "\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)

Review comment (suggestion):
nit: `j` could have a more meaningful name like
```suggestion
    n_removed_leaves = 0
```

Code after:

    dsk = dict(dsk)
    expected_len = len(dsk)

    if dependencies is None:
        dependencies = {k: get_dependencies(dsk, k) for k in dsk}

    dependents = reverse_dict(dependencies)
    num_needed, total_dependencies = ndependencies(dependencies, dependents)
    if len(total_dependencies) != len(dsk):
        cycle = getcycle(dsk, None)
        raise RuntimeError(
            "Cycle detected between the following keys:\n  -> %s"
            % "\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)