Non-blocking workflow execution

[iparask]

From @euhruska:

Feature request for entk to enable adding individual non-blocking tasks.

Two proposed new function calls (to be discussed):

appman.run_non_blocking()
Description: Starts an Application manager with an pipeline. Pipeline might contain only an empty stage. Keep running even if no tasks are left to be run until walltime runs out. This call has to be non-blocking.

appman.run_task_nonblocking(task)
Description: Add to above Application manager one task. This call has to be non-blocking. Will be executed as soon as possible. No data transfer from tasks back except task.state.

Mock code:
https://github.com/radical-cybertools/radical.midas/blob/master/midas_mock_non_blocking.py

Schema:
https://docs.google.com/presentation/d/1uXyYE10QkU33r6ucnk_PSScGLrsWbLccd4WSiG61bs8

[iparask]

My notes:

1. Currently, EnTK supports the execution of tasks where any logic that creates a dependency between tasks, apart from temporal, is defined with the task.
2. The definition of such logic is done at creating an application, whether it is just an EnTK script or a python executable. As a result, the logic is considered static.
3. Such logic can be executed while a stage is executing.
4. As a result, the execution of a workflow should be non-blocking.
5. As a workflow is executing, specifically a stage, those may change based on the logic's results.
6. Provide access to python's interpreter terminal, and thus point 4, to the user to do analysis on the client-side and push new tasks on pipelines and stages.
7. We need to be able to probe the state of a task during runtime.
8. Do we need to move data? No, since this use case is for systems like Summit where the execution is launched from the login node, and as such, the python interpreter already has access to the shared filesystem.
9. We need to pull information from a task as it is executing. What type of information?
10. Stressed that our systems are libraries and, as such, the user is responsible for any interactive use.

[mturilli]

appman.run_non_blocking()
Description: Starts an Application manager with an pipeline. Pipeline might contain only an empty stage. Keep running even if no tasks are left to be run until walltime runs out. This call has to be non-blocking.

appman.run_non_blocking() would be equivalent to submit a pilot with no tasks, i.e., submitting a job to an HPC machine running only RP Agent without running any task. As we pay (allocation and, ultimately, money) whether we run a task or not, we should wait to submit a pilot until we have at least one task to execute. We could still specify the Pipeline and the stage, do whatever we need to do before being able to specify the first task, and then add that first task to the stage. At that point, we could submit the pipeline for execution, triggering the submission of a pilot.

appman.run_task_nonblocking(task)
Description: Add to above Application manager one task. This call has to be non-blocking. Will be executed as soon as possible. No data transfer from tasks back except task.state.

If we follow the above, I would propose to have something like:

p = Pipeline()
s = Stage()
t1 = Task()

t1.name = 'my-first-task'
t1.executable = '/bin/echo'
t1.arguments = ['Hello World 1']

s.add_tasks(t1)
p.add_stages(s)

appman = AppManager(hostname=hostname, port=port, username=username, password=password)
res_dict = {'resource': 'local.localhost', 'walltime': 10, 'cpus': 1}
appman.resource_desc = res_dict
appman.workflow = p
appman.run_nonblocking(terminate=False)

t2.name = 'my-first-task'
t2.executable = '/bin/echo'
t2.arguments = ['Hello World 2']

s.add_tasks(t2)

That would imply that, if appman has submitted t1, appman would submit t2 as soon as it is added, avoiding the need of a dedicated new call to run a task. Terminate=false would enable appman to keep the pilot alive even if t1 has finished to execute before t2 is submitted.

As discussed, there are some implications on EnTK's state model. Currently, p would be marked as Done if t2 is submitted after t1 has ended. We can discuss different approaches to this, also taking into account what we learned about pausing/restarting/terminating pipelines in REPEX. For example, should we add an explicit call to terminate a pipeline and stages? We could replace appman.workflow with:

appman.add_workflow = p
...
appman.del_workflow = p

In that way, p would not be marked as terminated until del_workflow is called. We would have to automatically move p to a final state in some conditions, e.g., explicit pilot termination. Note that we would not do that when a pilot crashes and appman has been configured to resubmit that pilot. We could do the same with stages and manage automatic termination with priorities between p and s, i.e., terminate a state if its pipeline has reached a final state.

There are many details/conditions I am not explicitly addressing here and there are other approaches to consider but, if you like the approach I suggested, I will be happy to work together towards a full spec.

[euhruska]

I think RP is sufficient for this use case and the required task dependencies would be easier to implement in RP than in EnTK. Do we close this discussion?

[mturilli]

I am afraid adding non-blocking calls in RP is not in the current roadmap so I would stick to EnTK. @andre-merzky would you confirm that? @euhruska in our last call, my notes tell me that we explicitly discuss the necessity of representing dependencies among tasks. I am not sure I wrote it down correctly, but I noted that you had 4 types of tasks that needed to execute one after the other. Did I get it wrong?

[euhruska]

umgr.submit_units() appears to behave non-blocking and can be called for each task separately, just ran a mock code demonstrating this functionality. @mturilli The dependencies would be executed outside of EnTK/RP, between the non-blocking calls. There would be just one type of tasks from EnTK/RP perspective, but the user could define each task as desired.

[mturilli]

If RP and umgr.submit_units() work for you happy to go that way. The whole idea of RCT is to enable to users not to have to code the abstractions that each RCT provides. For example, in this case you seem to wanting to write code to manage task dependencies. The idea of EnTK is to provide those capabilities for you. This is why we would be interested anyway to see how we would have to use EnTK to satisfy your use case.

[euhruska]

Agreed. The requirement of non-blocking task submission made EnTK not currently applicable. Once the use case is demonstrated on RP level or if the task dependencies would be too complex, we still can follow the above discussed approach to support non-blocking tasks in EnTK.

[iparask]

A question just came to my mind. Will we allow to mix blocking and non-blocking execution? Something like:

appman.submit_pipelines(set([p]))
appman.run(blocking=False)
... I'm doing something

appman.submit_pipelines(set([p]))
appman.run(blocking=True)