Run same job multiple times in parallel threads

[seimnseimn]

Using a slightly modified Task Queue example from (https://rocketry.readthedocs.io/en/stable/cookbook/controlling_runtime.html#task-queue)

I'd like to run the "same" task multiple times in parallel (executing="thread"). I know how to do it with different tasks but not with the same. How could I accomplish this (running job_a with arg 1 & job_a with arg 2 at the same time)?

import time

from rocketry import Rocketry
from rocketry.args import Session

app = Rocketry(config={"task_execution": "async"})

@app.task(execution="thread")
def job_a(arg):
    print(f"Running job_a {arg}...")
    time.sleep(3)
    print(f"job_a {arg} has finished!")

@app.task(execution="thread")
def job_b(arg):
    print(f"Running job_b {arg}...")
    time.sleep(3)
    print(f"job_b {arg} has finished!")

def next_task():
    yield "job_a", {"arg": "1"}
    yield "job_a", {"arg": "2"}
    yield "job_b", {"arg": "3"}

@app.task(on_startup=True, execution="thread")
def task_queue(session=Session()):
    queue = next_task()
    for task_name, params in queue:
        task = app.session[task_name]
        task.run(**params)

if __name__ == "__main__":
    app.run()

output:

Running job_a 1...
Running job_b 3...
job_a 1 has finished!
job_b 3 has finished!
Running job_a 2...
job_a 2 has finished!

desired output:

Running job_a 1...
Running job_a 2...
Running job_b 3...
job_a 1 has finished!
job_a 2 has finished!
job_b 3 has finished!

[Miksus]

Thanks for this! I have thought that allowing only one run per task should not be an issue because it's easy to create duplicates, but in this case, it's indeed not so. I think the only approach at the moment is to duplicate the tasks though.

Perhaps it should be made so that the runs are separate from the tasks: we log which task started and we also log some sort of a run ID. There are still some problems: what is the definition of the task status if it was started twice and the other run succeeded? Is it running or was it a success?

But I think we could have another attribute for tasks that releases the run lock and allows a specified task to be run multiple times at the same time. It should be only used when necessary as it can cause a mess in the log. This should be pretty easy to do at least so this could be the next step.

Perhaps something like this:

@app.task(multilaunch=True)
def do_things():
    ...

The major problem is that if there is a condition that's not tied to the runs (ie. time_of_day.at("10:00")), you will end up with a lot of do_things running at the same time. Thus this option should be reserved only for this sort of situation.

What's your opinion?

What comes to the task queues, I had a brilliant (or stupid) idea yesterday. I'll open another discussion about it soon.

[seimnseimn]

Hmmm... I see your point regarding task status and more over controlling/manipulating the task etc...
simply having a task.name which is the func_name isn't sufficient anymore. And I really love the simplicity of your framework :-)

Funny enough when looking at your #123 this is very close to what I'm trying to accomplish.

Sticking to your example in #123 maybe we're adding a lot of person's into the database and process_person() is taking some time to process (e.g. I/O intense or whatever) so it would make also sense to be able to parallelise the process_person() ?

[Miksus]

Thanks! Actually, I created Rocketry because Cron was just unreadable and Airflow wasn't really beginner friendly. It's increasingly looking like Rocketry is not only making things easy but also enabling patterns that are not possible with other frameworks.

so it would make also sense to be able to parallelise the process_person() ?

That's true. Rocketry pre-2.4.0 did not support batched (run specific) parameters thus this kind of use case emerged quite recently.

But I think an optional bypass of the run lock is pretty easy to implement. Just add the attribute to the Task base class and override is_not_running check here: rocketry/core/scheduler.py.

What did you think of having it named as multilaunch=True? Basically this would be False by default but if set as True, the task is allowed to be started multiple times.

Furthermore, I think we could make the task.run() to override this restriction as well. By default, it shouldn't due to backward compatibility and due to that it could cause more confusion, but there could be a config option to do this. Perhaps named as multilaunch_on_batch: if this is True the task.run() will start the task regardless if it was already running. Not sure which config option name would be the most understandable.

Thanks again for your thoughts. I got more ideas on #123.

[seimnseimn]

The simplicity of Rocketry was the main reason why I chose to go with it. Airflow or Celery was just to complex and cumbersome for my current project.

What did you think of having it named as multilaunch=True? Basically this would be False by default but if set as True, the task is allowed to be started multiple times.

I agree. ignore_alive_check=True seems too low-level. So yes, multilaunch=True is easy understandable.

Furthermore, I think we could make the task.run() to override this restriction as well

Not sure If I understand this enough. If I have

@app.task(execution="thread", multilaunch=True)
def mytask():
    print("executing...")

and start it with:

task = app.session["mytask"]
task.run()

wouldn't this be enough?

Thank you for your effort. The more I'm working with Rocketry the more applications I see for it! Especially in the DevOps area.

[Miksus]

wouldn't this be enough?

I was thinking that it could be preferred that manual run (task.run()) ignores always the alive check even if the task normally doesn't. I mean, for example: your production runs data for today. You don't want that to accidentally run twice. However, if you manually set a task running for the previous day (ie. task.run(report_date="2022-09-28")) you might want it to run parallel with the production run as that should not interfere with it.

However, now as I thought about it more and I think maybe that's not a good idea to be as default. But having the multilaunch should be a good starting point. I'll make a ticket about this soon.

Thank you for your effort. The more I'm working with Rocketry the more applications I see for it! Especially in the DevOps area.

And thank you for the kind comment! And I have the same: the more I think about it the more use cases I find for it: automating manual tasks, RPA, web scraping, data engineering, IOT, and indeed DevOps. I have also been thinking of doing a simulator with it at some point. And there is a fair chance that someone will use Rocketry for something game-changing.

[Miksus]

I went to shower and had more ideas about the subject.

It's actually not as simple as it might initially sound. The problem is that the processes/threads/async tasks are maintained in single variables. If multi-launch was enabled, the latest of these will be in the variable. Thus this would require a bit of refactoring and I think we would need some sort of a run stack that contains the process/threads/tasks. This could possibly be a dict so that we could have the start time there and perhaps have some sort of batch logging.

But I think this is doable but probably will take some days to complete (cannot promise when this is implemented but probably in the next version).