कर्मचारी शटडाउन

Question

के अजवाइन कार्य को सूचित करें मैं अजगर 2.4.1 का उपयोग कर पाइथन 2.6, खरगोश बैकएंड और डीजेंगो के साथ कर रहा हूं। यदि कार्यकर्ता बंद हो जाता है तो मैं अपने काम को ठीक से साफ करने में सक्षम होना चाहता हूं। जहां तक ​​मुझे पता है कि आप एक कार्य विनाशक की आपूर्ति नहीं कर सकते हैं इसलिए मैंने worker_shutdown सिग्नल में हुकिंग करने की कोशिश की।

नोट: AbortableTask केवल डेटाबेस बैकएंड के साथ काम करता है, इसलिए मैं इसका उपयोग नहीं कर सकता।

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

हालांकि, बंद हुक कभी नहीं कहा जाता हो जाता है। कार्यकर्ता को Ctrl-Cing करना कार्य को मार नहीं देता है और मुझे इसे खोल से मैन्युअल रूप से मारना होगा।

तो यदि यह इसके बारे में जाने का उचित तरीका नहीं है, तो मैं कार्यों को शानदार तरीके से बंद करने की अनुमति कैसे दूं?

Answer 1

worker_shutdown केवल MainProcess द्वारा भेजा गया है, न कि बाल पूल श्रमिकों। सभी worker_* संकेतों except for worker_process_init, MainProcess को देखें।

हालांकि, शट डाउन हुक कभी नहीं बुलाया जाता है। कार्यकर्ता को Ctrl-C'ing कार्य को मारता नहीं है और मुझे इसे खोल से मैन्युअल रूप से मारना है।

कर्मचारी कभी भी सामान्य (गर्म) शटडाउन के तहत एक कार्य को समाप्त नहीं करता है। भले ही कोई कार्य पूरा होने में दिन लगे, कार्यकर्ता को पूरा होने तक बंद नहीं करेगा। जब आप कार्य को समाप्त करना ठीक करते हैं तो उदाहरण के लिए आप --soft-time-limit, या --time-limit से सेट कर सकते हैं।

तो किसी भी प्रकार की प्रक्रिया क्लीनअप प्रक्रिया को जोड़ने के लिए आपको पहले की आवश्यकता है सुनिश्चित करें कि कार्य वास्तव में पूर्ण हो सकते हैं। चूंकि सफाई होने से पहले को क्लीनअप नहीं कहा जाएगा।

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)