Is there a distributed task queue in Python that allows me to kill hanging tasks that don't want to collaborate

Question

Basically, I have many tasks (in batches of about 1000), and the execution time for these tasks can vary widely (from less than seconds to 10 minutes). I know that if a task runs for more than a minute, I can kill it. These tasks are steps in optimizing some data mining model (but are not dependent on each other) and spend most of the time inside some extension function C, so they will not work together if I try to kill them gracefully.

Is there a distributed task queue that fits into this scheme --- AFAIK: celery allows you to cancel tasks that are ready for cooperation. But I could be wrong.

I recently asked a similar question about destroying dangling functions in pure python Kill a dangling function in Python in a multi-threaded environment .

I think I could subclass the celery task so that it spawns a new process and then fulfills its payload, interrupting its execution if it takes a long time, but then I will be killed due to the initialization of the new interpreter.

Answer 1

Celery supports time limit . You can use time limits to kill long tasks. In addition to tasks related to killing, you can use soft constraints that allow you to handle SoftTimeLimitExceeded exceptions in tasks and perform tasks cleanly.

from celery.task import task from celery.exceptions import SoftTimeLimitExceeded @task def mytask(): try: do_work() except SoftTimeLimitExceeded: clean_up_in_a_hurry() 

Answer 2

Pistil allows you to manage multiple processes, including killing incompatible tasks.

But:

• this is a beta version of the software, even if it protects against weapons,
• I do not know how it processes 1000 processes.
• The connection between the process is not yet enabled, so you will need to configure your own use, for example zeromq

Another possibility is to use a timer signal, so it throws an exception after 36000 seconds. But the signals are not interrupted if someone acquires a GIL that can execute program C.

Answer 3

When you cancel the celery task, you can provide it with the optional keyword terminate=True .

 task.revoke(terminate=True) 

This does not meet your requirements, because it is not performed by the process itself, but you must be able to either expand the task class to be able to commit suicide, or perform a second cleaning task or a process that kills tasks that were not completed on time.