python multiprocessing pool terminate

Question

Sto lavorando su un renderfarm e ho bisogno che i miei client possano avviare più istanze di un renderer, senza bloccare in modo che il client possa ricevere nuovi comandi. Ho funzionato correttamente, tuttavia ho problemi a terminare i processi creati.

A livello globale, definisco la mia piscina (in modo che io possa accedere da qualsiasi funzione):

p = Pool(2) 

Ho poi chiamo il mio renderer con apply_async:

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

quella funzione finiture , avvia i processi in background e attende nuovi comandi. Ho fatto un semplice comando che ucciderà il cliente e fermare la rende:

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

Non sembra dare un errore (che sarebbe stampare l'errore), le termina Python, ma i processi in background sono ancora in esecuzione. Qualcuno può raccomandare un modo migliore per controllare questi programmi lanciati?

Answer 1

Trovato la risposta alla mia domanda. Il problema principale era che stavo chiamando un'applicazione di terze parti piuttosto che una funzione. Quando chiamo il sottoprocesso [usando call() o Popen()] si crea una nuova istanza di python il cui unico scopo è chiamare la nuova applicazione. Tuttavia, quando Python termina, ucciderà questa nuova istanza di python e lascerà l'applicazione in esecuzione.

La soluzione è di farlo nel modo più difficile, trovando il pid del processo python che viene creato, ottenendo i figli di quel pid e uccidendoli. Questo codice è specifico per osx; c'è un codice più semplice (che non si basa su grep) disponibile per linux.

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

Se si riscontra ancora questo problema, è possibile provare a simulare uno Pool con daemonic processes (presupponendo che si stia avviando il pool/i processi da un processo non demonico). Dubito che questa sia la soluzione migliore dal momento che sembra che i tuoi processi Pool debbano essere chiusi, ma questo è tutto ciò che potrei inventare. Non so come funziona la tua richiamata, quindi non sono sicuro di dove metterla nel mio esempio qui sotto.

Inoltre, suggerisco di provare a creare il numero Pool in __main__ a causa della mia esperienza (e dei documenti) con stranezze che si verificano quando i processi vengono generati globalmente. Questo è particolarmente vero se sei su 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

ho trovato la soluzione: smettere di piscina in thread separato, in questo modo:

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) 

funziona bene e sempre ho provato.