Как отслеживать состояние при многопроцессорности и pool.map?

Question

Я настраиваю модуль многопроцессорной в первый раз, и в основном, я планирую сделать что-то вдоль линий

from multiprocessing import pool 
pool = Pool(processes=102) 
results = pool.map(whateverFunction, myIterable) 
print 1 

Как я понимаю, 1 будет напечатано, как только все процессы вернулись, и результаты были полными. Я хотел бы получить некоторое обновление статуса. Каков наилучший способ реализации этого?

Я как бы не решаюсь сделать whateverFunction() печать. Особенно, если есть около 200 значений, я собираюсь сделать что-то вроде «процесса», напечатанного 200 раз, что не очень полезно.

Я ожидаю выход как

10% of myIterable done 
20% of myIterable done 

Answer 1

pool.map блоков, пока все параллельные вызовы функций не завершены. pool.apply_async не блокируется. Кроме того, вы можете использовать его параметр callback , чтобы сообщить о прогрессе. Функция обратного вызова, log_result, вызывается один раз каждый раз, когда foo завершает работу. Получено значение, возвращаемое foo.

from __future__ import division 
import multiprocessing as mp 
import time 

def foo(x): 
    time.sleep(0.1) 
    return x 

def log_result(retval): 
    results.append(retval) 
    if len(results) % (len(data)//10) == 0: 
     print('{:.0%} done'.format(len(results)/len(data))) 

if __name__ == '__main__': 
    pool = mp.Pool() 
    results = [] 
    data = range(200) 
    for item in data: 
     pool.apply_async(foo, args=[item], callback=log_result) 
    pool.close() 
    pool.join() 
    print(results) 

дает

10% done 
20% done 
30% done 
40% done 
50% done 
60% done 
70% done 
80% done 
90% done 
100% done 
[0, 1, 2, 3, ..., 197, 198, 199] 

---

log_result выше функция изменяет глобальную переменную results и получает доступ к глобальной переменной data. Вы не можете передать эти переменные в log_result, поскольку функция обратного вызова, указанная в pool.apply_async, равна , которая всегда вызывается только с одним аргументом, возвращаемое значение foo.

Вы можете, однако, сделать замыкание, что, по крайней мере ясно, какие переменные log_result зависит от:

from __future__ import division 
import multiprocessing as mp 
import time 

def foo(x): 
    time.sleep(0.1) 
    return x 

def make_log_result(results, len_data): 
    def log_result(retval): 
     results.append(retval) 
     if len(results) % (len_data//10) == 0: 
      print('{:.0%} done'.format(len(results)/len_data)) 
    return log_result 

if __name__ == '__main__': 
    pool = mp.Pool() 
    results = [] 
    data = range(200) 
    for item in data: 
     pool.apply_async(foo, args=[item], callback=make_log_result(results, len(data))) 
    pool.close() 
    pool.join() 
    print(results)