Dynamically adjusting priority via Futures

[jakirkham]

Basically copied from this comment in PR ( #1651 ). Also related to issue ( dask/dask#2860 ).

One thing that would be nice is to be able to propagate priority changes from Futures themselves. Namely if one tries to call result, it would be good to bump the priority. Maybe this could be configurable if not generally desirable.

To provide an example use case, imagine one has a Dask object (Array, Dataframe, etc.) and has called persist on the object. Then one wants to inspect a small piece of it while the computation proceeds. So they select out that piece and call compute. It would be good if this already was interpreted as the user wanting this piece of the result sooner than the rest. That way the user can play with this piece as the rest of the computation completes. If needed, this process can be repeated with other pieces.

cc @mrocklin @p-himik

[mrocklin]

There are a few things that would have to happen here:

1. Update the priority on the scheduler. This is relatively straightforward. Priorities are usually set in Scheduler.update_graph. Whether or not we want to reuse this function in this case is an open question.
2. Propagate this priority change to all dependencies of this task still waiting or processing
3. For each of these tasks that are currently processing, bump their priority on the worker. This means changing the Worker.priority value for the key as well as possibly re-inserting them into the ready heap with the new priority if they are already there (I think having duplicate values there is probably safe)
4. Decide what we want our policy to be on the client, if Client.gather should or should not send a signal to reprioritize by default.

[jakirkham]

cc @philippjfr (as this may be of interest given recent Dask support in HoloViews)

[mrocklin]

A small snippet of recursing through priorities of tasks on the scheduler. This came up when discussing this topic with @jakrikham

    def bump_priority(ts, priority):
        if ts.status in ('memory', 'released', 'erred', ...):
            return

        if priority < ts.priority[0]:
            raise ...

        ts.priority[0] = priority
        for dep in ts.dependencies:
            self.bump_priority(dep, priority)

            if ts.status == 'processing':
                self.send_to_worker(ts.processing_on, {'op': 'bump-priority',
                                                       'priority': ts.priority})

[SultanOrazbayev]

I'm trying to manually adjust the priorities for futures (e.g. for tree reduction I want to assign higher priority to tasks that are closer to the final node), but I don't know how to find out the priority of a given future. From the snippet above it looks like it was possible to get priority as a property of future, but it doesn't work in 2021.05.1 version.

This is my code:

from dask.distributed import Client
client = Client()

def sq(x):
    return x**2

futs = client.map(sq, range(3))

print(futs[0].priority)
# AttributeError: 'Future' object has no attribute 'priority'

[mrocklin]

Those are TaskState objects and are internal to the scheduler. There is no mechanism today to adjust priorities dynamically. This is what this issue is proposing.