Notificar la tarea de apio del cierre del trabajador

Question

Estoy usando apio 2.4.1 con Python 2.6, el servidor de fondo rabbitmq y django. Me gustaría que mi tarea pueda limpiarse correctamente si el trabajador se cierra. Por lo que sé, no se puede suministrar un destructor de tareas, así que intenté conectarme a la señal worker_shutdown.

Nota: AbortableTask solo funciona con el back-end de la base de datos, así que no puedo usar eso.

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() 

Sin embargo, nunca se llama al gancho de desconexión. Ctrl-C'ing el trabajador no mata la tarea y tengo que matarlo manualmente desde el shell.

Entonces, si esta no es la forma correcta de hacerlo, ¿cómo puedo permitir que las tareas se apaguen correctamente?

Answer 1

worker_shutdown solo lo envía MainProcess, no a los trabajadores de grupos secundarios. Todos worker_* señales except for worker_process_init, consulte MainProcess.

Sin embargo, nunca se llama al gancho de desconexión. Ctrl-C'ing the worker no mata la tarea y tengo que matarla manualmente desde el shell.

El trabajador nunca finaliza una tarea bajo el apagado normal (cálido). Incluso si una tarea tarda días en completarse, el trabajador no completará el cierre hasta que se complete. Puede configurar --soft-time-limit o --time-limit en para indicarle a la instancia cuándo puede terminar la tarea.

Para agregar cualquier tipo de proceso de limpieza del proceso primero necesita asegúrese de que las tareas se puedan completar. Como la limpieza no se llamaría a antes de que eso suceda.

Para agregar un paso de limpieza para el trabajo de grupo de procesos puede utilizar algo como:

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)