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

人气:947 发布:2022-10-16 标签: python pipe multiprocessing tkinter python-multiprocessing

问题描述

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

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实例,它一直运行到收到键盘中断。它还处理Process1Process2两个进程,前者不断发送,后者不断接收。

上面的代码是一个更大的项目的框架,涉及复杂的图形用户界面(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实例

692