[Documentation] How does `SingleNodeExecutor` touch the file system?

[liamhuber]

I'm trying to understand where SingleNodeExecutor with a cache_directory gets its instructions to actually write it's result to file.

For instance, consider the following:

from executorlib import SingleNodeExecutor

def foo(x):
 return x + 1

with SingleNodeExecutor() as exe:
 f = exe.submit(foo, 1, resource_dict={"cache_key": "my_key", "cache_directory": "my_dir"})
 print("Result", f.result())

My understanding of the path of events is

Initialization:

• SingleNodeExecutor.__init__ triggers BaseExecutor.__init__ with a DependencyTaskScheduler as its underlying _task_scheduler
• That DependencyTaskScheduler is initialized with a OneProcessTaskScheduler as its executor arg, which gets stored in _process_kwargs["executor"]

Submission:

• SingleNodeExecutor.submit is inherited directly fromBaseExecutor.submit
• BaseExecutor.submit passes everything (including the resource_dict as a single kwarg) to the _task_scheduler.submit
• Since DependencyTaskScheduler._generate_dependency_graph has fed through to False with all the default values, this generates the future by a plain super() call to TaskSchedulerBase.submit
• TaskSchedulerBase.submit sends our information (function, args, kwargs, resource dict, empty future) to self._future_queue.put

Here I get out of my depth, but it seems to me like putting stuff on the future queues is activating the associated Thread, which all the TaskSchedulerBase children initialize in _set_process using a Thread taking some function and the _process_kwargs (which includes the _future_queue!). On that assumption, that means that the self._future_queue.put call we got to from TaskSchedulerBase.submit would route back to the Thread set in the DependencyTaskScheduler._set_process invocation -- i.e. _execute_tasks_with_dependencies

Continuing submission:

• _execute_tasks_with_dependencies indeed takes an executor, which is DependencyTaskScheduler._process_kwargs["executor"] i.e. our OneProcessTaskScheduler; there's lots going on, but...
  • I don't see any reference to the cache, so I don't think the file system interaction is happening here
  • It looks like we ultimately do a executor_queue.put onto the underlying OneProcessTaskScheduler._future_queue
  • I.e. we move to _execute_task_in_separate_process
• _execute_task_in_separate_process is in turn re-directing to _wrap_execute_task_in_separate_process and both of these are now taking a spawner: type[BaseSpawner] argument
• But I'm at the end of the line, I don't see anything other than the default MpiExecSpawner being leveraged, and I never find any references to the cache_directory or any of the task_scheduler.file module tools

With the SlurmClusterExecutor we sometimes route through create_file_executor, in which case the file system connection is obvious, but in the other case we're still going through DependencyTaskScheduler -- this time with an SrunSpawner instead of a MpiExecSpawner. In this later case I also don't see the connection to file system tools, so I feel like I must be missing something at the diverging point: DependencyTaskScheduler.

What am I missing here? When does the SingleNodeExecutor figure out it needs to leverage the "cache_directory" field in the resource_dict?

[jan-janssen]

The quick answer is

executorlib/executorlib/task_scheduler/interactive/shared.py

Line 135 in f75a2f4

