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

Code before:
        # is used as first choice, others benefit from the opposite.
        candidates = leaf_nodes
        skey: Callable = sort_key

        preferred_candidates = set()
        if reachable_hull:
            skey = lambda k: (num_needed[k], sort_key(k))
            preferred_candidates = reachable_hull & candidates

        if reachable_hull:
            # 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.
            if not preferred_candidates:
                for r in processed_roots:
                    leafs_connected_to_loaded_roots.update(leafs_connected[r])
                processed_roots.clear()

                leafs_connected_to_loaded_roots.intersection_update(candidates)
                preferred_candidates = leafs_connected_to_loaded_roots
            candidates = preferred_candidates
        else:
            # We reach a new and independent branch, so throw away previous branch
            leafs_connected_to_loaded_roots.clear()

        if longest_path and not runnable_hull and not reachable_hull:
            return leaf_nodes_sorted.pop()
        else:
            # FIXME: This can be very expensive
            return min(candidates, key=skey)

    def use_longest_path() -> bool:
        size = 0
        # Heavy reducer / splitter topologies often benefit from a very
        # traditional critical path that expresses the longest chain of
        # tasks.
        if abs(len(root_nodes) - len(leaf_nodes)) / len(root_nodes) < 0.8:
            # If the graph stays about the same, we are checking for symmetry
            # and choose a "quickest path first" approach if the graph appears
            # to be asymmetrical
            for r in root_nodes:
                if not size:
                    size = len(leafs_connected[r])
                elif size != len(leafs_connected[r]):
                    return False

        return True

    longest_path = use_longest_path()
    leaf_nodes_sorted = []
    if longest_path:

Review comment (suggestion):

```suggestion
        if longest_path and not reachable_hull:
```

Code after:
        # 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):
            return leaf_nodes_sorted.pop()
        else:
            # FIXME: This can be very expensive
            return min(candidates, key=skey)

    def use_longest_path() -> bool:
        size = 0
        # Heavy reducer / splitter topologies often benefit from a very
        # traditional critical path that expresses the longest chain of
        # tasks.
        if abs(len(root_nodes) - len(leaf_nodes)) / len(root_nodes) < 0.8:
            # If the graph stays about the same, we are checking for symmetry
            # and choose a "quickest path first" approach if the graph appears
            # to be asymmetrical
            for r in root_nodes:
                if not size:
                    size = len(leafs_connected[r])
                elif size != len(leafs_connected[r]):
                    return False

        return True