पायथन मल्टीप्रोसेसिंग पूल

Question

मैं रेंडरफार्म पर काम कर रहा हूं, और मुझे अपने क्लाइंट को रेंडरर के कई उदाहरण लॉन्च करने में सक्षम होने की आवश्यकता है, बिना अवरुद्ध किए, ताकि क्लाइंट नए आदेश प्राप्त कर सके। मुझे यह सही ढंग से काम कर रहा है, हालांकि मुझे बनाई गई प्रक्रियाओं को समाप्त करने में परेशानी हो रही है।

वैश्विक स्तर पर, मैं अपने पूल को परिभाषित (ताकि मैं इसे किसी भी समारोह से उपयोग कर सकते हैं):

p = Pool(2) 

मैं तो apply_async के साथ मेरी रेंडरर फोन:

for i in range(totalInstances): 
    p.apply_async(render, (allRenderArgs[i],args[2]), callback=renderFinished) 
p.close() 

यही खत्म ढंग से काम , पृष्ठभूमि में प्रक्रियाओं को लॉन्च करता है, और नए आदेशों की प्रतीक्षा करता है। मैं एक साधारण आदेश है कि ग्राहक को मारने और बंद हो जाएगा बनाया है प्रस्तुत करता है:

def close(): 
'close this client instance' 
tn.write ("say "+USER+" is leaving the farm\r\n") 
try: 
    p.terminate() 
except Exception,e: 
    print str(e) 
    sys.exit() 
sys.exit() 

यह एक त्रुटि (यह त्रुटि मुद्रित होगा) देने के लिए नहीं लगता है, अजगर समाप्त लेकिन पृष्ठभूमि प्रक्रियाओं अभी भी कर रहे हैं चल रहा है। क्या कोई इन लॉन्च प्रोग्राम्स को नियंत्रित करने का बेहतर तरीका सुझा सकता है?

Answer 1

मिले अपने ही सवाल का जवाब। प्राथमिक समस्या यह थी कि मैं एक समारोह के बजाय तीसरे पक्ष के आवेदन को बुला रहा था। जब मैं उपप्रोसेसर को कॉल करता हूं [या तो कॉल() या पॉपन() का उपयोग करके यह पाइथन का एक नया उदाहरण बनाता है जिसका एकमात्र उद्देश्य नया एप्लिकेशन कॉल करना है। हालांकि जब पायथन निकलता है, तो यह पाइथन के इस नए उदाहरण को मार देगा और एप्लिकेशन को चल रहा है।

समाधान यह है कि इसे बनाया गया पाइथन प्रक्रिया की पिड ढूंढकर, उस पिड के बच्चों को प्राप्त करने और उन्हें मारने का समाधान कठिन तरीका है। यह कोड ओएसएक्स के लिए विशिष्ट है; लिनक्स के लिए उपलब्ध सरल कोड (जो grep पर निर्भर नहीं है) है।

for process in pool: 
    processId = process.pid 
    print "attempting to terminate "+str(processId) 
    command = " ps -o pid,ppid -ax | grep "+str(processId)+" | cut -f 1 -d \" \" | tail -1" 
    ps_command = Popen(command, shell=True, stdout=PIPE) 
    ps_output = ps_command.stdout.read() 
    retcode = ps_command.wait() 
    assert retcode == 0, "ps command returned %d" % retcode 
    print "child process pid: "+ str(ps_output) 
    os.kill(int(ps_output), signal.SIGTERM) 
    os.kill(int(processId), signal.SIGTERM) 

Answer 2

यदि आप अभी भी इस समस्या का सामना कर रहे हैं, तो आप Pool को daemonic processes के साथ अनुकरण करने का प्रयास कर सकते हैं (मान लें कि आप गैर-डेमोनिक प्रक्रिया से पूल/प्रक्रियाएं शुरू कर रहे हैं)। मुझे संदेह है कि यह सबसे अच्छा समाधान है क्योंकि ऐसा लगता है कि आपकी Pool प्रक्रियाएं निकलनी चाहिए, लेकिन यह सब मैं साथ आ सकता हूं। मुझे नहीं पता कि आपका कॉलबैक क्या करता है, इसलिए मुझे यकीन नहीं है कि इसे नीचे मेरे उदाहरण में कहां रखा जाए।

मैं आपके अनुभव (और दस्तावेज़) के कारण में बनाने की कोशिश करने का भी सुझाव देता हूं, जब प्रक्रियाएं वैश्विक स्तर पर फैली हुई होती हैं तो अजीबता होती है। यह विंडोज पर विशेष रूप से सच है अगर आप कर रहे हैं है: http://docs.python.org/2/library/multiprocessing.html#windows

from multiprocessing import Process, JoinableQueue 

# the function for each process in our pool 
def pool_func(q): 
    while True: 
     allRenderArg, otherArg = q.get() # blocks until the queue has an item 
     try: 
      render(allRenderArg, otherArg) 
     finally: q.task_done() 

# best practice to go through main for multiprocessing 
if __name__=='__main__': 
    # create the pool 
    pool_size = 2 
    pool = [] 
    q = JoinableQueue() 
    for x in range(pool_size): 
     pool.append(Process(target=pool_func, args=(q,))) 

    # start the pool, making it "daemonic" (the pool should exit when this proc exits) 
    for p in pool: 
     p.daemon = True 
     p.start() 

    # submit jobs to the queue 
    for i in range(totalInstances): 
     q.put((allRenderArgs[i], args[2])) 

    # wait for all tasks to complete, then exit 
    q.join() 

Answer 3

मैं समाधान नहीं मिला: अलग थ्रेड में पूल बंद करो, इस तरह:

def close_pool(): 
    global pool 
    pool.close() 
    pool.terminate() 
    pool.join() 

def term(*args,**kwargs): 
    sys.stderr.write('\nStopping...') 
    # httpd.shutdown() 
    stophttp = threading.Thread(target=httpd.shutdown) 
    stophttp.start() 
    stoppool=threading.Thread(target=close_pool) 
    stoppool.daemon=True 
    stoppool.start() 


signal.signal(signal.SIGTERM, term) 
signal.signal(signal.SIGINT, term) 
signal.signal(signal.SIGQUIT, term) 

ठीक काम करता है और हमेशा मैं परीक्षण किया गया।