многопроцессорная совместная память в pyqt5 python 3.8

Question

Я немного запутался, почему следующее не обновляет массив numpy между процессами? В этом нет ошибок, но основной процесс не отражает обновления дочернего элемента в общий массив sample. Я не уверен, почему это не работает. Пример на странице shared_memory отлично работает на моей машине. Общие события также работают нормально. Возможно, это проблема, связанная с Qt?

import time
from PyQt5.QtCore import QTimer
from PyQt5.QtWidgets import QWidget, QVBoxLayout, QProgressBar, QPushButton
import multiprocessing
from multiprocessing import shared_memory
import numpy as np

DATA_SIZE = 1
DTYPE = np.int64


class DataPuller(multiprocessing.get_context("spawn").Process):
    def __init__(
            self,
            event,
            shared_mem_name,
    ):
        super().__init__()
        self.event = event
        self.shm = shared_memory.SharedMemory(name=shared_mem_name)
        self.sample = np.ndarray(
            shape=(DATA_SIZE, 1), dtype=DTYPE, buffer=self.shm.buf)
        self.counter = 0

    def run(self):
        print("child process started")
        while not self.event.is_set():
            time.sleep(0.1)
            self.sample[:, 0] = self.counter
            print("child: ", self.sample[0, 0])
            self.counter += 1
            if self.counter >= 1000:
                self.counter = 0
        self.shm.close()
        print("child process finished")


class Window(QWidget):
    def __init__(self, *args, **kwargs):
        super(Window, self).__init__(*args, **kwargs)
        layout = QVBoxLayout(self)
        self.progressBar = QProgressBar(self)
        self.timer = QTimer(self)
        self.timer.timeout.connect(self.onTimer)
        self.progressBar.setRange(0, 1000)
        layout.addWidget(self.progressBar)
        layout.addWidget(QPushButton('开启线程', self, clicked=self.onStart))

        self.counter = 0
        self.shared_mem_name = "data"
        self.base_array = np.zeros((DATA_SIZE, 1), dtype=DTYPE)
        self.shm = shared_memory.SharedMemory(
            create=True,
            size=self.base_array.nbytes,
            name=self.shared_mem_name)
        self.sample = np.ndarray(
            shape=self.base_array.shape,
            dtype=self.base_array.dtype,
            buffer=self.shm.buf)
        self.event = multiprocessing.Event()
        self._process = DataPuller(self.event, self.shared_mem_name)

    def onStart(self):
        if not self._process.is_alive():
            print("main starting process")
            self._process.start()
            self.timer.start(500)
        else:
            pass

    def onTimer(self):
        print("main: ", self.sample[0, 0])
        self.progressBar.setValue(self.sample[0, 0])

    def closeEvent(self, event):
        if self._process.is_alive():
            self.event.set()
            self._process.join()
        self.shm.close()
        self.shm.unlink()
        self.close()
        print("main process finished")


if __name__ == '__main__':
    import sys
    from PyQt5.QtWidgets import QApplication
    app = QApplication(sys.argv)
    w = Window()
    w.show()
    sys.exit(app.exec_())

Answer 1

Проблема в том, что Событие не относится к контексту DataPuller, поэтому оно никогда не войдет в то время. Решение заключается в создании события с использованием контекста.

Проблема заключается в том, что метод DataPuller __init__ выполняется в начальном процессе, когда создается общая память, поэтому он недоступен для вторичного процесса DataPuller. Решением в этом случае является создание разделяемой памяти в методе run.

import time
from PyQt5.QtCore import QTimer
from PyQt5.QtWidgets import QWidget, QVBoxLayout, QProgressBar, QPushButton
import multiprocessing
from multiprocessing import shared_memory
import numpy as np

DATA_SIZE = 1
DTYPE = np.int64

context = multiprocessing.get_context("spawn")


class DataPuller(context.Process):
    def __init__(
        self, event, shared_mem_name,
    ):
        super().__init__()
        self.event = event
        self.shared_mem_name = shared_mem_name

    def run(self):
        self.shm = shared_memory.SharedMemory(name=self.shared_mem_name)
        self.sample = np.ndarray(shape=(DATA_SIZE, 1), dtype=DTYPE, buffer=self.shm.buf)
        self.counter = 0

        while not self.event.is_set():
            time.sleep(0.1)
            self.sample[:, 0] = self.counter
            print("child: ", self.sample[0, 0])
            self.counter += 1
            if self.counter >= 1000:
                self.counter = 0
        self.shm.close()
        print("child process finished")


class Window(QWidget):
    def __init__(self, *args, **kwargs):
        super(Window, self).__init__(*args, **kwargs)
        layout = QVBoxLayout(self)
        self.progressBar = QProgressBar()
        self.timer = QTimer(self)
        self.timer.timeout.connect(self.onTimer)
        self.progressBar.setRange(0, 1000)
        layout.addWidget(self.progressBar)
        layout.addWidget(QPushButton("开启线程", clicked=self.onStart))

        self.counter = 0
        self.shared_mem_name = "data"
        self.base_array = np.zeros((DATA_SIZE, 1), dtype=DTYPE)
        self.shm = shared_memory.SharedMemory(
            create=True, size=self.base_array.nbytes, name=self.shared_mem_name
        )
        self.sample = np.ndarray(
            shape=self.base_array.shape,
            dtype=self.base_array.dtype,
            buffer=self.shm.buf,
        )
        self.event = context.Event()
        print(self.event.is_set())
        self._process = DataPuller(self.event, self.shared_mem_name)

    def onStart(self):
        if not self._process.is_alive():
            print("main starting process")
            self._process.start()
            self.timer.start(500)
        else:
            pass

    def onTimer(self):
        print("main: ", self.sample[0, 0])
        self.progressBar.setValue(self.sample[0, 0])

    def closeEvent(self, event):
        if self._process.is_alive():
            self.event.set()
            self._process.join()
        self.shm.close()
        self.shm.unlink()
        self.close()
        print("main process finished")


if __name__ == "__main__":
    import sys
    from PyQt5.QtWidgets import QApplication

    app = QApplication(sys.argv)
    w = Window()
    w.show()
    sys.exit(app.exec_())