Python multiprocessing——跟踪pool.map操作的过程

Question

我有一个函数可以执行一些模拟和返回字符串格式的数组。

我想运行模拟(功能)不同的输入参数值，超过 10000 个可能的输入值，并将结果写入单个文件。

我正在使用多处理，特别是 pool.map 函数并行运行模拟。

全程模拟功能运行超过10000次需要很长时间，我很想跟踪整个操作过程。

我认为下面当前代码中的问题是，pool.map 运行函数 10000 次，在这些操作期间没有任何进程跟踪。一旦并行处理完成运行 10000 次模拟(可能是几小时到几天)，然后我会继续跟踪 10000 次模拟结果何时被保存到文件中。所以这并不是真正跟踪 pool.map 操作的处理。

我的代码是否有允许进程跟踪的简单修复？

def simFunction(input):
    # Does some simulation and outputs simResult
    return str(simResult)

# Parallel processing

inputs = np.arange(0,10000,1)

if __name__ == "__main__":
    numCores = multiprocessing.cpu_count()
    pool = multiprocessing.Pool(processes = numCores)
    t = pool.map(simFunction, inputs) 
    with open('results.txt','w') as out:
        print("Starting to simulate " + str(len(inputs)) + " input values...")
        counter = 0
        for i in t:
            out.write(i + '\n')
            counter = counter + 1
            if counter%100==0:
                print(str(counter) + " of " + str(len(inputs)) + " input values simulated")
    print('Finished!!!!')

Answer 1

请注意，我使用的是 pathos.multiprocessing 而不是 multiprocessing。 它只是 multiprocessing 的一个分支使您能够使用多个输入执行 map 函数，具有更好的序列化，并允许您在任何地方执行 map 调用(不仅仅是在 __main__ ).您也可以使用 multiprocessing 来执行以下操作，但代码会略有不同。

如果您使用迭代的 map 函数，跟踪进度非常容易。

from pathos.multiprocessing import ProcessingPool as Pool
def simFunction(x,y):
  import time
  time.sleep(2)
  return x**2 + y
 
x,y = range(100),range(-100,100,2)
res = Pool().imap(simFunction, x,y)
with open('results.txt', 'w') as out:
  for i in x:
    out.write("%s\n" % res.next())
    if i%10 is 0:
      print "%s of %s simulated" % (i, len(x))

0 of 100 simulated
10 of 100 simulated
20 of 100 simulated
30 of 100 simulated
40 of 100 simulated
50 of 100 simulated
60 of 100 simulated
70 of 100 simulated
80 of 100 simulated
90 of 100 simulated

或者，您可以使用异步 map。在这里，我会做一些不同的事情，只是为了混合起来。

import time
res = Pool().amap(simFunction, x,y)
while not res.ready():
  print "waiting..."
  time.sleep(5)
 

waiting...
waiting...
waiting...
waiting...

res.get()
[-100, -97, -92, -85, -76, -65, -52, -37, -20, -1, 20, 43, 68, 95, 124, 155, 188, 223, 260, 299, 340, 383, 428, 475, 524, 575, 628, 683, 740, 799, 860, 923, 988, 1055, 1124, 1195, 1268, 1343, 1420, 1499, 1580, 1663, 1748, 1835, 1924, 2015, 2108, 2203, 2300, 2399, 2500, 2603, 2708, 2815, 2924, 3035, 3148, 3263, 3380, 3499, 3620, 3743, 3868, 3995, 4124, 4255, 4388, 4523, 4660, 4799, 4940, 5083, 5228, 5375, 5524, 5675, 5828, 5983, 6140, 6299, 6460, 6623, 6788, 6955, 7124, 7295, 7468, 7643, 7820, 7999, 8180, 8363, 8548, 8735, 8924, 9115, 9308, 9503, 9700, 9899]

无论是迭代还是异步 map 都可以让您编写任何您想要的代码来更好地进行流程跟踪。例如，将唯一的“id”传递给每个作业，然后观察哪个返回，或者让每个作业返回它的进程 id。有很多方法可以跟踪进度和流程……但以上内容应该可以让您入门。

你可以获得pathos here .