python - Python中的多处理、传递多个参数以及进程之间的数据同步

Question

我正在尝试使用多处理和池在Python中进行并行处理。下面给出了并行处理的代码。

 all_rects = [[[] for x in range(net_config["grid_width"])] for y in range(net_config["grid_height"])]
 for n in range(len(bbox_list)):
     for k in range(net_config["grid_height"] * net_config["grid_width"]):
         conf = conf_list[n][k,1].flatten()[0]
         if conf > 0.75:
             y = int(k / net_config["grid_width"])
             x = int(k % net_config["grid_width"])
             bbox = bbox_list[n][k]

             abs_cx = pix_per_w/2 + pix_per_w*x + int(bbox[0,0,0])
             abs_cy = pix_per_h/2 + pix_per_h*y+int(bbox[1,0,0])
             w = bbox[2,0,0]
             h = bbox[3,0,0]
             all_rects[y][x].append(Rect(abs_cx,abs_cy,w,h,conf))

len(bbox_list) 每次迭代都不同。我喜欢根据len(bbox_list)的数量设置进程的数量。

所以我有一个函数

def doWork(box_list, conf_list, all_rects, w, h, n, pix_per_w):
    for k in range(h * w):
        conf = conf_list[n][k,1].flatten()[0]
        if conf > 0.75:
           y = int(k / w)
           x = int(k % w)
           bbox = bbox_list[n][k]
           abs_cx = pix_per_w/2 + pix_per_w*x + int(bbox[0,0,0])
           abs_cy = pix_per_h/2 + pix_per_h*y+int(bbox[1,0,0])
           w = bbox[2,0,0]
           h = bbox[3,0,0]
           all_rects[y][x].append(Rect(abs_cx,abs_cy,w,h,conf))

然后从main开始，我进行并行处理

all_rects = [[[] for x in range(net_config["grid_width"])] for y in range(net_config["grid_height"])]
pool = multiprocessing.Pool(len(bbox_list))
pool.map(doWork, (bbox_list, conf_list, all_rects, net_config["grid_width"], net_config["grid_height"], [len(bbox_list)], pix_per_w))

我有一个错误(TypeError:doWork() 恰好需要 7 个参数(给定 1 个))，另一个问题是 all_rects。 all_rects 将是三维数组，all_rects 的数据同步如何，多处理安全吗？

编辑:

def doWork(worker_args, b_range):
    box_list = worker_args[0]
    conf_list = worker_args[1]
    all_rects = worker_args[2]
    w = worker_args[3]
    h = worker_args[4]    
    pix_per_w = worker_args[5]
    n = b_range
    for k in range(h * w):
        conf = conf_list[n][k,1].flatten()[0]
        if conf > 0.75:
           y = int(k / w)
           x = int(k % w)
           bbox = bbox_list[n][k]
           abs_cx = pix_per_w/2 + pix_per_w*x + int(bbox[0,0,0])
           abs_cy = pix_per_h/2 + pix_per_h*y+int(bbox[1,0,0])
           w = bbox[2,0,0]
           h = bbox[3,0,0]
           all_rects[y][x].append(Rect(abs_cx,abs_cy,w,h,conf))

然后从 main 开始，

         all_rects = [[[] for x in range(net_config["grid_width"])] for y in range(net_config["grid_height"])]
         pool = multiprocessing.Pool(len(bbox_list))
         box_ranges = range(len(bbox_list))
         worker_args = [bbox_list, conf_list, all_rects, net_config["grid_width"], net_config["grid_height"], pix_per_w]
         pool.map(doWork, worker_args, l) for l in box_ranges

我尝试通过元组发送，但出现语法无效的错误

pool.map(doWork, worker_args, l) for l in box_ranges

谢谢

Answer 1

您可以编写 worker 函数来接受元组作为参数并在函数中扩展该元组，例如:

def doWork(worker_args):
  bbox_list, conf_list, all_rects, w, h, n, pix_per_w = worker_args

对于您关心的all_rects，据我所知，append 是线程安全的。请参阅Are lists thread-safe

编辑:

Pool.map 与常规 map 函数非常相似。它接收两个参数，第一个是工作程序函数的名称，第二个应该是传递给工作程序的参数的迭代器。所以你的第二个参数应该是元组列表。每个元组将作为 doWork 函数的参数传递给一个进程:

pool.map(doWork, [(bbox_list, conf_list, all_rects, w, h, n, pix_per_w) for _ in xrange(len(bbox_list))])