mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-09-21 01:18:08 +00:00
9f80289a89
本轮压测是从 /var/tmp 里现写脚本跑的，能复现才能长期用 —— 固化到这。
为什么要有它（与 go test 的分工）：单测都在进程内、参数显式给，抓不到集成面问题；
这一轮它抓到了三个只有真二进制才暴露的问题（DataDir 漏接线、插件通道未登记为
inputch、create 后子不开工），以及两处生成器缺陷（旧 SDK 导致生成工程编译失败；
模板工程从不示范通道登记）。
- `launch.sh`：在**私有 netns** 里同时起 mock LLM 与内核实例。
必须隔离的原因：生产实例占着 *:8080/*:9890/*:9876 且插件设了 SO_REUSEADDR，
同机再起一个实例会在 127.0.0.1 上与之并存绑定（首次实测抢到 9890 约 1 分钟）；
netns 里只有 lo，结构上不可能碰到生产端口。unix socket 是文件系统对象，
所以驱动脚本在 netns 外照样能连。
- `mockllm.py`：OpenAI 兼容 mock（可控延迟 + SSE 分块 + 按 `!resident`/`!notify`
标记回工具调用），让无外网也能跑、且**流式段可被中断**（压抢占路径的前提）。
注意必须实现 `do_HEAD`：内核探活用 HEAD，501 会被判不可达 → degraded → rollback 循环。
- `stress.py`：多并发连接轰炸（单连接是串行的，造不出队列压力）+ 中断线程 + 峰值采样。
- `kcli.py`：CLI socket 客户端（`/auth` → `/kernel` / 广播）。
- README：前置配置（含探活端点、rollback 关掉）、用法、指标读法、已知坑。
实测结果（详见 6e60351 提交信息）：密集 274 任务 executed=274/rejected=0；
稀疏中断下 suspended=resumed=preempted=27；驻留子全链路 + 优雅退出不留孤儿。
82 lines
2.6 KiB
Python
82 lines
2.6 KiB
Python
import json, os, socket, sys, threading, time
|
||
|
||
SOCK = sys.argv[1]
|
||
|
||
KEY = os.environ.get("KCLI_KEY", "")
|
||
|
||
def connect(auth=True):
|
||
s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
||
s.settimeout(20)
|
||
s.connect(SOCK)
|
||
if auth and KEY:
|
||
s.sendall(("/auth %s\n" % KEY).encode())
|
||
buf = b""
|
||
while b"\n" not in buf:
|
||
c = s.recv(65536)
|
||
if not c:
|
||
break
|
||
buf += c
|
||
if b"authenticated" not in buf:
|
||
raise RuntimeError("auth failed: %r" % buf[:120])
|
||
return s
|
||
|
||
def kernel_stats(sock=None, timeout=15):
|
||
"""连一条新连接查 /kernel，返回解析后的 dict。"""
|
||
s = sock or connect()
|
||
try:
|
||
s.sendall(b"/kernel\n")
|
||
buf = b""
|
||
deadline = time.time() + timeout
|
||
while time.time() < deadline:
|
||
chunk = s.recv(65536)
|
||
if not chunk:
|
||
break
|
||
buf += chunk
|
||
for line in buf.split(b"\n"):
|
||
if not line.strip():
|
||
continue
|
||
try:
|
||
obj = json.loads(line.decode("utf-8", "replace"))
|
||
except Exception:
|
||
continue
|
||
if obj.get("type") == "response" and obj.get("content", "").lstrip().startswith("{"):
|
||
return json.loads(obj["content"])
|
||
raise RuntimeError("no /kernel response")
|
||
finally:
|
||
if sock is None:
|
||
s.close()
|
||
|
||
def blast(n_queued, n_interrupt, tag):
|
||
"""一条连接狂发：排队输入与中断交错。响应在后台线程里丢弃，避免缓冲阻塞。"""
|
||
s = connect()
|
||
threading.Thread(target=lambda: drain(s), daemon=True).start()
|
||
sent = 0
|
||
for i in range(max(n_queued, n_interrupt)):
|
||
if i < n_queued:
|
||
s.sendall(("queued-%s-%d\n" % (tag, i)).encode())
|
||
sent += 1
|
||
if i < n_interrupt:
|
||
s.sendall(("/interrupt intr-%s-%d\n" % (tag, i)).encode())
|
||
sent += 1
|
||
return sent
|
||
|
||
def drain(s):
|
||
try:
|
||
while s.recv(65536):
|
||
pass
|
||
except Exception:
|
||
pass
|
||
|
||
if __name__ == "__main__":
|
||
cmd = sys.argv[2] if len(sys.argv) > 2 else "stats"
|
||
if cmd == "stats":
|
||
st = kernel_stats()
|
||
if len(sys.argv) > 3 and sys.argv[3] == "full":
|
||
print(json.dumps(st, ensure_ascii=False))
|
||
else:
|
||
sched = st.get("scheduler") or st
|
||
print(json.dumps(sched, ensure_ascii=False))
|
||
elif cmd == "blast":
|
||
n = blast(int(sys.argv[3]), int(sys.argv[4]), sys.argv[5] if len(sys.argv) > 5 else "b")
|
||
print("sent=%d" % n)
|