我正在使用 multiprocessing 模块 (Python 3.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，前者不断发送，后者不断接收。
上面的代码是一项涉及复杂 GUI (PySide)、图像处理 (OpenCV) 和游戏引擎 (Panda3D) 的大型项目的框架。所以我尝试添加 Tkinter 作为 GUI 示例:
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:\Users\USER\Anaconda3\envs\THS\lib\tkinter\__init__.py", line 1705, in __call__
return self.func(*args)
File "C:\Users\USER\PycharmProjects\PythonPlayground\pipes_advanced.py", line 26, in btn1_clicked
self.pipe_out.send(["hello", 1, 2])
File "C:\Users\USER\Anaconda3\envs\THS\lib\multiprocessing\connection.py", line 206, in send
self._send_bytes(_ForkingPickler.dumps(obj))
File "C:\Users\USER\Anaconda3\envs\THS\lib\multiprocessing\connection.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说:
poll([timeout])
Return whether there is any data available to be read.
If timeout is not specified then it will return immediately.
在第一个示例中，它可以工作，因为在启动 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 并且它甚至没有启动该循环。如果您注意到，它几乎会立即在终端中打印出
"Process 2 done with receiving"
要解决这个问题，似乎最简单的方法是使用
while self.pipe_in.poll(None):
根据文档的意思
"If timeout is None then an infinite timeout is used."
对于用户界面之类的东西，这似乎是最合适的(至少从用户的角度来看(或者我认为))所以基本上你的 Process2 中的 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 继承(或 object in Process2(除非你真的需要让它与 Python2 兼容))，你几乎可以从 tk.Tk 继承，这应该更容易实际使用它作为主窗口因为 self 将是 Tk 实例
关于python - 尝试通过运行 Tkinter 的发送进程在进程之间通过管道发送任何内容时发生管道错误，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/69912856/