Function process_runnables

dask/order.py:298–404

Compute all currently runnable paths and either cache or execute them This is designed to ensure we are running tasks that are free to execute (e.g. the result of a splitter task) not too eagerly. If we executed such free tasks too early we'd be walking the graph in a too wi

()

    @_with_offset
    def process_runnables() -> None:
        """Compute all currently runnable paths and either cache or execute them

        This is designed to ensure we are running tasks that are free to execute
        (e.g. the result of a splitter task) not too eagerly. If we executed
        such free tasks too early we'd be walking the graph in a too wide /
        breadth first fashion that is not optimal. If instead we were to only
        execute them once they are needed for a final result, this can cause
        very high memory pressure since valuable reducers are executed too
        late.

        The strategy here is to take all runnable tasks and walk forwards until
        we hit a reducer node (i.e. a node with more than one dependency). We
        will remember/cache the path to this reducer node.
        If this path leads to a leaf or if we find enough runnable paths for a
        reducer to be runnable, we will execute the path.

        If instead of a reducer a splitter is encountered that is runnable, we
        will follow its splitter paths individually and apply the same logic to
        each branch.
        """
        while runnable:
            candidates = runnable.copy()
            runnable.clear()
            while candidates:
                key = candidates.pop()
                if key in runnable_hull or key in result:
                    continue
                if key in leaf_nodes:
                    add_to_result(key)
                    continue
                path = [key]
                branches = deque([(0, path)])

                while branches:
                    nsplits, path = branches.popleft()
                    while True:
                        # Loop invariant. Too expensive to compute at runtime
                        # assert not set(known_runnable_paths).intersection(runnable_hull)
                        current = path[-1]
                        runnable_hull.add(current)
                        deps_downstream = dependents[current]
                        deps_upstream = dependencies[current]
                        if not deps_downstream:
                            # FIXME: The fact that it is possible for
                            # num_needed[current] == 0 means we're doing some
                            # work twice
                            if num_needed[current] <= 1:
                                for k in path:
                                    add_to_result(k)
                            else:
                                runnable_hull.discard(current)
                        elif len(path) == 1 or len(deps_upstream) == 1:
                            if len(deps_downstream) > 1:
                                nsplits += 1
                                for d in sorted(deps_downstream, key=sort_key):
                                    # This ensures we're only considering splitters
                                    # that are genuinely splitting and not