在使用PyAutoGUI进行多线程时,如何避免X11的竞态条件?

编程语言 2026-07-12

我正在使用X11显示服务器。我有一个Python脚本,使用PyAutoGUI将屏幕上的触摸转换为鼠标移动和控制。用于发送触摸的坐标(及其他数据)数据包是通过一个命名管道(fifo)发送的。正因为如此,我需要在一个线程中运行一个PyAutoGUI函数,用来检查最后一个数据包是否超时,从而让鼠标抬起(这是为了在轻触和拖拽之间做出判断,并结束拖拽)。

为了让程序运行得更顺畅,我需要设置:

pyautogui.PAUSE = 0

然而,这似乎会在pyautogui与 X11服务器之间引发竞争条件,因此当它收到一个它未预期的回复时,程序会崩溃:

RuntimeError: Expected reply for request 833, but got 834.  Can't happen!

以下是相关代码:

import time
import threading
import Xlib.threaded # this is to make the Xlib dependency of pyautogui threadsafe
import pyautogui as pag
pag.FAILSAFE= False
pag.PAUSE = 0

PIPE_PATH = "my path to named pipe"

screenWidth, screenHeight = pag.size()

global first_touch
first_touch = True
global previous_time
previous_time = None

def touch_timeout() -> None:
    global first_touch
    global previous_time

    while True:
        if previous_time != None:
            while True:
                local_time = time.time()
                if local_time - previous_time > 0.01:
                    pag.mouseUp() # error appears here
                    first_Touch = True

timeout_thread = threading.Thread(target=touch_timeout)
timeout_thread.start()

with open(PIPE_PATH, "r") as f:
    while True:
        for message in f:
            previous_time = time.time()

            # packet handling logic here ...
            # ...

            pag.moveTo(xcoord, ycoord)
            if first_touch:
                pag.mouseDown()
                first_Touch = False

以下是我尝试过的:

  • 我已禁用该线程(这会移除鼠标向上移动的功能,但对于此测试来说这是可以的)。这导致没有错误,性能也很流畅。
  • 我把pag.PAUSE调整到了不同的数值。除了0 之外的任何值都太慢,无法提供所需的性能。然而,数值越高,出现上述错误的频率越低。

解决方案

你很可能需要在对pag/X11的调用周围进行互斥锁保护(mutex locking)。

站内所有文章版权归属LeftHeroAI导航站,无授权禁止任何主体转载、抄袭、复制内容,亦不得私自架设镜像站点。一经侵权,本站将通过法律途径追责。

相关文章