Pula wieloprocesorowa Pythona kończy

Question

Pracuję nad renderfarm, i potrzebuję moich klientów, aby móc uruchomić wiele wystąpień renderer, bez blokowania, aby klient mógł otrzymywać nowe polecenia. Działa to poprawnie, ale mam problemy z zakończeniem tworzonych procesów.

Na poziomie globalnym, określam moją basen (tak, że mogę do niego dostęp z dowolnej funkcji):

p = Pool(2) 

I wtedy zadzwonić do mojego renderujący z apply_async:

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

tej funkcji wykończeń , uruchamia procesy w tle i czeka na nowe polecenia. Zrobiłem proste polecenia, które będą zabijać klienta i zatrzymać renderuje:

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

Nie wydaje się, aby dać błąd (to wydrukować błąd), kończy python ale procesy tła są nadal bieganie. Czy ktoś może polecić lepszy sposób kontrolowania uruchomionych programów?

Answer 1

znaleźć odpowiedź na moje własne pytanie. Głównym problemem było to, że dzwoniłem do aplikacji innej firmy niż funkcji. Kiedy wywołuję podproces [albo za pomocą funkcji call() lub Popen()], tworzy on nową instancję pythona, której jedynym celem jest wywołanie nowej aplikacji. Jednak po wyjściu pytona zabije to nowe wystąpienie Pythona i pozostawi aplikację uruchomioną.

Rozwiązaniem jest zrobić to na własnej skórze, znajdując pid procesu Pythona, który jest tworzony, pobierając dzieci z tego pid i zabijając je. Ten kod jest specyficzny dla osx; jest prostszy kod (który nie opiera się na grep) dostępny dla Linuksa.

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

Jeśli nadal występuje ten problem, można spróbować przeprowadzić symulację Pool z daemonic processes (zakładając, że uruchamiasz pulę/procesy z procesu innego niż demoniczny). Wątpię, aby to było najlepsze rozwiązanie, ponieważ wydaje się, że twoje procesy powinny wyjść, ale to wszystko, co mogłem wymyślić. Nie wiem, co robi twój callback, więc nie jestem pewien, gdzie umieścić to w moim przykładzie poniżej.

Proponuję także stara się stworzyć swój Pool w __main__ powodu mojego doświadczenia (i docs) z niesamowitość występujące podczas procesów zrodził globalnie. Jest to szczególnie prawdziwe, jeśli używasz systemu Windows: 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

znalazłem rozwiązanie: stop basen w osobnym wątku, tak:

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) 

działa dobrze i zawsze i przetestowane.