def _execute_task_with_cache(

I am going to extend this answer later.

[liamhuber]

The quick answer is

executorlib/executorlib/task_scheduler/interactive/shared.py

Line 135 in f75a2f4
def _execute_task_with_cache(

I am going to extend this answer later.

Awesome, thanks! That will already be something to get me rolling if you don't get to it today.

[jan-janssen]

Let's take an example of an SingleNodeExecutor(disable_dependencies=True), the following classes and threads are executed:

• executorlib.executor.single.SingleNodeExecutor - the interface the user interacts with - all submitted tasks are placed in one queue.
  • executorlib.task_scheduler.interactive.blockallocation.BlockAllocationTaskScheduler or executorlib.task_scheduler.interactive.onetoone.OneProcessTaskScheduler are attached to SingleNodeExecutor()._task_scheduler - the BlockAllocationTaskScheduler recycles a fixed number of processes for the execution while the OneProcessTaskScheduler creates a new process for each task. They connect to the task queue and schedule the tasks for execution.
    • executorlib.task_scheduler.interactive.shared.execute_tasks() - internally calls either _execute_task_without_cache() or _execute_task_with_cache() depending on whether a cache_directory is defined or not.

When the user submits a function it is placed in the queue, it can be grabbed by either the next BlockAllocationTaskScheduler or OneProcessTaskScheduler. For the BlockAllocationTaskScheduler they are all started at the initialization of the SingleNodeExecutor while the OneProcessTaskScheduler are created for each function separately depending on the available computational resources (max_workers).

For the case of SingleNodeExecutor(disable_dependencies=False) there is another layer to check the dependencies of the future objects:

• executorlib.executor.single.SingleNodeExecutor - the interface the user interacts with - all submitted tasks are placed in one queue.
  • executorlib.task_scheduler.interactive.dependency.DependencyTaskScheduler - is attached to SingleNodeExecutor()._task_scheduler
    • executorlib.task_scheduler.interactive.dependency._execute_tasks_with_dependencies() - consumes the queue, checks if all dependent future objects are done() and then places the task in a second queue so it is available for execution.
      • executorlib.task_scheduler.interactive.blockallocation.BlockAllocationTaskScheduler or executorlib.task_scheduler.interactive.onetoone.OneProcessTaskScheduler - execute the task when the resources become available.
        • executorlib.task_scheduler.interactive.shared.execute_tasks() - internally calls either _execute_task_without_cache() or _execute_task_with_cache() depending on whether a cache_directory is defined or not.

While the SingleNodeExecutor already comes with a decent level of complexity, the great part is that the SlurmJobExecutor and the FluxJobExecutor use exactly the same functionality, they just differ in the way they start new processes by having different process spawner interfaces: https://executorlib.readthedocs.io/en/latest/4-developer.html#interface-class-hierarchy

The FluxClusterExecutor and the SlurmClusterExecutor are different as they always rely on file based communication:

• executorlib.executor.slurm.SlurmClusterExecutor - the interface the user interacts with - all submitted tasks are placed in one queue.
  • executorlib.task_scheduler.file.task_scheduler.FileTaskScheduler - is attached to SlurmClusterExecutor()._task_scheduler
    • executorlib.task_scheduler.file.shared.execute_tasks_h5() - stores the input in the HDF5 file and then calls executorlib.task_scheduler.file.queue_spawner.execute_with_pysqa to submit the job using https://pysqa.readthedocs.io . By using a separate thread for writing the files the user does not have to wait for the IO to complete.

@liamhuber I hope this helps to understand the internal workings of executorlib. I tried to design the classes in a way that this hierarchy is also represented in the UML diagram https://executorlib.readthedocs.io/en/latest/4-developer.html#interface-class-hierarchy . Still I understand that the use of threads results in a complex hierarchy.

Any feedback is highly appreciated.

[liamhuber]

@liamhuber I hope this helps to understand the internal workings of executorlib. I tried to design the classes in a way that this hierarchy is also represented in the UML diagram https://executorlib.readthedocs.io/en/latest/4-developer.html#interface-class-hierarchy .

Ok perfect, thank you for the detailed reply! I had failed to appreciate the centrality of task_scheduler.interactive.shared.execute_tasks, so when the OneProcessTaskScheduler (et al.) started taking a spawner I got totally distracted and missed this line. Together with your explanation that line brings it all together and the mystery is cleared-up for me!

Still I understand that the use of threads results in a complex hierarchy.

Any feedback is highly appreciated.

You're correct that the hierarchy is complex, and it's true that this complexity caused me a lot of headache yesterday, but this is a place where frustration != criticism. It's possible that there is a better way to do things, but (a) I certainly don't see it, and (b) I wouldn't expect it ever to make things easy. What you do is fundamentally complex and so I anticipate and accept complexity in the solution.

One pain point I had that may be more soluble is in tracking the flow of data. For example, here I was trying to track and follow where the cache directory got used. It's a named and typed kwarg in execute_tasks, but at least in the path of SingleNodeExecutor->...->ONeProcessTaskScheduler, gets

• passed in as kwargs=task_kwargs
• which are updated from resource_dict
• which was popped from a task_dict
• which got .get from a queue
• I've lost track here, but I imagine it got .put to the queue
• All of which is quite a ways from where my SingleNodeExecutor passes it to the OneProcessTaskScheduler with executor_kwargs=resource_dict
• Where I know that the dict was constructed with my cache directory from the user-facing class instantiation

The path being long is fine (or at least a tolerable price to pay for the complex behaviour achieved), and there are particular endpoints that are demanding the information in a pure-dict form (e.g. Thread initialization, the queue .put, etc.) However, when the information is exclusively in this dict format, it makes it extremely difficult to know what information is produced/consumed in what location.

So rather than the hierarchy, for me the most useful refactor would be to have the dictionaries pass through structured dataclasses (at least intermittently) so I can get touch-points about what possible sets of data I'm looking at in a given part of the process. Right now I'm dependent on looking at the variable names for the dictionaries (which very reasonably change depending on their context) and manually following the dictionary manipulations back to source. Having intermediate points where the dictionaries get cast to a specific format (or a DC1 | DC2 | DC3 subset of formats) would really help me to follow the process easier, know where my critical piece of information may still exist, and differentiate branches of the flow.

[jan-janssen]

I reopened the pull request, to remind me to extend the documentation later on.