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

Posted 2023-02-16

【中文标题】处理对象方法的Python3多处理池未获取对象的更新数据

【英文标题】：Python3 multiprocessing pool working on an objects method does not get updated data of object

【发布时间】：2021-12-27 09:50:28

【问题描述】：

我有正在更新的 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)

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

【参考方案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方法

【讨论】：

我只是碰巧偶然发现了这个“答案”（这确实应该是对您的问题的更新，它提供了更多详细信息，因为这不能解决您的问题），因为您从未回复我对 我的答案，所以我不知道这一点。现在我看到您在原始帖子中遗漏了一个重要的细节，即您甚至没有使用标准 Python 库中的任何一个多处理包，而是使用 pathos，对此我一无所知，这就是为什么您应该 总是在一开始就发布一个最小的、可重复的例子。

【参考方案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()

【讨论】：

感谢您的回答。是的，起初这看起来像是一个解决方案，我发现我自己，但问题是多处理库不能腌制对象。出于这个原因，我做了一些研究，发现可以使用多处理库的 Pathos (github.com/uqfoundation/pathos) 分支。但是我仍然有同样的泡菜错误：“TypeError：不能泡菜'cv2.TrackerKCF'对象。安装Pathos模块的依赖项似乎出了点问题。...

一般来说，当然，多处理库可以腌制对象——也许不能你的对象。但我不明白的是，如果你的工作函数是t.update，你必须能够将t从主进程pickle到处理池中才能执行update方法。那么update 是否添加了一个最初在t 中不存在的属性，现在使它无法被选中以返回它？这就是您不发布minimal, reproducible example 时的问题。