Accessing dispatch_id and node_id in Executor.run()

[cjao]

With the recent revision to the Executor interface, we no longer expose the original Executor.execute() method to the public and instead require custom executors to override a simplified run() method:

@abstractmethod
def run(function: Callable, args: List, kwargs: Dict):

While this appears cleaner than the full execute() method, it provides no way to access dispatch_id or node_id. This is a problem for many executors:

• For remote executors, one is required to ship the function to be run and its associated data to the remote machine. It is very natural (not to mention useful for tracing the execution cycle if things don't work as expected) to use a naming scheme for the remote files derived from dispatch_id and node_id.
• Even for local executors like Dask, having full identifying information about the task in run() would allow more readable logging (the current logging statements in Dask are pretty generic, offering no visibility into exactly which task is being handled at any given time).

Presently, the only way to access this information is to override execute() directly, which is precisely what the introduction of run() was trying to discourage.

To make matters worse, as execute() grows more sophisticated logic to manage the executor's lifecycle (e.g. provisioning and tearing down resources, monitoring remote files), it wouldn't even be enough for executor subclasses to simply invoke super().execute() in their custom execute() implementations. They would need to duplicate all of the lifecyle handling code.

How to expose this information to run?

1. Passing them as arguments to run():

@abstractmethod
def run(function: Callable, args: List, kwargs: Dict, dispatch_id: str, node_id: int):

2. Assigning dispatch_id and node_id to executor plugin instance attributes in execute()
3. ...?

The first approach passes the data to run() explicitly, and I think being explicit generally enhances code readability since there are fewer surprises.

The second approach is taken by the SSH and Slurm plugins. However it will cease to work in the future because we are planning to allow sharing executor instances (in fact their underlying hardware resources) between multiple tasks. I also considered storing the information in context-local variables, but unfortunately ContextVars are not picklable.

[cjao]

@santoshkumarradha @wjcunningham7

[venkatBala]

@cjao I agree and I am having to deal with this short coming as I am implementing the AWSLambda executor. Given the current version where dispatch and node id are only being passed in via the execute method, I am having to overwrite that in my executor class and pass those into the setup and teardown methods.

To add to this discussion, the UX can be enhanced further if the dispatch and node ids are passed into the setup/teardown(cleanup) methods as well. This will allow creation of resources (be it cloud, or even local (say shell scripts, pickled objects)) that have the dispatch and node id as part of their name or tags. This way resources can be uniquely tagged with the dispatch and node ids. Passing those values to the teardown/cleanup methods will allow removal of those resources (if needed) as well. the mapping of these actions would be one to one

@cjao thoughts?

[santoshkumarradha]

quick questions, Ideally both dispatchid and nodeid are something that is not required by the executer but is just optionally used by us (@venkatBala in theory we dont need to use the same dispatchid or nodeid to save/store pickle files or anything in executer as it just needs to be unique). But I do agree that it makes the our life easier in debugging the if the executer has the same metadata as to that of what is being used by our dispatcher.

With this in mind, @cjao / @venkatBala can you guys tell me why

1. setting the nodeid / dispatchid right before sending off as an attribute does not work ?
2. If we are going the route of exposing it directly into the run function (ideally not a big fan as it is polluting the run namespace with variables that is not needed by end user who is defining the executer plugins), I vote for a more robust solution of passing in a dictionary that contains all meta-data to make it future proof.

(Not a bit fan of 2 as we want the creation of executers to be as simple as possible, even if it involves extra work for us.)

[cjao]

Hi @santoshkumarradha ,

1. The executor instance attribute approach works right now because a fresh instance is created for each task. However, it breaks down when multiple concurrent tasks start sharing an instance because a) each task may refer to the instance attributes multiple times during execution, and b) the execution steps by different tasks may be interleaved.

To illustrate, a typical remote executor such as Slurm performs the following steps to run a task and retrieve its output.

1. Serialize and upload the function and input to the remote machine
2. Upload a small Python program to deserialize the data, run the function, and save the results and outputs.
3. Invoke the remote Python script via ssh.
4. Download the completed output files.
5. Unpack the output files and return the output to the dispatcher.

Omitted for brevity (but important for debugging) are logging statements throughout the entire process.

Each of these steps may refer to dispatch_id and node_id to compute the various function filenames and for logging.

Now suppose these are accessed via self.dispatch_id and self.node_id. If two tasks A and B are running concurrently in the same executor instance, the dispatcher event loop will typically interleave their steps like

A1, B1, A2, A3, B2, ...

If task A sets self.node_id right before step A1, that data will be corrupted from task A's point of view by step B1. Thus when step A2 refers to self.node_id, it will end up retrieving the value set by task B. The situation gets even worse in a multithreaded setting (for non-async executors) since tasks A and B could be preempted at any time, even in the middle of a step.

2. I agree that a dictionary like task_metadata would be more future-proof. That said,

it is polluting the run namespace with variables that is not needed by end user who is defining the executer plugins

the characterization "not needed by end users" may be true in theory, but at least in practice, practically all the remote executor plugins so far (both public and private) use file naming scheme based on dispatch_id and node_id. Debugging would be much more difficult for executor plugin authors if a completely random naming system were used, not to mention the implications on logging. As mentioned in the OP, even the Dask logging would be more informative if the messages were keyed by dispatch_id and node_id.

[santoshkumarradha]

✅ Lets go ahead with task_metadata dict.

[santoshkumarradha]

Implemented as task_metadata in #951