#!/usr/bin/env python3 """ecat_proc.py — ecat_native 进程治理(跨进程锁 / autostart / kill / RealTime)。 从 esp32_server.py 外提(R8)。职责是「谁有权把 ecat_native 拉起来/杀掉」, 不碰 HTTP/静态/lease——那些留在 esp32_server.py。 路径注入:BASE_DIR **只在 esp32_server 定义一次**,启动时调用 `init(BASE_DIR)` 设置本模块的 ECAT_DIR / ECAT_NATIVE_EXE,并向 ecat_client 注册 autostart 钩子。 保留 P0 行为(不得回退): - 跨进程命名互斥锁 _EcatProcXLock(GUI 进程 + 服务子进程串行 kill/start) - **先复用健康监听实例**,绝不 kill 对方刚拉起的 native - 默认网卡 I210(与 js/common/device-link.js / pages/motor.html 一致,R2) - 拉起时带 --parent-pid,native 在本服务退出后自我了断 - stdout 落盘(DEVNULL 会吞掉崩溃现场) """ from __future__ import annotations import ctypes import os import subprocess import threading import time import ecat_client # 网卡默认子串:与 js/common/device-link.js / pages/motor.html 保持一致(R2) ECAT_DEFAULT_ADAPTER = "I210" ECAT_DIR = None ECAT_NATIVE_EXE = None g_ecat_child = None # 由本服务拉起的 ecat 服务子进程(复用已有实例时为 None) g_ecat_log_fp = None # native stdout 落盘句柄(见 autostart 内注释) g_ecat_lock = threading.Lock() # 串行化自动启动, 避免并发请求拉起多个进程 # R3: 跨进程互斥(GUI 进程 + 服务子进程都可能 autostart/kill native) _ECAT_XPROC_MUTEX_NAME = "Local\\DCU_Esp32Tools_EcatNativeProc" def _resolve_ecat_dir(base_dir): """定位 tools/ecat(ecat_native.exe 的仓库,与本仓库平级)。 从 git worktree(.../tools/ESP32/.worktrees/)启动时,BASE_DIR 的 上一级是 .worktrees 而不是 tools,直接 join("..","ecat") 会拼成 .../ESP32/.worktrees/ecat,找不到 exe。先走回主检出再取兄弟目录。 """ d = os.path.abspath(base_dir) parent = os.path.dirname(d) if os.path.basename(parent) == ".worktrees": d = os.path.dirname(parent) # → .../tools/ESP32 主检出 return os.path.normpath(os.path.join(d, "..", "ecat")) def init(base_dir: str) -> dict: """由 esp32_server 在导入时调用:注入 BASE_DIR,注册 autostart 钩子。""" global ECAT_DIR, ECAT_NATIVE_EXE ECAT_DIR = _resolve_ecat_dir(base_dir) ECAT_NATIVE_EXE = os.path.join(ECAT_DIR, "build_native", "ecat_native.exe") ecat_client.set_autostart_hook(autostart) return {"ecat_dir": ECAT_DIR, "native_exe": ECAT_NATIVE_EXE} class _EcatProcXLock: """Windows 命名互斥锁:串行化所有会 kill/拉起 ecat_native 的路径。""" def __init__(self, timeout_ms=60000): self.timeout_ms = timeout_ms self._handle = None self._owned = False def __enter__(self): k32 = ctypes.WinDLL("kernel32", use_last_error=True) self._handle = k32.CreateMutexW(None, False, _ECAT_XPROC_MUTEX_NAME) if not self._handle: return self r = k32.WaitForSingleObject(self._handle, self.timeout_ms) # WAIT_OBJECT_0=0, WAIT_ABANDONED=0x80 self._owned = r in (0, 0x80) return self def __exit__(self, *exc): if not self._handle: return False k32 = ctypes.WinDLL("kernel32", use_last_error=True) try: if self._owned: k32.ReleaseMutex(self._handle) finally: k32.CloseHandle(self._handle) self._handle = None return False def _set_ecat_realtime(pid: int) -> bool: """把 ecat_native 提到 RealTime 进程优先级。 native 自己 SetPriorityClass(RT) 会被系统静默钳成 High(令牌无 SeIncreaseBasePriorityPrivilege)。由本服务(启动方)从外部设置才能粘住。 pdo 线程已是 TIME_CRITICAL:进程类 High→线程 15,RealTime→线程 31, 后者才扛得住桌面 DPC/调度抢占。实测 1ms 档 late ~12/s → 个位数。 失败返回 False,不影响启动。 """ try: import ctypes.wintypes as wt # noqa: F401 PROCESS_SET_INFORMATION = 0x0200 REALTIME_PRIORITY_CLASS = 0x00000100 HIGH_PRIORITY_CLASS = 0x00000080 k32 = ctypes.WinDLL("kernel32", use_last_error=True) h = k32.OpenProcess(PROCESS_SET_INFORMATION, False, int(pid)) if not h: return False try: if k32.SetPriorityClass(h, REALTIME_PRIORITY_CLASS): return True return bool(k32.SetPriorityClass(h, HIGH_PRIORITY_CLASS)) finally: k32.CloseHandle(h) except Exception: return False def _ecat_reuse_listening_native(): """端口上已有健康 native → 直接复用(可能是另一进程刚拉起的),不 kill。""" g_port = ecat_client.g_ecat_port ports = ([g_port] if g_port else []) + \ [p for p in ecat_client.ECAT_PORTS if p != g_port] for p in ports: if not ecat_client._port_taken(p): continue try: status, data = ecat_client._ecat_post( p, "/api/native/health", b"{}", 1.0, connect_timeout=0.4) if status == 200: return p except Exception: continue return None def _kill_all_ecat_native(): """强杀所有 ecat_native.exe,返回杀掉的个数。 两个 native 并存会抢 Npcap 适配器:后起的抓不到从站、HTTP 超时、 PDO 空转。启动新实例前必须保证旧的死透。只按进程名杀,不碰 python。 """ try: out = subprocess.check_output( ["tasklist", "/FI", "IMAGENAME eq ecat_native.exe", "/FO", "CSV", "/NH"], text=True, stderr=subprocess.DEVNULL, timeout=8, creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), ) or "" except Exception: return 0 pids = [] for line in (out or "").splitlines(): parts = [p.strip().strip('"') for p in line.split(",")] if len(parts) >= 2 and parts[0].lower() == "ecat_native.exe": try: pids.append(int(parts[1])) except ValueError: pass if not pids: return 0 for pid in pids: try: subprocess.run( ["taskkill", "/F", "/PID", str(pid)], capture_output=True, timeout=5, creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), ) except Exception: pass # 等 Npcap 句柄随进程退出由内核回收 deadline = time.time() + 3.0 while time.time() < deadline: try: again = subprocess.check_output( ["tasklist", "/FI", "IMAGENAME eq ecat_native.exe", "/FO", "CSV", "/NH"], text=True, stderr=subprocess.DEVNULL, timeout=4, creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), ) or "" if "ecat_native.exe" not in (again or "").lower(): break except Exception: break time.sleep(0.2) else: time.sleep(0.3) return len(pids) def autostart(): """按需拉起 tools/ecat 的服务进程。返回 (port, log_lines)。 单实例策略(2026-09-12,R3 加固 2026-09-23): - 跨进程命名互斥锁串行化 kill/start(GUI 进程与服务子进程共用)。 - **先探测健康监听实例并复用**,绝不 kill 对方刚拉起的 native。 - 本进程子进程仍存活 → 只等它听口,**绝不 kill_all**(否则启动中的实例 会被自己下一次请求杀掉,形成 8s kill_stale 死循环)。 - 子进程已死 / 从未拉起且无健康监听 → 先 taskkill 全部残留 ecat_native,再起唯一实例。 port 为 None 表示拉不起来。 """ global g_ecat_child, g_ecat_log_fp if not ECAT_NATIVE_EXE: return None, ["ecat_proc.init 未调用(BASE_DIR 未注入)"] with _EcatProcXLock(): with g_ecat_lock: logs = [] # ── R3: 另一进程(GUI/服务)已在跑的健康实例 → 复用 ── reuse = _ecat_reuse_listening_native() if reuse: ecat_client.g_ecat_port = reuse return reuse, [f"复用已在监听的 EtherCAT native (端口 {reuse})"] # ── 我们自己拉起的还活着:只复用/等它,不杀 ── if g_ecat_child is not None and g_ecat_child.poll() is None: port = ecat_client._ecat_find_port() if port: ecat_client.g_ecat_port = port return port, [] # 仍在启动(kill_stale/绑口最多约 10s):等完再判 deadline = time.time() + 10 while time.time() < deadline: if g_ecat_child.poll() is not None: break port = ecat_client._ecat_find_port() if port: ecat_client.g_ecat_port = port return port, [] time.sleep(0.3) if g_ecat_child.poll() is None: return None, logs + ["EtherCAT 服务仍在启动中(未监听端口),请稍候再试"] g_ecat_child = None if g_ecat_child is not None and g_ecat_child.poll() is not None: g_ecat_child = None if not os.path.isfile(ECAT_NATIVE_EXE): return None, [ f"找不到 EtherCAT 原生守护进程: {ECAT_NATIVE_EXE}", "请先运行 tools\\ecat\\build_native.bat 编译(需 VS2022 + Npcap)。", ] # 无健康监听才清残留:禁止双实例抢网卡 killed = _kill_all_ecat_native() if killed: logs.append(f"已清掉 {killed} 个残留 ecat_native(单实例,防抢网卡)") try: # 日志落文件而不是 DEVNULL:native 的 stdout 是崩溃/楔死/自动重拉的 # 唯一现场(WEDGE、kill_stale、进 OP 全打这里),DEVNULL 吞掉后只能靠 # Windows 事件日志反推(2026-09-12 教训)。追加写,重启服务不清零。 if g_ecat_log_fp is None: g_ecat_log_fp = open( os.path.join(ECAT_DIR, "build_native", "native_stdout.log"), "ab", buffering=0) g_ecat_child = subprocess.Popen( [ECAT_NATIVE_EXE, "--substr", ECAT_DEFAULT_ADAPTER, "--mask", "0x50", "--web", "src/web", "--parent-pid", str(os.getpid())], cwd=ECAT_DIR, stdout=g_ecat_log_fp, stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL, creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), ) logs.append(f"已自动启动 EtherCAT 原生守护进程(pid {g_ecat_child.pid})") if _set_ecat_realtime(g_ecat_child.pid): logs.append(f"EtherCAT 进程优先级已提到 RealTime(pid {g_ecat_child.pid})") else: logs.append("EtherCAT 进程 RealTime 提权失败(保持默认/High, 可手动设)") except OSError as e: return None, logs + [f"启动 EtherCAT 服务失败: {type(e).__name__}: {e}"] deadline = time.time() + 12 while time.time() < deadline: if g_ecat_child.poll() is not None: return None, logs + [ f"EtherCAT 守护进程启动后立刻退出(exit={g_ecat_child.returncode}) —— " "多为 Npcap 未安装或 8090..8095 全被占; " "可双击 tools/ecat/ecat_web.bat 看具体报错"] port = ecat_client._ecat_find_port() if port: ecat_client.g_ecat_port = port logs.append(f"EtherCAT 服务已就绪(端口 {port})") return port, logs time.sleep(0.3) return None, logs + ["EtherCAT 服务启动超时(12s 内未监听端口)"] # 旧名兼容(server / 脚本) _ecat_autostart = autostart def watch_parent(ppid): """兜底看门狗(Job Object 万一不可用时): 父 GUI 进程消失就自我了断, 避免窗口关了服务还在后台占端口。""" def _loop(): k32 = ctypes.windll.kernel32 h = k32.OpenProcess(0x00100000, False, ppid) # SYNCHRONIZE if not h: return k32.WaitForSingleObject(h, 0xFFFFFFFF) # 父进程退出即返回 os._exit(0) threading.Thread(target=_loop, daemon=True).start() _watch_parent = watch_parent