Multiprocessing Pause-Restart funzionalità usando `event`

Question

Sto usando un codice postato qui sotto per abilitare la funzionalità pausa-riavvio per multiprocessing Pool.

Apprezzerei se mi spieghi perché la variabile event deve essere inviata come argomento alla funzione setup(). Perché allora una variabile globale unpaused è dichiarata all'interno del campo di applicazione setup() funzione e poi si è impostato per essere lo stesso di event variabile:

def setup(event): 
    global unpaused 
    unpaused = event 

Vorrei anche conoscere un logistico alle spalle della seguente dichiarazione:

pool=mp.Pool(2, setup, (event,)) 

Il primo argomento inviato è il numero di core CPU utilizzati da Pool. Il secondo argomento inviato è una funzione setup() che è menzionata sopra.

Perché non dovrebbe tutto essere realizzato come:

global event 
event=mp.Event() 
pool = mp.Pool(processes=2) 

E ogni volta che abbiamo bisogno di mettere in pausa o riavviare un lavoro ci sarebbe solo usare:

Per mettere in pausa:

event.clear() 

Per riavviare:

event.set() 

Perché avremmo bisogno di una variabile globale unpaused? Non capisco! Si prega di avvisare.

---

import time 
import multiprocessing as mp 

def myFunct(arg): 
    proc=mp.current_process() 
    print 'starting:', proc.name, proc.pid,'...\n' 
    for i in range(110): 
     for n in range(500000): 
      pass 
    print '\t ...', proc.name, proc.pid, 'completed\n' 

def setup(event): 
    global unpaused 
    unpaused = event 

def pauseJob(): 
    event.clear() 

def continueJob(): 
    event.set() 


event=mp.Event() 

pool=mp.Pool(2, setup, (event,)) 
pool.map_async(myFunct, [1,2,3]) 

event.set() 

pool.close() 
pool.join() 

Answer 1

sei malinteso come Event opere. Ma prima, parlerò di ciò che sta facendo setup.

La funzione setup viene eseguita in ogni processo figlio all'interno del pool non appena viene avviato. Pertanto, stai impostando una variabile globale denominata event all'interno di ciascun processo in modo che sia lo stesso oggetto multiprocessing.Event creato nel processo principale. Si finisce con ogni processo secondario con una variabile globale chiamata event che fa riferimento allo stesso oggetto multiprocessing.Event. Questo ti permetterà di segnalare i processi figli dal processo principale, proprio come vuoi tu. Vedere questo esempio:

import multiprocessing 

event = None 
def my_setup(event_): 
    global event 
    event = event_ 
    print "event is %s in child" % event 


if __name__ == "__main__": 
    event = multiprocessing.Event() 
    p = multiprocessing.Pool(2, my_setup, (event,)) 
    print "event is %s in parent" % event 
    p.close() 
    p.join() 

uscita:

dan@dantop2:~$ ./mult.py 
event is <multiprocessing.synchronize.Event object at 0x7f93cd7a48d0> in child 
event is <multiprocessing.synchronize.Event object at 0x7f93cd7a48d0> in child 
event is <multiprocessing.synchronize.Event object at 0x7f93cd7a48d0> in parent 

Come si può vedere, è la stessa event nei due processi figli e il genitore. Proprio come vuoi tu.

Tuttavia, il passaggio da event all'impostazione non è necessario.Si può solo eredita l'istanza event dal processo padre:

import multiprocessing 

event = None 

def my_worker(num): 
    print "event is %s in child" % event 

if __name__ == "__main__": 
    event = multiprocessing.Event() 
    pool = multiprocessing.Pool(2) 
    pool.map_async(my_worker, [i for i in range(pool._processes)]) # Just call my_worker for every process in the pool. 

    pool.close() 
    pool.join() 
    print "event is %s in parent" % event 

uscita:

dan@dantop2:~$ ./mult.py 
event is <multiprocessing.synchronize.Event object at 0x7fea3b1dc8d0> in child 
event is <multiprocessing.synchronize.Event object at 0x7fea3b1dc8d0> in child 
event is <multiprocessing.synchronize.Event object at 0x7fea3b1dc8d0> in parent 

Questo è molto più semplice, ed è il modo migliore per passare un semaforo tra genitore e figlio. Infatti, se si dovesse cercare di passare il event direttamente a una funzione operaio, si otterrebbe un errore:

RuntimeError: Semaphore objects should only be shared between processes through inheritance 

Ora, di nuovo a come si sta fraintendendo il modo Event opere. Event è destinato a essere utilizzato in questo modo:

import time 
import multiprocessing 

def event_func(num): 
    print '\t%r is waiting' % multiprocessing.current_process() 
    event.wait() 
    print '\t%r has woken up' % multiprocessing.current_process() 

if __name__ == "__main__": 
    event = multiprocessing.Event() 

    pool = multiprocessing.Pool() 
    a = pool.map_async(event_func, [i for i in range(pool._processes)]) 

    print 'main is sleeping' 
    time.sleep(2) 

    print 'main is setting event' 
    event.set() 

    pool.close() 
    pool.join() 

uscita:

main is sleeping 
    <Process(PoolWorker-1, started daemon)> is waiting 
    <Process(PoolWorker-2, started daemon)> is waiting 
    <Process(PoolWorker-4, started daemon)> is waiting 
    <Process(PoolWorker-3, started daemon)> is waiting 
main is setting event 
    <Process(PoolWorker-2, started daemon)> has woken up 
    <Process(PoolWorker-1, started daemon)> has woken up 
    <Process(PoolWorker-4, started daemon)> has woken up 
    <Process(PoolWorker-3, started daemon)> has woken up 

Come si può vedere, i processi figli hanno bisogno di chiamare esplicitamente event.wait() per loro di essere messi in pausa. Non vengono recapitati quando viene chiamato il numero event.set nel processo principale. Al momento nessuno dei tuoi dipendenti chiama lo event.wait, quindi nessuno di essi potrà mai essere messo in pausa. Suggerisco di dare un'occhiata ai documenti per threading.Event, che replica multiprocessing.Event.