在线程中中断执行脚本的Python无限循环

Question

一个图形用户界面（GUI）应用程序允许用户编写一个Python脚本，然后运行它（通过exec）。问题是，如果用户不小心（我不担心恶意行为，只是用户在编码时的诚实错误）写的脚本里有一个无限循环，控制权就永远不会返回到我的应用程序。因为GUI的原因，键盘中断也不起作用。

我找到了四种处理这个问题的方法：

1. 追踪：使用sys.settrace，这样在每一行代码执行时都会调用一个函数；在这个函数里，我可以加入逻辑来尝试识别是否同一段代码被反复执行（实际操作起来有多难我还不太清楚）。我猜这可能会大大降低执行速度。
2. 异步异常：在一个单独的线程中运行脚本，并使用ctypes让主线程通过PyThreadState_SetAsyncExc在脚本线程中引发异常，从而将其踢出循环（可以参考这个链接和这个链接）。这样，调用exec的线程就能重新获得控制权，并采取适当的行动（比如给用户显示消息等）。我知道，线程不应该被外部强制终止，但在这种情况下，即使这样做会让脚本正在修改的对象处于不确定状态，至少可以让GUI丢弃那些不再可靠的对象（因为所有可供脚本修改的内容都在一个大对象里，可以被丢弃并从磁盘重新加载）。
3. 单独进程：使用multiprocessing模块，在一个单独的进程中运行脚本；不过脚本可能会调用在父进程主线程中的对象，这样会变得相当复杂。
4. 守护线程：在守护线程中运行脚本，如果脚本在一定时间内没有返回，就认为它“卡住”了。当应用程序退出时，线程会继续运行，用户可以通过任务管理器强制结束它（或者退出的线程可以通过引发SystemException之类的方式来结束自己）。

所以问题是：如果我必须提供这个功能，以上四种方法中哪一种是最不糟糕的选择？还有其他方法吗？

Answer 1

如果你事先不知道代码会执行什么，使用 #1 settrace 来判断代码是否仍然活跃会很困难，而且确实会让代码执行变慢，可能会慢很多，这取决于代码的用途。可以看看 停机问题。

选项 #3 是最好的（可以和 #4 半结合）。这就是为什么要使用独立进程的原因——让它们单独工作，同时让第一个进程继续做其他事情。设置起来并不难。

新的进程（P2）不需要包含与图形界面（GUI）相关的对象，关注点应该分开。这并不意味着它们不能互动。比如，可以使用一对套接字进行通信，双方都使用 pickle 来互相发送 Python 对象。图形界面的主循环应该每隔例如 50 毫秒检查一次套接字，设置为非阻塞模式，以便接收来自 P2 的通信。如果 P2 想的话，可以向图形界面发送消息，图形界面会做出回应。

P2 在一个新进程 P3 中执行代码（虽然不是绝对必要，但这样设置更好）。P2 还会检查来自图形界面的命令（如果需要，可以定期检查），比如当图形界面中的停止按钮被点击时，命令是“停止脚本执行”，此时 P2 可以执行 os.kill(P3ID..) 或 P3.terminate()（如果使用 multiprocessing 等）。或者它也可以向图形界面发送命令，并接收所需的数据，图形界面会在最多 50 毫秒内做出回应。

下面的代码只是一些片段，完全没有经过测试，只是给你一个架构的概念。最好将其分成不同的部分，比如一个 SocketCom 类，用于封装每个套接字，处理数据转换为字节的过程，使用 pickle，并通过底层套接字发送，或者在阻塞或非阻塞模式下接收单个消息，使用底层套接字接收并解包消息，然后再传回去，等等。

通用代码

sGUI, s2 = socket.socketpair(socket.AF_UNIX, socket.SEQPACKET)
sGUI.setblocking(False) # for direct use, but will raise system-dependent errors, better to use select
P2 = multiprocessing.Process(target=P2process, args=(s2,codeToRun) )

def P2process (sock, codeToRun) :
    # sock is the socket connected to GUI socket
    P3 = multiprocessing.Process(target=P3process, args=(codeToRun,) ) # note, args is tuple
    # block (up to you) for messages from/to GUI, check P3 periodically, etc.
    # e.g. can do :
    sock.sendall(pickle.dumps("sendMeX"))
    X = pickle.loads(sock.recv(length))
    # or e.g. a blocking loop that responds only to messages from GUI :
    while True :
        msg = pickle.loads(sock.recv(length))
        if msg == 'status' :
            sock.sendall(pickle.dumps(P3.is_alive()))
        elif msg == 'stop' :
            P3.terminate()
        elif msg == 'newExecCode' :
            newExecCode = pickle.loads(sock.recv(length))
        elif msg == 'quit' :
            P3.terminate()
            break
    ...

