celery task_postrun सिग्नल में SystemExit को बढ़ाकर शटडाउन कार्यकर्ता की कोशिश कर रहा है लेकिन हमेशा लटकता है और मुख्य प्रक्रिया कभी भी

Question

से बाहर नहीं निकलती है। मैं task_postrun सिग्नल में किशमिश SystemExit() द्वारा मुख्य अजवाइन प्रक्रिया को बंद करने का प्रयास कर रहा हूं। सिग्नल ठीक से निकाल दिया जाता है, और अपवाद उठाया जाता है, लेकिन कार्यकर्ता कभी पूरी तरह से बाहर नहीं निकलता है और बस वहां लटकता है।

मैं यह काम कैसे कर सकता हूं?

क्या मैं कहीं कुछ सेटिंग भूल रहा हूं?

नीचे कोड है कि मैं कार्यकर्ता के लिए उपयोग कर रहा हूँ (worker.py) है:

from celery import Celery 
from celery import signals 

app = Celery('tasks', 
    set_as_current = True, 
    broker='amqp://guest@localhost//', 
    backend="mongodb://localhost//", 
) 

app.config_from_object({ 
    "CELERYD_MAX_TASKS_PER_CHILD": 1, 
    "CELERYD_POOL": "solo", 
    "CELERY_SEND_EVENTS": True, 
    "CELERYD_CONCURRENCY": 1, 
    "CELERYD_PREFETCH_MULTIPLIER": 1, 
}) 

def shutdown_worker(**kwargs): 
    print("SHUTTING DOWN WORKER (RAISING SystemExit)") 
    raise SystemExit() 

import tasks 

signals.task_postrun.connect(shutdown_worker) 

print("STARTING WORKER") 
app.worker_main() 
print("WORKER EXITED!") 

और नीचे tasks.py के लिए कोड है:

from celery import Celery,task 
from celery.signals import task_postrun 
import time 

from celery.task import Task 

class Blah(Task): 
    track_started = True 
    acks_late = False 

    def run(config, kwargs): 
     time.sleep(5) 
     return "SUCCESS FROM BLAH" 

def get_app(): 
    celery = Celery('tasks', 
     broker='amqp://guest@localhost//', 
     backend="mongodb://localhost//" 
    ) 
    return celery 

इस परीक्षण के लिए,

>>> import tasks 
>>> results = [] 
>>> results.append(tasks.Blah.delay({})) 

उत्पादन: पहली बात मैं कर कार्यकर्ता कोड (python worker.py) चलाया जाता है, तो मैं बहुत की तरह एक कार्य हाथ में लिए कतार मैं कार्यकर्ता से देख यह है:

STARTING WORKER 

-------------- celery@hostname v3.0.9 (Chiastic Slide) 
---- **** ----- 
--- * *** * -- [Configuration] 
-- * - **** --- . broker:  amqp://guest@localhost:5672// 
- ** ---------- . app:   tasks:0x26f3050 
- ** ---------- . concurrency: 1 (solo) 
- ** ---------- . events:  ON 
- ** ---------- 
- *** --- * --- [Queues] 
-- ******* ---- . celery:  exchange:celery(direct) binding:celery 
--- ***** ----- 

[2012-11-06 15:59:16,761: WARNING/MainProcess] celery@hostname has started. 
[2012-11-06 15:59:21,785: WARNING/MainProcess] SHUTTING DOWN WORKER (RAISING SystemExit) 

मैं अजगर कोड उम्मीद कर रहा था कॉल से app.worker_main() पर लौटने के लिए और उसके बाद WORKER EXITED मुद्रित करने के लिए और उसके बाद प्रक्रिया पूरी तरह से बाहर निकलने के लिए के लिए। यह कभी नहीं होता है, और कार्यकर्ता प्रक्रिया kill -KILL {PID} इसे दूर जाने के लिए (यह या तो किसी अन्य कार्यों का उपभोग करने में सक्षम नहीं है के लिए एड हो गया है

मुझे लगता है कि मेरी मुख्य सवाल होगा:।

कैसे मैं नहीं देते कोड को वापस app.worker_main() से?

मैं के बाद कार्यों की X संख्या के बाद मार दिया (प्रक्रिया पूरी तरह से बाहर निकलने के होने से) पूरी तरह से कार्यकर्ता प्रक्रिया को पुनः प्रारंभ करने में सक्षम होना चाहते हैं।

अद्यतन मैं पता लगा कार्यकर्ता पर क्या व्यतीत कर रहा है - कार्यकर्ता (WorkController) self.stop के लिए एक कॉल पर लटका होने के बाद यह SystemExit अपवाद फैल जाती है।

Answer 1

जब मैं अपने प्रश्न का उत्तर समाप्त करता हूं तो इसे नफरत है।

Anywhoo, यह WorkController अंदर Mediator घटक पर एक में शामिल होने के आह्वान पर अवरुद्ध किया गया (Mediator घटक, अंदर बंद है, यह join रों पर stop() कॉल)।

मैंने सभी दर सीमाओं को अक्षम करके Mediator घटक से छुटकारा पा लिया (डिफ़ॉल्ट रूप से यह होना चाहिए, लेकिन यह किसी कारण से नहीं है)।

आप सेटिंग के साथ सभी दर सीमा निष्क्रिय कर सकते हैं:

CELERY_DISABLE_ALL_RATE_LIMITS: True 

आशा इस मदद करता है किसी और सड़क भी नीचे।

शांति