How to call the same task, over and over, in a cyclic way?

[kinow]

Hello,

After some discussion in the Workflows Community slack channel, I got curious about Looping in Covalent. In CWL, there's a work-in-progress implementation of loops. But the CWL implementation allows users to create cycles in their workflows.

In other workflows like Cylc and ecFlow, cyclic graphs are extremely important. It's hard to find a good definition of how portable is the definition of cycles and loops between workflow managers, so I decided to run some experiments.

I managed to implement something like this:

a -> a

Meaning, the task a calls a, with this code:

import covalent as ct

@ct.electron
def a(n=-1):
    return n + 1

@ct.lattice
def run_experiment():
    r1 = a()
    r2 = a(r1)
    return r2

dispatch_id = ct.dispatch(run_experiment)()
result = ct.get_result(dispatch_id, wait=True)
print(result)

Now, I'm trying to implement what in Cylc is called "integer cycle points", where each task instance has a cycle number (the cycle point identifier). Whenever a new instance is created, we start a new cycle, increasing its number/identifier. A task then starts its next instance in the next cycle point, and so it goes either until it reaches a final condition, or ad infinitum.

a.1 -> a.2 -> a.3 -> ...

My question is, would it be possible in Covalent? This is the code that I wrote.

import covalent as ct
from time import sleep

@ct.electron
def a(cycle=0):
    # TODO: how to get current task name in covalent?
    print(f'This is task "a" cycle "{cycle}"')
    return cycle + 1

@ct.lattice
def run_experiment():
    cycle_result = a()
    while True:
        cycle_result = a(cycle_result)
        sleep(2)
    return cycle_result

dispatch_id = ct.dispatch(run_experiment)()
result = ct.get_result(dispatch_id, wait=False)
print(result)

Thank you!

Bruno

[santoshkumarradha]

Hey @kinow ,

Welcome to Covalent and thank you for your question! It's great to see that you've explored Covalent. I'll help clarify how to achieve cyclic patterns and dynamic graphs in Covalent and provide examples for your specific use case.

Understanding @ct.lattice and Electron objects

First, let's briefly discuss the @ct.lattice decorator and Electron objects. @ct.lattice objects are static compiler-like objects that gauge the dependencies between Electron objects, which are functions tagged via @ct.electron. The key words here are "static" and "compiler". When you dispatch a lattice function, Covalent executes it directly without actually calling the inside functions, but instead overrides their call method to return future Electron objects. This helps in building the dependency graph you see in the UI.

Considering your example, there are three critical points to note:

1. The sleep statement you had there wouldn't add much value, as it will just delay the graph-building step.
2. The while True statement is a dynamic compilation, meaning that it depends on what the a function does (break the loop) and will keep going. At the time of building the graph, it's impossible to exit the graph compilation stage, as the electron functions aren't executed but are only lazy-evaluated to return the Electron object.
3. What you are aiming for is a pattern called "Dynamic" graph, where parts of your graph's structure depend on the results of some previous steps.

Handling cyclic patterns with static graphs

Assuming the graph is static, we can create a cyclic pattern by slightly modifying your code:

import covalent as ct
from time import sleep

@ct.electron
def a(cycle=0):
    # TODO: how to get current task name in covalent?
    print(f'This is task "a" cycle "{cycle}"') #Not this will not be printed but will be stored in node's result as electrons are typically meant for remote running.
    return cycle + 1

@ct.lattice
def run_experiment(iterations):
    cycle_result = a()
    for _ in range(iterations):
        cycle_result = a(cycle_result)
        # sleep(2) Adding non electron calls inside lattice function is not much useful as lattice is just used to build dynamic functions
    return cycle_result

dispatch_id = ct.dispatch(run_experiment)(20)
result = ct.get_result(dispatch_id, wait=True)
print(result)

In the example above, the pattern is indeed cyclic as you asked, but the stopping criteria is determined (number of iterations, not infinite).

This is how the cyclic graph looks like for me :D

