python - 如何访问包含在 Qthread 中的多处理 worker 本身内部的变量？

Question

我正在尝试访问 QThread 中多处理 worker 中的变量。
我做了一个最小的例子来强调我的观点:

from PyQt5.QtGui import *
from PyQt5.QtCore import *
from PyQt5.QtWidgets import *

import sys
import numpy as np
import multiprocessing

class my_Thread(QThread):
    finished = pyqtSignal()

    def __init__(self,M=100, N=100, nbCore=2):
        QThread.__init__(self, )
        self.M = M
        self.N = N
        self.nbCore = nbCore

    def run(self):
        self.my_worker = mp_worker_class()
        self.Mean = self.my_worker.start(self.nbCore, self.M, self.N)
        self.finished.emit()

    def returninfo(self):
        return self.my_worker.nbiter

class mp_worker_class():
    nbiter = 0
    def __init__(self,):
        pass

    @classmethod
    def start(self, nbCore=2, M=100, N=100 ):
        self.nbiter = 0
        X = np.random.rand(nbCore,M,N)
        pipe_list = []
        for i in range(nbCore):
            recv_end, send_end = multiprocessing.Pipe()
            p = multiprocessing.Process(target=self.mp_worker, args=(X[i,:,:] , send_end))
            p.start()
            pipe_list.append(recv_end)

        for idx, recv_end in enumerate(pipe_list):
            Ymean =recv_end.recv()
            print(Ymean)

    @classmethod
    def mp_worker(self, X=None, send_end=None):
        mean = 0
        nb =0
        for i in range(X.shape[0]):
            for j in range(X.shape[1]):
                # print(self.nbiter)
                mean += X[i,j]
                nb += 1
                self.nbiter += 1
        mean /= nb
        send_end.send([mean])


class GUI(QMainWindow):
    def __init__(self, parent=None):
        super(GUI, self).__init__()
        self.parent = parent

        self.centralWidget = QWidget()
        self.setCentralWidget(self.centralWidget)

        self.VBOX = QVBoxLayout()
        self.info_LE = QLineEdit()
        self.start_PB = QPushButton('Start')
        self.start_PB.clicked.connect(self.startThread)
        self.VBOX.addWidget(self.info_LE)
        self.VBOX.addWidget(self.start_PB)

        self.centralWidget.setLayout(self.VBOX)

    def startThread(self):
        self.thread = my_Thread(M=10000, N=10000, nbCore=5)
        self.thread.finished.connect(self.threadFinished)

        self.timer = QTimer()
        self.timer.setInterval(100)
        self.timer.timeout.connect(lambda msg=self.info_LE, x=self.thread: self.updatemsgs(msg, x))
        self.timer.start()
        self.ElapsedTimer = QElapsedTimer()
        self.ElapsedTimer.start()

        self.thread.start()

    def threadFinished(self):
        self.timer.stop()
        self.thread.exit()
        print('Finished')

    def updatemsgs(self, msg, Obj):
        nbiter = Obj.returninfo()
        print(nbiter)
        msg.setText(str(nbiter))
        self.parent.processEvents()

if __name__ == "__main__":
    app = QApplication(sys.argv)
    ex = GUI(app)
    ex.show()
    sys.exit(app.exec())

在这个例子中，我创建了类 my_Thread继承自 Qthread .在此 QThread类我通过类调用多处理 worker mp_worker_class它调用了 5 次函数 mp_worker在平行下。在类(class) mp_worker_class我有一个变量 nbiter = 0每次我在函数 mp_worker 中执行循环时都会增加 1 .我可以验证 nbiter确实在增加，因为我可以通过打印品看到它的值(value)。但是来自 my_Thread.returninfo()我刚刚返回 nbiter 的函数来自 mp_worker_class 的值类，我只得到零。
我想要的是打印 mp_worker_class.nbiter 的值在我可以在 GUI 中看到的 pyqt5 QlineEdit wiget (info_LE) 中。我每 0.1 秒更新一次文本。现在它只打印零。

Answer 1

Python 中的子进程默认不共享内存——在一个进程中运行的代码不能访问或更改另一个进程使用的内存。这意味着每个进程都有自己正在使用的每个变量的副本，包括 mp_worker_class.nbiter多变的。因此，您看不到子进程对其 mp_worker_class.nbiter 所做的更改。来自父进程(或任何其他子进程，就此而言)的变量。
如您所见，我们可以使用 args 从父进程获取数据到子进程。 multiprocessing.Process 的关键字参数构造函数。但是，这只是将数据从父级复制到子级；我们仍然没有在两个进程之间共享内存。

import multiprocessing

def my_example(arg):
    arg.append(35)
    print("Child arg:",arg)
    
