vine: function infile load mode

[JinZhou5042]

Proposed Changes

We had only one mode (cloudpickle) to wrap function arguments and send to workers.

But sometimes arguments can also be passed of a json format and we should support that.

Merge Checklist

The following items must be completed before PRs can be merged.
Check these off to verify you have completed all steps.

• make test Run local tests prior to pushing.
• make format Format source code to comply with lint policies. Note that some lint errors can only be resolved manually (e.g., Python)
• make lint Run lint on source code prior to pushing.
• Manual Update: Update the manual to reflect user-visible changes.
• Type Labels: Select a github label for the type: bugfix, enhancement, etc.
• Product Labels: Select a github label for the product: TaskVine, Makeflow, etc.
• PR RTM: Mark your PR as ready to merge.

[tphung3]

What is the inspiration behind this change, or can you please provide more context or use case? I'm thinking about the corner case of a Python class as an argument, which AFAIK isn't JSON-serializable or -deserializable out of the box with the json module.

Can you also add tests to make sure this proposed feature works?

[tphung3]

If you want to support this feature you also need to add a flag somewhere around here to serialize Python objects by JSON too, as arguments via "infile" are always serialized with cloudpickle at the moment (see

cctools/taskvine/src/bindings/python3/ndcctools/taskvine/task.py

Lines 1113 to 1125 in 2c8f80d

	name = os.path.join(self.manager.staging_directory, "arguments", self._id)
	with open(name, "wb") as wf:
	cloudpickle.dump(self._event, wf)
	self._input_file = self.manager.declare_file(name, unlink_when_done=True, cache=False, peer_transfer=True)
	if self._tmp_output_enabled:
	self._output_file = self.manager.declare_temp()
	else:
	name = os.path.join(self.manager.staging_directory, "outputs", self._id)
	self._output_file = self.manager.declare_file(name, cache=self._cache_output, unlink_when_done=False)
	self._event = None # free args memory. Once in a file they are not needed anymore.
	self.add_input(self._input_file, "infile")

)

[JinZhou5042]

What is the inspiration behind this change, or can you please provide more context or use case? I'm thinking about the corner case of a Python class as an argument, which AFAIK isn't JSON-serializable or -deserializable out of the box with the json module.

Can you also add tests to make sure this proposed feature works?

This feature enables the design of a new graph executor, where the entire graph is installed within the library. Each function call carries the key that identifies the function to execute, whose format is string. Since the task key is a string, we can pass the text directly instead of serializing it via cloudpickle.

For example, a task graph can be defined as:

def add(x, y):
    return x + y

graph = {
    "a": (add, 1, 2),
    "b": (add, "a", 3),
    "c": (add, "a", "b"),
}

Here, graph is installed in the library, the executor takes a key, retrieves the corresponding task definition, and executes it. Thus, the infile for the three tasks would look like:

{args=("a"), kwargs={}}
{args=("b"), kwargs={}}
{args=("c"), kwargs={}}

Although a single argument could suffice for representing the key, json is used for better flexibility.

[JinZhou5042]

If you want to support this feature you also need to add a flag somewhere around here to serialize Python objects by JSON too, as arguments via "infile" are always serialized with cloudpickle at the moment (see

cctools/taskvine/src/bindings/python3/ndcctools/taskvine/task.py

Lines 1113 to 1125 in 2c8f80d

	name = os.path.join(self.manager.staging_directory, "arguments", self._id)
	with open(name, "wb") as wf:
	cloudpickle.dump(self._event, wf)
	self._input_file = self.manager.declare_file(name, unlink_when_done=True, cache=False, peer_transfer=True)
	if self._tmp_output_enabled:
	self._output_file = self.manager.declare_temp()
	else:
	name = os.path.join(self.manager.staging_directory, "outputs", self._id)
	self._output_file = self.manager.declare_file(name, cache=self._cache_output, unlink_when_done=False)
	self._event = None # free args memory. Once in a file they are not needed anymore.
	self.add_input(self._input_file, "infile")

)

The new feature doesn't go through task.py, instead it creates tasks directly on the C side, I will create some examples later in a later PR!

[tphung3]

What is the inspiration behind this change, or can you please provide more context or use case? I'm thinking about the corner case of a Python class as an argument, which AFAIK isn't JSON-serializable or -deserializable out of the box with the json module.
Can you also add tests to make sure this proposed feature works?

This feature enables the design of a new graph executor, where the entire graph is installed within the library. Each function call carries the key that identifies the function to execute, whose format is string. Since the task key is a string, we can pass the text directly instead of serializing it via cloudpickle.

For example, a task graph can be defined as:

def add(x, y):
    return x + y

graph = {
    "a": (add, 1, 2),
    "b": (add, "a", 3),
    "c": (add, "a", "b"),
}

Here, graph is installed in the library, the executor takes a key, retrieves the corresponding task definition, and executes it. Thus, the infile for the three tasks would look like:

{args=("a"), kwargs={}}
{args=("b"), kwargs={}}
{args=("c"), kwargs={}}

Although a single argument could suffice for representing the key, json is used for better flexibility.

I see, you need to be careful when arguments aren't simple objects. The json module can serialize basic types but anything complex enough to be a class then this approach wouldn't work.

Also if the serialization time of cloudpickle is of a concern, can you post a benchmark result between that and json?

[JinZhou5042]

I see, the json argument is meant to be plain text in this case, so that should be fine.

Regarding the benchmark, I ran a test earlier with a DAG of one million phony tasks on 1,600 cores: using cloudpickle took one hour while the json approach took five minutes.

[btovar]

I am not completely convinced this is needed? It sounds like a library inside a library?

[JinZhou5042]

I may be misunderstanding what "a library inside a library" means here.

The goal is simply to make the infile loading mode configurable for function tasks. The default remains cloudpickle, since the manager dumps infiles via cloudpickle. This change adds an opt-in json mode for cases where the manager already dumps infiles in JSON format, and that is what DAGVine does.

[btovar]

@JinZhou5042 I'm referring in particular to what this comment implies. Why is not the function part of the library in the first place? Why not generate the library with the functions needed instead of loading them from a file which may create difficulties when debugging, create new paths for errors, etc.?