Is there a way to rerun a successfully completed Future?

[aa1371]

I see that client.recreate_error_locally already exists. But, I don't see a publicly exposed way to do the same thing for successfully completed futures, does this already exist? This can be useful if a task completes successfully, but produces some unexpected results that you want to debug.

[fjetter]

afaik, there is no standard API to do this but there is enough to put a recipe together and/or contribute this as a feature. For non-erred tasks I would argue, the logic is almost identical to the replay exception case. What the recreate_error_locally call does is to fetch the task runspec (i.e. the function + arguments) and calculates all the dependencies of that task.

This happens here

distributed/distributed/recreate_exceptions.py

Lines 125 to 131 in 5e150aa

	out = await self._get_futures_error(future)
	function, args, kwargs, deps = out
	futures = self.client._graph_to_futures({}, deps)
	data = await self.client._gather(futures)
	args = pack_data(args, data)
	kwargs = pack_data(kwargs, data)
	return (function, args, kwargs)

For a non-erred task, the only thing we'll need to change is the cause_of_failure (in _get_futures_error) which retrieves information from the scheduler. This could look like

def get_runspec(dask_scheduler, key):
    ts = dask_scheduler.tasks[key]
    return {
        "task": ts.run_spec,
        "deps": [dts.key for dts in ts.dependencies]
    }
client.run_on_scheduler(get_runspec)

then continue with

distributed/distributed/recreate_exceptions.py

Lines 86 to 92 in 5e150aa

	deps, task = out["deps"], out["task"]
	if isinstance(task, dict):
	function, args, kwargs = _deserialize(**task)
	return (function, args, kwargs, deps)
	else:
	function, args, kwargs = _deserialize(task=task)
	return (function, args, kwargs, deps)

I haven't tested this, therefore no guarantees but that should be roughly how it could look like. If you want to get your hands dirty, feel free to open a PR!

In a sense this would be a "calculate all dependencies on the cluster but the final task on my machine"-feature

[aa1371]

Hey, will take a look at this soon. Do you think it would make sense to consolidate this concept and the existing recreate_error_locally?

[aa1371]

@fjetter - opened a PR with an initial concept here:
#4813

[aa1371]

This now exists with client.recreate_task_locally