if __name__ == "__main__":
    l = [1,2,3]
    print("Before:",l)
    p = multiprocessing.Process(target=my_example, args=(l,))
    p.start()
    p.join()
    print("After:",l)
    
# Before: [1, 2, 3]
# Child arg: [1, 2, 3, 35]
# After: [1, 2, 3]

幸运的是， multiprocessing提供 Value class这使得在共享内存中创建变量变得容易。关键是创建 Value在父进程中，然后分发 Value通过 args 到子进程 multiprocessing.Process 的参数.

在您的代码中，您可以创建您的 multiprocessing.Value在 my_Thread 的构造函数中.例如，您可以添加

self.nbiter = multiprocessing.Value('i',0)

这将创建一个整数 multiprocessing.Value (这就是 i 的意思)并将其初始化为 0。然后你可以传递这个 Value到 self.my_worker.start classmethod，反过来可以通过 Value给它的 child mp_worker过程。
与 multiprocessing.Value 关联的原始值对象可以通过其 value 访问属性。因此，您需要更改 mp_worker 中的代码。 classmethod 更改 value您的 multiprocessing.Value 的属性目的。
您还需要考虑 += 的事实。使用 multiprocessing.Value 时操作不会是原子的.因此，您的代码需要获取对 Value 的锁。在递增之前。如果为 mp_worker 创建了一个新参数叫 nbiter ，您的递增代码 nbiter应该是这样的。

with nbiter.get_lock():
    nbiter.value += 1

您还需要更改您的 my_Thread.returninfo方法简单地返回 self.nbiter.value .您可能还想设置 self.nbiter.value = 0开头你的 my_Thread.run方法如果 my_Thread由于某种原因将重新启动。
总之，您的代码可能如下所示。

from PyQt5.QtGui import *
from PyQt5.QtCore import *
from PyQt5.QtWidgets import *

import sys
import numpy as np
import multiprocessing

class my_Thread(QThread):
    finished = pyqtSignal()

    def __init__(self,M=100, N=100, nbCore=2):
        QThread.__init__(self, )
        self.M = M
        self.N = N
        self.nbCore = nbCore
        self.nbiter = multiprocessing.Value('i',0)

    def run(self):
        self.nbiter.value = 0
        self.my_worker = mp_worker_class()
        self.Mean = self.my_worker.start(self.nbCore, self.M, self.N, self.nbiter)
        self.finished.emit()

    def returninfo(self):
        return self.nbiter.value

class mp_worker_class():
    def __init__(self,):
        pass

    @classmethod
    def start(self, nbCore=2, M=100, N=100, nbiter=None ):
        X = np.random.rand(nbCore,M,N)
        pipe_list = []
        for i in range(nbCore):
            recv_end, send_end = multiprocessing.Pipe()
            p = multiprocessing.Process(target=self.mp_worker, args=(X[i,:,:] , send_end, nbiter))
            p.start()
            pipe_list.append(recv_end)

        for idx, recv_end in enumerate(pipe_list):
            Ymean =recv_end.recv()
            print(Ymean)

    @classmethod
    def mp_worker(self, X=None, send_end=None, nbiter=None):
        mean = 0
        nb =0
        for i in range(X.shape[0]):
            for j in range(X.shape[1]):
                # print(self.nbiter)
                mean += X[i,j]
                nb += 1
                with nbiter.get_lock():
                    nbiter.value += 1
        mean /= nb
        send_end.send([mean])


class GUI(QMainWindow):
    def __init__(self, parent=None):
        super(GUI, self).__init__()
        self.parent = parent

        self.centralWidget = QWidget()
        self.setCentralWidget(self.centralWidget)

        self.VBOX = QVBoxLayout()
        self.info_LE = QLineEdit()
        self.start_PB = QPushButton('Start')
        self.start_PB.clicked.connect(self.startThread)
        self.VBOX.addWidget(self.info_LE)
        self.VBOX.addWidget(self.start_PB)

        self.centralWidget.setLayout(self.VBOX)

    def startThread(self):
        self.thread = my_Thread(M=10000, N=10000, nbCore=5)
        self.thread.finished.connect(self.threadFinished)

        self.timer = QTimer()
        self.timer.setInterval(100)
        self.timer.timeout.connect(lambda msg=self.info_LE, x=self.thread: self.updatemsgs(msg, x))
        self.timer.start()
        self.ElapsedTimer = QElapsedTimer()
        self.ElapsedTimer.start()

        self.thread.start()

    def threadFinished(self):
        self.timer.stop()
        self.thread.exit()
        print('Finished')

    def updatemsgs(self, msg, Obj):
        nbiter = Obj.returninfo()
        print(nbiter)
        msg.setText(str(nbiter))
        self.parent.processEvents()

if __name__ == "__main__":
    app = QApplication(sys.argv)
    ex = GUI(app)
    ex.show()
    sys.exit(app.exec())