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

Code before:
                            if (
                                len(dependencies[current]) > 1
                                and num_needed[current] <= 1
                            ):
                                for k in path:
                                    add_to_result(k)
                            else:
                                known_runnable_paths[current] = [path]
                                runnable_hull.discard(current)
                        break

    # Pick strategy
    # Note: We're trying to be smart here by picking a strategy on how to
    # determine the critical path. This is not always clear and we may want to
    # consider just calculating both orderings and picking the one with less
    # pressure. The only concern to this would be performance but at time of
    # writing, the most expensive part of ordering is the prep work (mostly
    # connected roots + sort_key) which can be reused for multiple orderings.

    def get_target(longest_path: bool = False) -> Key:
        # Some topologies benefit if the node with the most dependencies
        # is used as first choice, others benefit from the opposite.
        candidates = leaf_nodes
        skey: Callable = sort_key
        if reachable_hull:
            skey = lambda k: (num_needed[k], sort_key(k))

        all_leafs_accessible = len(leafs_connected_to_loaded_roots) >= len(candidates)
        if reachable_hull and not all_leafs_accessible:
            # Avoid this branch if we can already access all leafs. Just pick
            # one of them without the expensive selection process in this case.

            candidates = reachable_hull & candidates
            if not candidates:
                # We can't reach a leaf node directly, but we still have nodes
                # with results in memory, these notes can inform our path towards
                # a new preferred leaf node.
                for r in processed_roots:
                    leafs_connected_to_loaded_roots.update(leafs_connected[r])
                processed_roots.clear()

                leafs_connected_to_loaded_roots.intersection_update(leaf_nodes)
                candidates = leafs_connected_to_loaded_roots
            else:
                leafs_connected_to_loaded_roots.update(candidates)

        elif not reachable_hull:
            # We reach a new and independent branch, so throw away previous branch
            leafs_connected_to_loaded_roots.clear()

        if longest_path and (not reachable_hull or all_leafs_accessible):

Review comment (bug):
This was actually wrong logic and was basically "overfitting" the tests. We're trying to pick final nodes / leaf nodes here and `num_needed` is only a relevant quantity iff leaves connect directly to roots or if leafs are in the runnable hull after the first pass (i.e. we touched one of the dependencies of the leaf).
So in most practical examples, `num_needed` would be identical to `dependencies` which is only on second position of `sort_key` (because `total_dependencies` is on place one, i.e. a global measure is stronger than a local one).

So in most realistic graphs where this logic was triggered, it made bad choices. One unit test where this is making now a better choice is `test_run_smaller_sections`.

An example for the "overfitting" with roots that connect directly to leafs is `test_diagnostics` (e.g. remove this line from main and see which tests break)

Code after:
                                    if nsplits > 1:
                                        path = []
                                        for pruned in pruned_branches:
                                            path.extend(pruned)
                                        branches.append((nsplits - 1, path))
                                        break

                                    while pruned_branches:
                                        path = pruned_branches.popleft()
                                        for k in path:
                                            if num_needed[k]:
                                                pruned_branches.append(path)
                                                break
                                            add_to_result(k)
                        else:
                            if (
                                len(dependencies[current]) > 1
                                and num_needed[current] <= 1
                            ):
                                for k in path:
                                    add_to_result(k)
                            else:
                                known_runnable_paths[current] = [path]
                                runnable_hull.discard(current)
                        break

    # Pick strategy
    # Note: We're trying to be smart here by picking a strategy on how to
    # determine the critical path. This is not always clear and we may want to
    # consider just calculating both orderings and picking the one with less
    # pressure. The only concern to this would be performance but at time of
    # writing, the most expensive part of ordering is the prep work (mostly
    # connected roots + sort_key) which can be reused for multiple orderings.

    # Degree in this context is the number root nodes that have to be loaded for
    # this leaf to become accessible. Zero means the leaf is already accessible
    # in which case it _should_ either already be in result or be accessible vie
    # process_runnables
    # When picking a new target, we prefer the leafs with the least number of
    # roots that need loading.
    leafs_degree = {}
    leafs_per_degree = defaultdict(set)
    min_leaf_degree = len(dsk)
    max_leaf_degree = len(dsk)
    for leaf in leaf_nodes - root_nodes:
        degree = len(roots_connected[leaf])
        min_leaf_degree = min(min_leaf_degree, degree)
        max_leaf_degree = max(max_leaf_degree, degree)
        leafs_degree[leaf] = degree
        leafs_per_degree[degree].add(leaf)