处理对象方法的Python3多处理池未获取对象的更新数据

Question

我有正在更新的 openCV 跟踪器对象。为了让事情变得更快，我使用了带有 map_async 函数的多处理池来并行工作。它按预期工作，我得到了显着的加速。但是一件事很奇怪/不起作用。当我用 openCV 重新初始化我的跟踪器时（意味着给它们一个新的边界框），跟踪器没有更新，而是继续使用之前的边界框。这仅在使用多处理池时发生，但在跟踪器列表上使用顺序循环时不会发生。我怀疑该过程会制作自己的跟踪器对象副本，因此跟踪器的重新初始化不适用于该新创建的对象。但是据我了解，当我调用函数 map_async 时，会创建一个新进程，并且使用 process.wait() 它正在等待该进程完成其工作。

我已经尝试在每次调用 updateTrackers() 函数时创建一个新池。这并没有解决问题。

工作重新初始化顺序代码：

    def updateTrackers(self, frame):
        for t in self.trackers:
            t.update()

不工作的重新初始化顺序代码：

    def updateTrackers(self, frame):
        processes = []
        # create a process for each tracker
        for t in self.trackers:
            processes.append(self.ProcessPool.map_async(t.update, (frame, )))

        # wait for the processes to finish
        for p in processes:
            p.wait()

两种情况下重新初始化跟踪器对象的代码都是一样的：

    def reInitTracker(self, index, frame):
        if index >= self.nmbTrackers:
            return

        initBB = cv2.selectROI("Camera view", frame, fromCenter=False,
                showCrosshair=True)
        self.trackers[index].tracker.clear()
        self.trackers[index].tracker = cv2.TrackerKCF_create()
        self.trackers[index].tracker.init(frame, initBB)

编辑： 我刚刚发现跟踪器在并行化它们时没有得到更新。现在更一致的是它们也没有被初始化。

Answer 1

参考您的 BooBoos 评论：

不更新方法不添加任何属性。你是对的，这确实很奇怪。我创建了一个最小的、可重现的示例，它会导致相同的错误：

import cv2
import numpy as np
from pathos.multiprocessing import ProcessPool

tracker = cv2.TrackerKCF_create()
Pool = ProcessPool()

# dummy values
frame = np.array((100,100,3))
initBB = (1,2,3,4)
tracker.init(frame, initBB)

p = Pool.apipe(tracker.update, frame)
p.get()

问题似乎是get方法

Answer 2

如果我了解您的问题，当您在调用 map_async 时将 t.update 作为工作函数传递时，t（跟踪器）对象不会更新

是的，t 将被序列化/反序列化到多处理池以执行update 方法，如果t 的状态被update 修改，它将不会反映回主进程。顺便说一句，像您正在做的那样，使用由单个元素组成的 iterable 参数调用 map 方法是相当不寻常的； apply_async 会更合适。

解决方案是让您的update 方法（不幸的是，您没有显示）返回self，然后使用此返回值更新主进程中的跟踪器对象：

    def updateTrackers(self, frame):
        results = [self.ProcessPool.apply_async(t.update, (frame,)) for t in self.trackers]
        # It is assumed that t.update now returns with t.self
        # It is also assumed that self.trackers is a list or otherwise indexable:
        for idx, result in enumerate(results):
            # Wait for task to finish and update tracker with returned value:
            self.trackers[idx] = result.get()