查询root.main() / Tkinter / TCL事件队列深度

Question

我有一个使用 tkinter UI 编写的应用程序，该应用程序通过 GPIO 堆栈执行现实世界的活动。

举个例子，假设我的应用程序有一个温度探头来检查室温，然后有一个连接空调的外部硬件接口来调整 HVAC 系统的温度输出。这是通过 GPIO 堆栈以简单的方式完成的。

应用程序可以执行一些“安全关键”类型的操作。有很多硬件互锁，但我仍然不希望应用程序崩溃并导致空调全力运行并将我理论上的房子变成冰柜。

问题始终是......如何将 GUI 事件驱动的应用程序转换为 GUI 应用程序，该应用程序还运行另一个循环而无需：

1. 通过将所有内容推入“while(true): do things”并添加 root.update() 来打破 tkinter / tcl 设计。
2. 向应用程序添加线程既困难又可能不安全

所以，我选择了以下架构：

将 tkinter 导入为 tk

def main(args):
    
    # Build the UI for the user
    root = tk.Tk()
    buttonShutDown = tk.Button(root, text="PUSH ME!")#, command=sys.exit(0))
    buttonShutDown.pack()
    
    # Go into the method that looks at inputs and outputs
    monitorInputsAndActionOutputs(root)
    
    # The main tkinter loop. NOTE, there are about 9000 articles on why you should NOT just shove everyhtnig in a loop and call root.update(). This is the corect way of doing it! Use root.after() calls!
    root.mainloop()

def monitorInputsAndActionOutputs(root): 

    print ("checked the room temperature")
    print ("adjusted the airconditioning to make comfy")
    root.after(100, monitorInputsAndActionOutputs, root) # readd one event of the monitorInputsAndActionOutputs process to the tkinter / tcl queue

    return None

if __name__ == '__main__':
    import sys
    sys.exit(main(sys.argv))

这实际上工作得很好，理论上它会永远有效，每个人都会有舒适（TM）。

但是，我看到一个潜在的问题，即应用程序有相当多的方法执行不同的闭环 PID 样式调整（读取、思考、更改、迭代），并且我担心 tkinter 事件队列将开始堵塞处理after() 命令和整个系统要么慢得像爬行一样，要么完全崩溃，要么冻结 UI，要么三者兼而有之。

所以我想做的是添加类似的内容（粗略的代码示例，这当然不会运行）：

def (checkQueueDepth):
   
   myQueue = root.getQueueLength()

   if(myQueue > 100 items):
       doSomethingToAlertSomeone()
       deprioritiseNonCriticalEventsTillThingsCalmTFDown()

为了实现这样的事情，我需要以编程方式获取 root / tkinter 队列深度。

在大面积刮胡子之后涉过

1. tkinter 文档
2. TCL / TK 文档
3. tkinter的源代码
4. stackoverflow 的其余部分
5. 互联网
6. 冰箱（用于零食）

我已经确定外部模块无法查询 tk 的队列长度，它基于 tcl 库，并且可能实现或扩展了它们的 notifier。

根据 tkinter 代码，看起来也不存在任何私有方法或对象/变量，我可以从前门冲过去进行询问。

第三，唯一的“解决方案”可能是为 TCL 实现一个新的通知系统，然后扩展其功能以提供此功能。我这样做的可能性为零，我没有知识或时间，我宁愿用另一种语言重写我的整个应用程序。

既然它不能以“正确”的方式完成，我就想到了“我们不会称其为错误，但也可能是”的方式。

我想出的是运行这样的东西：

def checkCodeLoops(root):
    
    # The maximum time we will accept the queue processing to blow out to is 3 seconds.
    maxDelayInQueueResponse = datetime.timedelta(seconds = 3)
    
    # Take a note of the time now as well as the last time this method was run
    lastCheckTime = checkTime
    checkTime = datetime.datetime.now()
    
    # If the time between this invocation of the checkCodeLoops method and the last one is greater than the max allowable time, panic!
    if(checkTime - lastCheckTime > maxDelayInQueueResponse):
        doSomethingToReduceLoad()
        lockUISoUserCantAddMoreWorkToTheQueue()
    
    # reque this method so that we can check again in 100ms (theoretically in 100ms at least - if the queue's flooded then this might never fire)
    root.after_idle(100, monitorInputsAndActionOutputs, root)
    
    return None

总的来说，这是一种很糟糕的做法。您基本上将工作添加到队列中，以查看队列是否繁忙（愚蠢），并且还有另一个问题 - 如果队列被淹没，那么此代码实际上不会运行，因为添加到 after_idle() 进程的任何事件都是仅当队列中没有 after() 和其他更高优先级的项目时才进行处理。

所以我可以将其更改为 root.after 而不是 root.after_idle，理论上这会有所帮助，但我们仍然存在使用排队事件来检查队列是否不起作用的问题，这有点愚蠢。

如果我有办法检查队列深度，我就可以在事情进入恐慌阶段之前开始实施负载控制策略。

真的很想听听是否有人有更好的方法。

Answer 1

不幸的是，实际上没有任何方法可以知道。当您提出问题时，许多事件源只知道是否有“至少一个”待处理事件（对应于单个操作系统事件），并且系统调用都能够接受来自相应源的多个事件那个点（并且由于非阻塞 I/O 可以安全地做到这一点）。 您

可以

做的是监视常规计时器事件之间是否发生空闲事件。当其他事件源没有任何贡献时，空闲事件就会触发，另一种选择是在 select() 系统调用中进入睡眠状态。 （或者

poll()

或通知程序已配置使用的任何内容，具体取决于平台和构建选项。）如果在这些常规计时器事件之间正在处理空闲事件，则队列是健康的；如果没有，你就处于超支的境地。