Running a sequential task 3 parallel

[kuatroka]

Hi, let's imagine I've got three tasks/functions running in parallel. They don't have any return statements. They download files from urls. Their runtime is different and sometimes different task finishes last.

I want to start a fourth task ONLY after all three parallel tasks mentioned above finished execution.

Is there a way to do it?
Thanks

[Jypear]

Not sure if it helps in your situation (Not tested in multiprocessing or async) but there is a after_all_finish from rocketry.conds you can use. It accepts *args from the tasks you want to run the task after.

The way it works is it will only execute a task after all the tasks passed into the function have run atleast once.

import logging
from rocketry import Rocketry
from rocketry.conds import after_all_finish

app = Rocketry()

logging.basicConfig(
     level=logging.INFO, 
     format= '[%(asctime)s] %(levelname)s - %(message)s',
     datefmt='%H:%M:%S'
 )

logger = logging.getLogger(__name__)

@app.task("every 4 seconds")
def taskone():
    logger.info("running task 1")

@app.task("every 3 seconds")
def tasktwo():
    logger.info("running task 2")


@app.task("every 10 seconds")
def taskthree():
    logger.info("running task 3")

@app.task(after_all_finish(taskone, tasktwo, taskthree))
def taskfour():
    logger.info("running task 4")
    # This task will only run once tasks 1, 2 and 3 have run and complete
    # atleast one time each.


app.run()

OUTPUT:

[15:19:57] INFO - Starting up...
[15:19:57] INFO - Startup complete.
[15:19:57] INFO - Task 'taskone' status: 'run'
[15:19:57] INFO - Task 'taskthree' status: 'run'
[15:19:57] INFO - Task 'tasktwo' status: 'run'
[15:19:57] INFO - running task 1
[15:19:57] INFO - Task 'taskone' status: 'success'
[15:19:57] INFO - running task 3
[15:19:57] INFO - Task 'taskthree' status: 'success'
[15:19:57] INFO - running task 2
[15:19:57] INFO - Task 'tasktwo' status: 'success'
[15:19:57] INFO - Task 'taskfour' status: 'run'
[15:19:57] INFO - running task 4
[15:19:57] INFO - Task 'taskfour' status: 'success'
[15:20:00] INFO - Task 'tasktwo' status: 'run'
[15:20:00] INFO - running task 2
[15:20:00] INFO - Task 'tasktwo' status: 'success'
[15:20:01] INFO - Task 'taskone' status: 'run'
[15:20:01] INFO - running task 1
[15:20:01] INFO - Task 'taskone' status: 'success'
[15:20:03] INFO - Task 'tasktwo' status: 'run'
[15:20:03] INFO - running task 2
[15:20:03] INFO - Task 'tasktwo' status: 'success'
[15:20:05] INFO - Task 'taskone' status: 'run'
[15:20:05] INFO - running task 1
[15:20:05] INFO - Task 'taskone' status: 'success'
[15:20:06] INFO - Task 'tasktwo' status: 'run'
[15:20:06] INFO - running task 2
[15:20:06] INFO - Task 'tasktwo' status: 'success'
[15:20:07] INFO - Task 'taskthree' status: 'run'
[15:20:07] INFO - running task 3
[15:20:07] INFO - Task 'taskthree' status: 'success'
[15:20:07] INFO - Task 'taskfour' status: 'run'
[15:20:07] INFO - running task 4
[15:20:07] INFO - Task 'taskfour' status: 'success'

As you can see, task 4 runs after all three tasks have one time each. Because tasks 1 and 2 run more frequently than task 3, The output shows them running a couple times before task 3 runs again. Once task 3 runs, becuase all the tasks have run and complete this triggers task 4. In your case, if task 1, 2 and 3 run at the same frequency, it should only trigger task 4 once they have all finished.

This is probably the best solution for your use case, but if it doesn't work in your scenario for whatever reason, you could also use some kind of global variable bodge with Rocketry params to track task states and then have "task 4" run more frequently and only execute once the tasks have finished running.

Something like this

import logging
from rocketry import Rocketry
from rocketry.args import Arg
from rocketry.conds import after_all_finish

app = Rocketry()

logging.basicConfig(
     level=logging.INFO, 
     format= '[%(asctime)s] %(levelname)s - %(message)s',
     datefmt='%H:%M:%S'
 )

app.params(
    task_one_state = False,
    task_two_state = False,
    task_three_state = False
)

logger = logging.getLogger(__name__)

@app.task("every 7 seconds")
def taskone():
    logger.info("running task 1")
    app.params(task_one_state = True)

@app.task("every 8 seconds")
def tasktwo():
    logger.info("running task 2")
    app.params(task_two_state = True)


@app.task("every 9 seconds")
def taskthree():
    logger.info("running task 3")
    app.params(task_three_state = True)

@app.task("every 2 seconds")
def taskfour(
    task_one_state = Arg("task_one_state"), 
    task_two_state = Arg("task_two_state"), 
    task_three_state = Arg("task_three_state")
    ):
    logger.info("running task 4")
    if (task_one_state 
        and task_two_state 
        and task_three_state
        ):
        logger.info("will run the rest of task 4 stuff")
        app.params(
            task_one_state = False,
            task_two_state = False,
            task_three_state = False
        )
    else:
        logger.info("all of the tasks haven't run yet")


app.run()

Not a big fan of that solution though, but it should work I think. Essentially each task is setting its own 'state' to True. Task 4 checks that all states are True and if not doesn't do anything. Once all are True task 4 can run, and once finished will set the task states back to 'False' again.

[kuatroka]

Thanks a lot @Jypear. A very helpful answer. I've never used rocketry and before investing my time in it I needed to know if I should dig deeper or go with things like Dagster, Prefect...etc. I see now that it makes sense to invest some time into reading the docs. My main objective will be to see how to make the three first functions run in parallel. I think while glancing over the docs, I saw a note that Rocketry has this functionality.

Thanks again