Python实现带停止信号的定时计算（适配UCI象棋引擎协议）

核心思路 #

用多进程分离计算逻辑与命令监听，结合事件信号（multiprocessing.Event）和队列（multiprocessing.Queue）实现：

• 主线程：非阻塞监听命令行输入，处理stop命令，同时管理超时逻辑，通过事件通知计算进程终止。
• 计算进程：独立运行搜索逻辑，定期向主线程推送中间结果，实时监听停止事件，一旦触发就返回当前最优结果。

这种方式不需要强行拆分计算步骤，而是利用Python进程间通信机制实现优雅中断，同时保留超时和手动停止的灵活性。

完整代码实现 #

import multiprocessing
import sys
import time
from select import select

def chess_search(stop_event, result_queue, search_time):
    """模拟象棋引擎的搜索逻辑，实际可替换为真实引擎代码"""
    best_move = None
    evaluation = -float('inf')
    depth = 0
    start_time = time.time()

    # 模拟深度递增的搜索循环（实际引擎此处为真实搜索逻辑）
    while not stop_event.is_set():
        depth += 1
        # 模拟单深度搜索耗时（实际替换为引擎搜索代码）
        time.sleep(0.5)
        
        # 生成当前深度的临时结果
        current_eval = evaluation + depth
        current_move = f"Move_{depth}"
        
        # 更新全局最优结果
        if current_eval > evaluation:
            evaluation = current_eval
            best_move = current_move
        
        # 向主线程推送中间结果
        result_queue.put({
            "depth": depth,
            "best_move": best_move,
            "evaluation": evaluation,
            "time_elapsed": time.time() - start_time
        })
        
        # 检查超时（search_time=-1表示无限搜索）
        if search_time != -1 and (time.time() - start_time) >= search_time:
            stop_event.set()
            break

    # 搜索结束，推送最终结果标记
    result_queue.put({"finished": True, "final_best_move": best_move, "final_eval": evaluation})

def main():
    stop_event = multiprocessing.Event()
    result_queue = multiprocessing.Queue()
    search_process = None
    search_time = -1  # 默认无限搜索

    print("UCI Engine Ready. Enter commands:")
    while True:
        # 非阻塞监听命令行输入，避免阻塞结果处理
        readable, _, _ = select([sys.stdin], [], [], 0.1)
        if readable:
            command = sys.stdin.readline().strip().lower()
            
            if command.startswith("go"):
                # 解析UCI的go命令参数，支持movetime指定时长
                parts = command.split()
                if len(parts) >= 3 and parts[1] == "movetime":
                    search_time = int(parts[2]) / 1000  # 毫秒转秒
                else:
                    search_time = -1
                
                # 启动新的搜索进程
                stop_event.clear()
                search_process = multiprocessing.Process(
                    target=chess_search,
                    args=(stop_event, result_queue, search_time)
                )
                search_process.start()
                print(f"Started search (time: {'infinite' if search_time == -1 else f'{search_time}s'})")
            
            elif command == "stop":
                # 发送停止信号给计算进程
                if search_process and search_process.is_alive():
                    stop_event.set()
                    print("Sent stop signal to search process")
            
            elif command == "quit":
                # 退出前清理进程
                if search_process and search_process.is_alive():
                    stop_event.set()
                    search_process.join()
                print("Exiting engine")
                break
        
        # 实时处理计算进程推送的结果
        while not result_queue.empty():
            result = result_queue.get()
            if "finished" in result:
                print(f"Search completed. Best move: {result['final_best_move']}, Eval: {result['final_eval']}")
                search_process.join()
                search_process = None
            else:
                # 按UCI协议格式输出中间信息
                print(f"info depth {result['depth']} score cp {result['evaluation']} time {int(result['time_elapsed']*1000)} pv {result['best_move']}")

if __name__ == "__main__":
    main()

关键细节说明 #

• 非阻塞命令监听：用select模块实现主线程对输入的非阻塞监听，确保同时能处理计算进程的结果推送。
• 停止信号机制：multiprocessing.Event是进程安全的信号量，计算进程可在搜索循环的自然间隙（比如每完成一层深度后）检查信号，无需强行拆分短步骤。
• 中间结果上报：通过队列让计算进程定期推送当前最优结果，主线程实时接收并输出，符合UCI协议的info指令要求。
• 超时处理：计算进程内部维护定时器，超时后自动触发停止信号，无需主线程额外干预。

适配真实引擎的修改点 #

1. 将chess_search函数内的模拟逻辑替换为实际的象棋搜索代码（比如 minimax、alpha-beta剪枝实现）。
2. 完善命令解析，支持UCI协议的depth、nodes等更多搜索参数。
3. 严格按照UCI规范优化info信息的输出格式。

内容的提问来源于stack exchange，提问作者mexx