Jak pobrać dane wyjściowe z wieloprocesowości w Pythonie?

Question

Próbuję przyśpieszyć jedną rutynę, używając modułu Multiprocessing w Pythonie. Chcę być w stanie przeczytać kilka plików .csv przez podział pracy pomiędzy kilku rdzeni, na które mam:

def csvreader(string): 

    from numpy import genfromtxt; 

    time,signal=np.genfromtxt(string, delimiter=',',unpack="true") 
    return time,signal 

Potem wywołać tę funkcję, mówiąc:

if __name__ == '__main__': 
    for i in range(0,2): 
     p = multiprocessing.Process(target=CSVReader.csvreader, args=(string_array[i],)) 
     p.start() 

Chodzi o to, że nie przechowuje żadnych danych wyjściowych. Przeczytałem wszystkie fora online i widziałem, że może istnieć sposób na multiprocessing.queue, ale nie rozumiem tego całkiem dobrze. Czy istnieje prosta i bezpośrednia metoda?

Answer 1

Twój najlepszy zakład to multiprocessing.Queue lub multiprocessing.Pipe, które są przeznaczone właśnie dla tego problemu. Umożliwiają wysyłanie danych między procesami w bezpieczny i łatwy sposób.

Jeśli chcesz zwrócić dane wyjściowe funkcji csvreader, powinieneś przekazać mu inny argument, czyli multiprocessing.Queue, dzięki któremu dane zostaną przesłane z powrotem do głównego procesu. Zamiast wartości, umieść je w kolejce, a proces główny pobierze je w pewnym momencie później. Jeśli nie jesteś gotowy, gdy proces próbuje je zdobyć, domyślnie będzie to po prostu blokować (oczekiwanie), dopóki nie są dostępne

Twoja funkcja będzie teraz wyglądać tak:

def cvsreader(string, q): 
    q.put(np.genfromtxt(string, delimiter=',', unpack="true")) 

Głównym by rutyna być:

if __name__ == '__main__' 
    q = multiprocessing.Queue() 
    for i in range(2): 
     p = multiprocessing.Process(target=csvreader, args=(string_array[i], q,)) 
     p.start() 

# Do anything else you need in here 

time=np.empty(2,dtype='object') 
signal=np.empty(2,dtype='object') 
for i in range(2): 
    time[i], signal[i] = q.get() # Returns output or blocks until ready 
    # Process my output 

należy pamiętać, że trzeba zadzwonić Queue.get() dla każdy element, który chcesz wrócić. Aby uzyskać więcej przykładów i informacji, zapoznaj się z dokumentacją na stronie multiprocessing module.

Answer 2

Używając przykład z the introduction to the documentation:

if __name__ == '__main__': 
    pool = Pool(2) 
    results = pool.map(CSVReader.csvreader, string_array[:2]) 
    print(results)