148 lines
4.8 KiB
Python
148 lines
4.8 KiB
Python
# modules/sbx_debug.py - 沙箱崩溃排查的临时调试日志模块（诊断完成后移除）
|
||
#
|
||
# 双通道日志：
|
||
# 1) UDP 发送到独立监控进程（后端死亡前消息已送达，监控进程独立存活）
|
||
# 2) 文件追加 .wsl-exp/sbx_debug.log（每行立即落盘，保底冗余）
|
||
# 环境变量：
|
||
# SBX_DEBUG=0 关闭全部调试日志
|
||
# SBX_DEBUG_PORT=9956 UDP 监控端口
|
||
# SBX_DEBUG_SWALLOW_CTRL=0 控制台事件不吞掉（记录后恢复默认行为=真实死亡）
|
||
|
||
from __future__ import annotations
|
||
|
||
import datetime
|
||
import os
|
||
import signal
|
||
import socket
|
||
import threading
|
||
import traceback
|
||
|
||
_ENABLED = os.environ.get("SBX_DEBUG", "1").strip().lower() not in {"0", "false", "no", "off"}
|
||
_PORT = int(os.environ.get("SBX_DEBUG_PORT", "8092") or "8092") # UDP，与后端 TCP 端口不冲突
|
||
_SWALLOW = os.environ.get("SBX_DEBUG_SWALLOW_CTRL", "1").strip().lower() not in {"0", "false", "no", "off"}
|
||
|
||
_LOCK = threading.Lock()
|
||
_SOCK = None
|
||
_PROBES_INSTALLED = False
|
||
|
||
|
||
def _log_path() -> str:
|
||
try:
|
||
root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
except Exception:
|
||
root = os.getcwd()
|
||
d = os.path.join(root, ".wsl-exp")
|
||
try:
|
||
os.makedirs(d, exist_ok=True)
|
||
except Exception:
|
||
pass
|
||
return os.path.join(d, "sbx_debug.log")
|
||
|
||
|
||
def _get_sock():
|
||
global _SOCK
|
||
if _SOCK is None:
|
||
try:
|
||
_SOCK = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||
except Exception:
|
||
_SOCK = False
|
||
return _SOCK or None
|
||
|
||
|
||
def sbx_log(event: str, **fields) -> None:
|
||
"""写一条调试日志（任何异常都吞掉，绝不影响主流程）。"""
|
||
if not _ENABLED:
|
||
return
|
||
try:
|
||
ts = datetime.datetime.now().strftime("%H:%M:%S.%f")[:-3]
|
||
parts = [f"{k}={fields[k]!r}" for k in sorted(fields)]
|
||
line = f"[{ts}] [pid={os.getpid()}] [th={threading.current_thread().name}] {event}"
|
||
if parts:
|
||
line += " | " + " ".join(parts)
|
||
data = (line + "\n").encode("utf-8", "replace")
|
||
with _LOCK:
|
||
sock = _get_sock()
|
||
if sock is not None:
|
||
try:
|
||
sock.sendto(data, ("127.0.0.1", _PORT))
|
||
except Exception:
|
||
pass
|
||
try:
|
||
with open(_log_path(), "a", encoding="utf-8") as f:
|
||
f.write(line + "\n")
|
||
except Exception:
|
||
pass
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def sbx_log_stack(event: str, **fields) -> None:
|
||
"""写一条带调用堆栈的调试日志。"""
|
||
try:
|
||
stack = "".join(traceback.format_stack(limit=14)[:-2])
|
||
except Exception:
|
||
stack = "<no stack>"
|
||
fields["stack"] = stack
|
||
sbx_log(event, **fields)
|
||
|
||
|
||
def console_pids():
|
||
"""返回当前控制台附加的进程 pid 列表（Windows），用于判断子进程是否共享后端控制台。"""
|
||
if os.name != "nt":
|
||
return []
|
||
try:
|
||
import ctypes
|
||
arr = (ctypes.c_ulong * 64)()
|
||
got = ctypes.windll.kernel32.GetConsoleProcessList(arr, 64)
|
||
if not got:
|
||
return []
|
||
return [int(arr[i]) for i in range(min(got, 64))]
|
||
except Exception:
|
||
return ["?"]
|
||
|
||
|
||
def install_ctrl_event_probes(tag: str = "backend") -> None:
|
||
"""安装 SIGINT/SIGBREAK 探针：控制台事件到达时先记录日志。
|
||
|
||
SBX_DEBUG_SWALLOW_CTRL=1（默认）：记录后吞掉事件，进程存活，可继续观察后续行为；
|
||
=0：记录后恢复原处理器语义（SIGINT→KeyboardInterrupt；SIGBREAK→默认终止）。
|
||
只能在主线程安装，失败静默。
|
||
"""
|
||
global _PROBES_INSTALLED
|
||
if not _ENABLED or os.name != "nt" or _PROBES_INSTALLED:
|
||
return
|
||
if threading.current_thread() is not threading.main_thread():
|
||
sbx_log("CTRL-PROBES-SKIPPED", reason="not main thread", tag=tag)
|
||
return
|
||
|
||
def _make_handler(sig, name):
|
||
prev = signal.getsignal(sig)
|
||
|
||
def _handler(signum, frame):
|
||
sbx_log_stack("CTRL-EVENT-RECEIVED", signal=name, tag=tag, swallowed=_SWALLOW)
|
||
if _SWALLOW:
|
||
return
|
||
if callable(prev):
|
||
prev(signum, frame)
|
||
return
|
||
try:
|
||
signal.signal(sig, signal.SIG_DFL)
|
||
os.kill(os.getpid(), sig)
|
||
except Exception:
|
||
os._exit(128 + int(signum))
|
||
|
||
return _handler
|
||
|
||
try:
|
||
signal.signal(signal.SIGINT, _make_handler(signal.SIGINT, "SIGINT(CTRL_C)"))
|
||
except Exception:
|
||
pass
|
||
try:
|
||
sigbreak = getattr(signal, "SIGBREAK", None)
|
||
if sigbreak is not None:
|
||
signal.signal(sigbreak, _make_handler(sigbreak, "SIGBREAK(CTRL_BREAK)"))
|
||
except Exception:
|
||
pass
|
||
_PROBES_INSTALLED = True
|
||
sbx_log("CTRL-PROBES-INSTALLED", tag=tag, swallow=_SWALLOW, port=_PORT)
|