Handling dynamic graphs with sublattice

To make the system dynamic, we can use a pattern called sublattice (here is more info in the docs). This enables lattice functions inside electron objects to wait until the previous execution is completed and builds a subgraph during the workflow once its inputs are available. Here's an example:

import covalent as ct
from time import sleep

@ct.electron
def a(cycle=0):
    # TODO: how to get current task name in covalent?
    print(f'This is task "a" cycle "{cycle}"') #Not this will not be printed but will be stored in node's result as electrons are typically meant for remote running.
    return cycle + 1

@ct.electron
@ct.lattice
def a_check_continue(cycle):
    if cycle >= 10:
        return cycle
    print(f'This is task "a" cycle "{cycle}"') #Not this will not be printed but will be stored in node's result as electrons are typically meant for remote running.
    return cycle + 1
@ct.lattice
def run_experiment(iterations):
    cycle_result = a()
    for _ in range(iterations):
        cycle_result = a_check_continue(cycle_result)
        # sleep(2) Adding non electron calls inside lattice function is not much useful as lattice is just used to build dynamic functions
    return cycle_result

dispatch_id = ct.dispatch(run_experiment)(20)
result = ct.get_result(dispatch_id, wait=True)
print(result)

Now if you look at the UI and click on each sublattice's output, you would see that the output stops after 10. Though this can be achieved without this sublattice pattern (since the sub graph is not dynamic), the key concept here is that the conditions you check on the outputs of Electron objects should be pushing inside the same or other electron functions. This ensures that the value is actually computed available while it is used, and if you want to use it in a workflow fashion, then you just slap a @ct.lattice decorator to this electron as now the lattice has parameter it needs to compute the (sub)graph.

I hope this helps you understand how to handle cyclic patterns and dynamic graphs in Covalent. If you have any further questions or need clarification, please don't hesitate to ask!

Happy coding!

[santoshkumarradha]

Forgot to add on, for

TODO: how to get current task name in covalent?

You are right, there is not yet a dynamic way to change the name of electron decorator, it would be nice to have a feature to change the name of the electron other than the function, but what you propose here (to change the name during runtime workflow) might be a bit non trivial, although could be done, if thats a feature you would like or would love to contribute, please do open a feature issue and we can start by discussing the details there.

[wjcunningham7]

I'll add here the best way to get the Covalent identifiers (dispatch ID, node ID) for a task within that task is using environment variable injection. This may require a modification within the executor plugins, since that data is accessible in the run method in each executor class.

[kinow]

Hi @santoshkumarradha!

Thanks a lot for the detailed response, and sorry for the delay. Last week someone asked in the CWL Con 2024 about workflow management systems supporting cycles. I replied with the ones I had already tested prefect/stackstorm/ecflow/cylc/and mentioned this issue where I was testing covalent.

After that I tried your examples, and they worked. The static and dynamic graph examples are great! Just to make sure I understand this part,

The while True statement is a dynamic compilation, meaning that it depends on what the a function does (break the loop) and will keep going. At the time of building the graph, it's impossible to exit the graph compilation stage, as the electron functions aren't executed but are only lazy-evaluated to return the Electron object.

so that means that this is not supported at the moment in Covalent, right? Coming back to Cylc/ecFlow, these WMSs support that, and this is used - for example - in weather forecast workflows.

These two workflow managers follow a similar pattern where the workflow is started in what looks like a while True: loop(), where the loop keeps checking the status of workflow tasks, and then - similar to your dynamic graph example - decide next tasks to be added to graph or scheduled/executed.

This way I can have a workflow with a static part (normally the initialization of the general data and parametrization for models & system environment), with a recurrent part (the execution of a numerical model with the parameters/initial conditions, plus maybe data from a previous cycle/repetition), and a dynamic part (every 3 cycles clean the data, every weekend send an email, on-error add 2 other tasks to the workflow).

I think Covalent is only missing the recurrent part, as it requires a fixed number of cycles (which these two tools support too).

Thanks!!