def P3process (codeToRun) :
    exec(codeToRun) # you should sandbox its context with custom globals & locals
    # up to you how to solve the halting problem ;)
    # user probably decides and GUI has 'stop' button if exec takes too long

对于图形界面：

图形界面可以随时使用 sGUI 向 P2 发送命令，比如在按钮点击时。监听来自 P2 的消息大致可以这样写：

def GUI_P2Com_loop (self) :
    # this is called once, and at the end of the function, it registers itself as a
    # callback to be called again after a timeout. It checks for messages from P2, 
    # and responds as necessary. It can also launch other processes to respond instead.
    try :
        reads, w, x = select.select([sGUI], [], [], 0) # 0 = non-blocking
        if sGUI in reads :
            msg = pickle.loads(sGUI.recv(length))
            # got message from P2, do whatever
            if msg == 'sendMeX' :
                sGUI.sendall(pickle.dumps(X))
    ...

    # so that the GUI can get on with responding to user interaction
    # all GUI frameworks should have a function for registering a callback after timeout
    # e.g. a tkinter widget would call :
    self._job = widget.after(50, GUI_P2Com_loop) # where widget could be self if class extends widget

这个 _job 被存储，以便可以用例如 tkinter 再次取消通信循环：

def cancel_com_loop (self) :
    self.after_cancel(self._job)

可以查看 tkinter effbot 文档。

Answer 2

虽然大家普遍不太看好选项1，但选项2或4和1结合起来会很好。

你可以创建一个线程，这个线程会安装你的 settrace 回调函数，然后继续加载和执行脚本代码模块，放在某种尝试/捕获的结构里。

实现 settrace 的时候，不需要进行复杂的分析，只要检查一下它自己线程的“年龄”，如果太老了，就抛出一个异常。

然后，这段包装代码可以通知图形界面（GUI）超时了。

我想这里的一个重点是，你可能想要执行用户的脚本时，不是直接用 exec，而是通过加载模块的方式来执行，这样你可以在周围加上一些Python代码，比如尝试/捕获的结构，以及安装像settrace或 Timer 线程这样的东西，这些可以注入提示等等。

Answer 3

可以让Python代码保持在主线程中，然后创建一个工作线程，这个工作线程会先休眠一段时间，然后用SIGINT来打断主线程。

当用户的进程结束时，你可以终止这个工作线程。但是如果工作线程被触发了，你就会得到一个类似于键盘中断的信号，这样你可以捕捉到这个异常并进行清理。

这样做的话，你就不会因为和主程序共享数据而出现内存问题，同时在固定的时间后，你也能以一种方式来终止那些失控的子程序。

Answer 4

一种实现方法是使用自定义的 multiprocessing.Manager 对象。这个对象可以帮你处理同步的问题。 （链接： multiprocessing）

下面是一个示例，展示了如何让多个进程调用同一个实例的方法。注意，我没有使用实例的状态，这部分留给读者自己去理解 :)

原来的代码是把“maths.add”（实例方法）传给了进程池，但方法是不能被序列化的。因此我创建了一个全局的“my_add”，它接受一个数学实例（这个是可以序列化的），然后再接受一些数字，把这些数字加起来并返回结果。

源代码

from multiprocessing import Pool
from multiprocessing.managers import BaseManager

class MathsClass(object):
    def add(self, x, y):
        return x + y
    def mul(self, x, y):
        return x * y

class MyManager(BaseManager):
    pass

MyManager.register('Maths', MathsClass)

def my_add(mobj, *args):
    return mobj.add(*args)

if __name__ == '__main__':
    manager = MyManager()
    manager.start()
    maths = manager.Maths()

    # pass the shared 'maths' object into each process

    pool = Pool()
    print pool.apply(my_add, [maths, 4, 3])
    # print maths.add(4, 3)         # prints 7
    print pool.apply(my_add, [maths, 7, 8])
    # print maths.mul(7, 8)         # prints 56

输出结果

7
15