Python Tkinter 应用程序在 Mac OS X 上导致 fork()/exec() 错误

Question

我正在 Mac OS X(10.7 和 10.8)上运行 python (2.7) Tkinter GUI 应用程序。 UI 位于一个单独的进程中，该进程使用多处理从主脚本中 fork 出来。然而，当我运行时，它失败了:

'The process has forked and you cannot use this CoreFoundation functionality safely. You MUST exec(). Break on THE_PROCESS_HAS_FORKED_AND_YOU_CANNOT_USE_THIS_COREFOUNDATION_FUNCTIONALITY_YOU_MUST_EXEC__() to debug.'

这在 Windows 上运行得很好。

我发现了一个与它相关的错误，用 python 记录:http://bugs.python.org/issue8713但遗憾的是，看起来该修复仅在 3.x 版本中实现，但我也需要支持 2.7。

我知道还有一些关于同一错误的其他问题，如下所示:Python multiprocessing bug on Mac OS X

但我无法弄清楚在我的特定情况下什么可以真正解决问题。

关于如何让它在 Mac 上运行有什么想法吗？

代码如下。 DriverVisualizer() 创建(在另一个脚本中)，它初始化另一个进程上的 UI。

from util import *
from multiprocessing import Process, Pipe
from Tkinter import *
import threading
import Queue
from time import *

class VisualizerUI:
    def __init__(self, conn, count, pixelSize):
        self._conn = conn
        self._master = Tk()
        self._q = Queue.Queue()

        self._count = count
        self._values = []
        self._leds = []

        self._pixelSize = pixelSize
        self._pixelPad = int(pixelSize / 2)
        self._pixelSpace = 4

        #init colors to all black (off)
        for i in range(self._count):
            self._values.append("#000000")

        self.initUI()

        self._thread = threading.Thread(target=self.commThread).start()

    def mainloop(self):
        self._master.mainloop()
        try:
            self._conn.send({"status" : False})
        except:
            pass

    def updateUI(self):
        try:
            for i in range(self._count):
                    self._canvas.itemconfig(self._leds[i], fill=self._values[i])
        except TclError:
            #Looks like the UI closed!
            pass

    def commThread(self):
        data = None
        error = False

        while True:
            #bit of a hack, but need to check occasionaly for window size change
            if self._width != self._master.winfo_width() or self._height != self._master.winfo_height():
                self._width = self._master.winfo_width()
                self._height = self._master.winfo_height()
                self._master.after_idle(self.layoutPixels)

            try:
                data = self._conn.recv()
            except EOFError:
                error = True
                break

            if data["run"]:
                self._values = data["data"]
                self.updateUI()
                self._conn.send({"status" : True})
            else:
                break
        if not error:
            self._conn.send("Killing UI...")
            self._master.destroy()

    def layoutPixels(self):
        self._canvas.config(width=self._width, height=self._height)
        newRow = True
        x_off = self._pixelPad
        y_off = self._pixelPad
        for i in range(self._count):
            if (x_off + (self._pixelSize * 2) + self._pixelSpace + self._pixelPad) > self._width:
                newRow = True
                y_off = y_off + self._pixelPad + self._pixelSize
            if newRow:
                x_off = self._pixelPad
                newRow = False
            else:
                x_off = x_off + self._pixelSize + self._pixelSpace

            self._canvas.coords(self._leds[i], x_off, y_off, x_off + self._pixelSize, y_off + self._pixelSize)

        y = (y_off + self._pixelSize + self._pixelPad)
        if self._height != y:
            self._master.geometry("{0}x{1}".format(self._width, y))
            self._master.update()

    def __CancelCommand(event=None): 
        pass

    def initUI(self):
        m = self._master
        m.protocol('WM_DELETE_WINDOW', self.__CancelCommand)

        m.title("LED Strip Visualizer")
        m.geometry("1400x50")
        m.update()
        self._width = m.winfo_width()
        self._height = m.winfo_height()
        m.minsize(self._width, self._height)

        self._canvas = Canvas(self._master, background="#000000")
        c = self._canvas
        c.pack(side=TOP)

        for i in range(self._count):
            index = c.create_oval(0,0,self._pixelSize,self._pixelSize, fill=self._values[i])
            self._leds.append(index)

        #m.bind("<Configure>", self.resize)

        self.layoutPixels()

def toHexColor(r,g,b):
    return "#{0:02x}{1:02x}{2:02x}".format(r,g,b)

def startUI(conn, count, pixelSize):
        ui = VisualizerUI(conn, count, pixelSize)
        ui.mainloop()

class DriverVisualizer(object):
    """Main driver for Visualizer UI (for testing)"""

    def __init__(self, leds, pixelSize = 15, showCurrent = False):
        self.leds = leds
        self._showCurrent = showCurrent
        if self._showCurrent:
            self._peakCurrent = 0;
        else:
            self._peakCurrent = None

        self._parent_conn, self._child_conn = Pipe()
        p = Process(target=startUI, args=(self._child_conn, self.leds, pixelSize))

        p.start()
        sleep(0.5) # give the UI some time to spin up before throwing data at it

    def __del__(self):
        self._parent_conn.send({"data" : None, "run" : False})
        print self._parent_conn.recv()

    def calcCurrent(self, data):
        c = 0
        for r, g, b in data:
            c = c + int(((r/255.0) * 20.0) + ((g/255.0) * 20.0) + ((b/255.0) * 20.0))
            if c > self._peakCurrent: 
                self._peakCurrent = c

        return c

    #Push new data to strand
    def update(self, data):
        c = None
        if self._showCurrent:
            c = self.calcCurrent(data)
        self._parent_conn.send({"data" : [toHexColor(*(data[x])) for x in range(self.leds)], "run" : True, "c" : c, "peak" : self._peakCurrent})
        resp = self._parent_conn.recv()
        if not resp["status"]:
            error = True
            parent_conn.close()

Answer 1

我遇到了同样的问题，请检查这个:
https://stackoverflow.com/a/19082049/1956309

这可以让您在此处找到 Tkinter 错误:
http://bugs.python.org/issue5527#msg195480

对我有用的解决方案(Mac OS 10.8 和 Python 2.7)是重新排列代码，以便在调用任何类型的 process.start() 之后放置“导入 Tkinter”

看起来像:

import multiprocessing

def cam_loop(the_q):
    while True:
        the_q.put('foo in the queue')

def show_loop(the_q):
    while True:
        from_queue = the_q.get()
        print from_queue

if __name__ == '__main__':
    try:
        the_q = multiprocessing.Queue(1)

        cam_process = multiprocessing.Process(target=cam_loop,args=(the_q, ))
        cam_process.start()

        show_process = multiprocessing.Process(target=show_loop,args=(the_q, ))
        show_process.start()

        import Tkinter as tk  # << Here!
        cam_process.join()
        show_loop.join()

    except KeyboardInterrupt:
        cam_process.terminate()
        show_process.terminate()

P.d:谢谢JW Lim向我展示了良好的举止!