尝试在发送进程运行Tkinter的进程之间通过管道发送任何内容时出现管道中断错误

问题描述

我正在使用multiprocessing模块(Python3.8)中的Pipe和Process。我的初始程序如下所示：

from multiprocessing import Process, Pipe


class Process1(object):
    def __init__(self, pipe_out):
        self.pipe_out = pipe_out

        self.run()

    def run(self):
        try:
            while True:
                print("Sending message to process 2")
                self.pipe_out.send(["hello"])
        except KeyboardInterrupt:
            pass


class Process2(object):
    def __init__(self, pipe_in):
        self.pipe_in = pipe_in

        self.run()

    def run(self):
        try:
            while self.pipe_in.poll():
                request = self.pipe_in.recv()
                method = request[0]
                args = request[1:]

                try:
                    getattr(self, method + "_callback")(*args)
                except AttributeError as ae:
                    print("Unknown callback received from pipe", str(ae))

            print("Process 2 done with receiving")
        except KeyboardInterrupt:
            pass

    def hello_callback(self):
        print("Process 1 said hello")


class Controller(object):
    def __init__(self):
        pipe_proc1_out, pipe_proc2_in = Pipe()

        self.proc1 = Process(
            target=Process1,
            args=(pipe_proc1_out, )
        )

        self.proc2 = Process(
            target=Process2,
            args=(pipe_proc2_in, )
        )

    def run(self):
        try:
            self.proc1.start()
            self.proc2.start()

            while True:
                continue
        except KeyboardInterrupt:
            print("Quitting processes...")
            self.proc1.join(1)
            if self.proc1.is_alive():
                self.proc1.terminate()

            self.proc2.join(1)
            if self.proc2.is_alive():
                self.proc2.terminate()

            print("Finished")


def pipes():
    c = Controller()
    c.run()


if __name__ == "__main__":
    pipes()

我有一个Controller实例，它一直运行到收到键盘中断。它还处理Process1和Process2两个进程，前者不断发送，后者不断接收。

上面的代码是一个更大的项目的框架，涉及复杂的图形用户界面(PySide)、图像处理(OpenCV)和游戏引擎(Panda3D)。因此，我尝试将Tkinter添加为一个图形用户界面示例：

from multiprocessing import Process, Pipe
import tkinter as tk


class Process1(tk.Frame):
    def __init__(self, pipe_out):
        self.pipe_out = pipe_out

        self.setup_gui()
        self.run()

    def setup_gui(self):
        self.app = tk.Tk()
        lb1 = tk.Label(self.app, text="Message:")
        lb1.pack()
        self.ent1 = tk.Entry(self.app)
        self.ent1.pack()
        btn1 = tk.Button(self.app, text="Say hello to other process",
                         command=self.btn1_clicked)
        btn1.pack()

    def btn1_clicked(self):
        msg = self.ent1.get()
        self.pipe_out.send(["hello", msg])

    def run(self):
        try:
            self.app.mainloop()
        except KeyboardInterrupt:
            pass


class Process2(object):
    def __init__(self, pipe_in):
        self.pipe_in = pipe_in

        self.run()

    def run(self):
        try:
            while self.pipe_in.poll():
                request = self.pipe_in.recv()
                method = request[0]
                args = request[1:]

                try:
                    getattr(self, method + "_callback")(*args)
                except AttributeError as ae:
                    print("Unknown callback received from pipe", str(ae))

            print("Process 2 done with receiving")
        except KeyboardInterrupt:
            pass

    def hello_callback(self, msg):
        print("Process 1 say"" + msg + """)


class Controller(object):
    def __init__(self):
        pipe_proc1_out, pipe_proc2_in = Pipe()

        self.proc1 = Process(
            target=Process1,
            args=(pipe_proc1_out, )
        )

        self.proc2 = Process(
            target=Process2,
            args=(pipe_proc2_in, )
        )

    def run(self):
        try:
            self.proc1.start()
            self.proc2.start()

            while True:
                continue
        except KeyboardInterrupt:
            print("Quitting processes...")
            self.proc1.join(1)
            if self.proc1.is_alive():
                self.proc1.terminate()

            self.proc2.join(1)
            if self.proc2.is_alive():
                self.proc2.terminate()

            print("Finished")


def pipes():
    c = Controller()
    c.run()


if __name__ == "__main__":
    pipes()

请注意，当前仅当父级进程通过键盘中断时，才能关闭Tkinter窗口。

每当我单击该按钮并调用该按钮的命令时，我的程序进入错误状态，并显示以下消息：

Exception in Tkinter callback
Traceback (most recent call last):
  File "C:UsersUSERAnaconda3envsTHSlib	kinter\__init__.py", line 1705, in __call__
    return self.func(*args)
  File "C:UsersUSERPycharmProjectsPythonPlaygroundpipes_advanced.py", line 26, in btn1_clicked
    self.pipe_out.send(["hello", 1, 2])
  File "C:UsersUSERAnaconda3envsTHSlibmultiprocessingconnection.py", line 206, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File "C:UsersUSERAnaconda3envsTHSlibmultiprocessingconnection.py", line 280, in _send_bytes
    ov, err = _winapi.WriteFile(self._handle, buf, overlapped=True)
BrokenPipeError: [WinError 232] The pipe is being closed

起初我认为问题出在我从Entry.get()调用中收到的值(我的Tkinter技能已经生疏)。我打印了msg并从小工具中获得了文本。

接下来我尝试将一个常量字符串作为通过管道发送的参数的值：

def btn1_clicked(self):
    self.pipe_out.send(["hello", "world"])

出现相同的错误。捕获异常BrokenPipeError实际上对我没有任何好处(我想除非我想要处理管道破裂的情况)。

如果我对程序的第一个版本(没有Tkinter)执行相同的操作，它就可以工作。这让我相信，我的问题来自于我整合Tkinter的方式。

---

解决方案

您的问题是轮询管道，但documentation显示：

轮询([超时])

返回是否有可供读取的数据。
如果未指定超时，则它将立即返回。

在第一个示例中，它之所以有效，是因为当启动Process1时，您会立即将数据发送到管道：

    def run(self):
        try:
            while True:
                print("Sending message to process 2")
                self.pipe_out.send(["hello"])
        except KeyboardInterrupt:
            pass

并持续执行此操作，以便.poll将返回True，并且Process2中的循环将继续。

与tkinter一样，不会立即将任何内容发送到管道，它会等待用户单击按钮，当可能发生任何一种情况时，Process2已经调用了poll，它立即返回False，但它甚至没有启动该循环。如果您注意到了，那么它也几乎立即在终端中打印

"；进程2已完成接收

要解决这个问题，似乎最简单的方法是

while self.pipe_in.poll(None):

根据文档，哪个意思是

如果超时为None，则使用无限超时。

对于像用户界面这样的东西，这似乎是最合适的(至少从用户的角度来看(或者我认为是这样))，所以基本上run中的run方法应该如下所示：

    def run(self):
        try:
            while self.pipe_in.poll(None):
                request = self.pipe_in.recv()
                method = request[0]
                args = request[1:]

                try:
                    getattr(self, method + "_callback")(*args)
                except AttributeError as ae:
                    print("Unknown callback received from pipe", str(ae))

            print("Process 2 done with receiving")
        except (KeyboardInterrupt, EOFError):
            pass

也与问题无关，但似乎没有必要继承Process1中的tk.Frame(或Process2中的object(除非您真的需要使其与Python2兼容))，您几乎可以继承自tk.Tk，这应该更容易将其实际用作主窗口，因为self将是Tk实例