Notifica task celeste di spegnimento operaio

Question

Sto usando il celenter 2.4.1 con python 2.6, il backend rabbitmq e django. Vorrei che il mio compito fosse in grado di ripulire correttamente se l'operatore si spegne. Per quanto ne so, non è possibile fornire un distruttore per le attività, quindi ho provato ad agganciare il segnale worker_shutdown.

Nota: AbortableTask funziona solo con il database back-end, quindi non posso utilizzarlo.

from celery.signals import worker_shutdown 

@task 
def mytask(*args) 

    obj = DoStuff() 

    def shutdown_hook(*args): 
    print "Worker shutting down" 
    # cleanup nicely 
    obj.stop() 

    worker_shutdown.connect(shutdown_hook) 

    # blocking call that monitors a network connection 
    obj.stuff() 

Tuttavia, il gancio di arresto non viene mai chiamato. Ctrl-Cing il lavoratore non uccide l'attività e devo ucciderlo manualmente dalla shell.

Quindi, se questo non è il modo corretto per farlo, come posso consentire alle attività di spegnersi con garbo?

Answer 1

worker_shutdown viene inviato solo dallo MainProcess, non dagli operatori di pool figlio. Tutti i segnali worker_*except for worker_process_init, fare riferimento a MainProcess.

Tuttavia, il gancio di arresto non viene mai chiamato. Ctrl-Cing the worker non uccide l'attività e devo ucciderla manualmente dalla shell.

L'operatore non termina mai un'attività durante lo spegnimento normale (caldo). Anche se un compito richiede giorni per essere completato, il lavoratore non completerà lo spegnimento fino al completamento. È possibile impostare --soft-time-limit o --time-limit su per indicare all'istanza quando è possibile terminare l'attività.

Quindi per aggiungere qualsiasi tipo di processo di pulizia del processo è necessario prima di assicurarsi che le attività possano effettivamente essere completate. Come la pulizia non sarebbe essere chiamato prima che ciò accada.

Per aggiungere una sottofase di cancellazione al lavoratore piscina processi è possibile utilizzare qualcosa di simile:

from celery import platforms 
from celery.signals import worker_process_init 

def cleanup_after_tasks(signum, frame): 
    # reentrant code here (see http://docs.python.org/library/signal.html) 

def install_pool_process_sighandlers(**kwargs): 
    platforms.signals["TERM"] = cleanup_after_tasks 
    platforms.signals["INT"] = cleanup_after_tasks 

worker_process_init.connect(install_pool_process_sighandlers)