From 7bfea74cc0232d3a09137a9d5eabc53d437f14c6 Mon Sep 17 00:00:00 2001 From: Blizzard Date: Fri, 26 Jun 2026 11:46:50 +0800 Subject: [PATCH] =?UTF-8?q?feat(perf):=20=E5=AE=B9=E9=87=8F=E5=8E=8B?= =?UTF-8?q?=E6=B5=8B=E5=99=A8=20+=20=E5=B9=B3=E5=8F=B0=E5=A4=A9=E8=8A=B1?= =?UTF-8?q?=E6=9D=BF=E5=AE=9E=E6=B5=8B=E6=9B=B2=E7=BA=BF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit cmd/loadtest：闭环加压器，阶梯并发，经 SSE 流检测完成（非轮询，避免轮询放大把网关 压成假瓶颈），输出吞吐 + 延迟分位。配套两个 benchmark 开关： - dispatcher LLM_FORCE_STUB=1 + LLM_STUB_TTFT_MS/INTERTOKEN_MS=0：绕开真实 LLM 推理， 量平台自身全链路天花板（不烧 token、不被模型节奏掩盖）。 - gateway RATE_LIMIT_PER_MIN 可配（缺省 120）：压测放开限流。 实测（单 dispatcher、并发64、stub）： - 单任务纯平台开销 ~42ms；吞吐峰值 ~110 全链路任务/秒（饱和点 并发32–64）； 并发128 优雅降速，256 硬崩。 - 吞吐瓶颈不是 DB 连接数（池 25→80 吞吐不变），是每任务多跳管线综合成本； 256 崩根因为 DSN 用 localhost、pgx 每连接解析 → 连接churn致 DNS 取消（易修）。 - 关键判断：平台 42ms ≪ LLM 秒级出答案，平台不是瓶颈、GPU 才是；横向拆服务喂满 GPU 集群 的意义成立。细节与曲线见 project_analysis「容量实测」。 stub 延迟改 env 可配（默认值不变）；不影响线上路径。dispatcher+gateway build/vet/test 全绿。 Co-Authored-By: Claude Opus 4.8 (1M context) --- project_analysis.md | 31 +++- sundynix-dispatcher/cmd/loadtest/main.go | 173 ++++++++++++++++++ sundynix-dispatcher/internal/llm/pool.go | 39 +++- sundynix-gateway/internal/middleware/auth.go | 11 ++ .../internal/middleware/guardrail.go | 4 +- 5 files changed, 246 insertions(+), 12 deletions(-) create mode 100644 sundynix-dispatcher/cmd/loadtest/main.go diff --git a/project_analysis.md b/project_analysis.md index ca68218..77f6fe1 100644 --- a/project_analysis.md +++ b/project_analysis.md @@ -203,7 +203,7 @@ Harness = 围绕 LLM 的可靠性 / 安全 / 质量治理层。4 个组件均为 | 优先级 | 项目 | 说明 | |:------:|------|------| -| P0 | 真实负载 + 长稳压测，给容量曲线 | "生产级并发"需数据背书 | +| ~~P0~~ 🟡 | ~~容量压测曲线~~（首版）：`cmd/loadtest` 闭环加压器 + 平台天花板实测（见下「容量实测」）。剩长稳/真实 LLM 大压 | "生产级并发"需数据背书 | | ~~P0~~ ✅ | ~~本地模型(vLLM/Ollama) + reasoning_content 适配~~：Pool provider 感知（Ollama/vLLM 自动补 /v1 + 占位 key），统一走 OpenAI 兼容路径；ChatStream 加 onReasoning，思考过程 surface 到 exec 轨迹（不污染答案）。控制台加 ollama 选项。live：Ollama qwen2.5:0.5b 端到端出答案；deepseek-v4-pro「推理过程」入轨迹 | 对齐生产 Qwen | | P1 | 高可用：网关/调度多副本 + NATS 集群 + 自愈 | 解单点 | | P1 | 备份/灾备演练（PG/Milvus/Neo4j） | 数据安全 | @@ -215,5 +215,34 @@ Harness = 围绕 LLM 的可靠性 / 安全 / 质量治理层。4 个组件均为 --- +## 容量实测（首版，2026-06-26） + +**方法**：`cmd/loadtest` 闭环加压（阶梯并发，SSE 流检测完成、非轮询，避免轮询放大）。 +dispatcher 开 `LLM_FORCE_STUB=1 LLM_STUB_*_MS=0` 绕开真实 LLM 推理，量**平台自身全链路天花板** +（网关→NATS/JetStream→调度→图执行→工具RTT→Token回流→PG状态写），单 dispatcher、DISPATCHER_CONCURRENCY=64。 + +| 并发 | 吞吐/s | p50 | p95 | 备注 | +|---:|---:|---:|---:|---| +| 1 | 23 | 42ms | 46ms | **单任务纯平台开销 ~42ms** | +| 8 | 76 | 103ms | 115ms | | +| 16 | 96 | 161ms | 176ms | | +| 32 | 112 | 269ms | 345ms | **吞吐峰值** | +| 64 | 112 | 557ms | 676ms | 饱和（延迟翻倍、吞吐不增）| +| 128 | 102 | 1173ms | 1590ms | 过载、优雅降速 | +| 256 | 0 | — | — | 硬崩（连接/DNS 抖动）| + +**结论**： +1. **单节点平台天花板 ≈ 110 全链路任务/秒**，单任务固有开销 ~42ms，饱和点 ~并发 32–64。 +2. **吞吐瓶颈不是 DB 连接数**（池 25→80 吞吐不变），而是每任务多跳管线（JetStream fsync + 多次 + NATS 回写 + Redis 录制 + 3 次 PG 状态写）的综合成本。 +3. **C=256 硬崩**根因：DSN 用 `localhost`，pgx 每新建连接解析一次，极端并发下连接churn致 DNS 取消 + → 易修（DSN 换 `127.0.0.1` / 连接复用）。 +4. **关键判断**：平台开销 42ms ≪ 真实 LLM 出一轮答案的秒级延迟，**平台不是瓶颈，GPU 才是**。 + 单 GPU 跑 Qwen 32B 约出几十轮/秒，远低于平台的 ~110/s → 横向拆服务（队列组多副本喂满 GPU 集群） + 的意义成立；要提总吞吐应加 dispatcher 副本与 GPU，而非优化平台代码。 +5. 待补：长稳（小时级）压测看内存/句柄泄漏；真实 LLM 端到端容量（受 provider 限流，需自部署后测）。 + +--- + *本报告由 Opus 4.8 基于全程搭建经验 + 本次实测撰写；刻意压低自评水分。功能盘点可信， 评分按成熟度折扣阅读。如与旧版（自动分析）冲突，以本版为准。* diff --git a/sundynix-dispatcher/cmd/loadtest/main.go b/sundynix-dispatcher/cmd/loadtest/main.go new file mode 100644 index 0000000..e9a8c2d --- /dev/null +++ b/sundynix-dispatcher/cmd/loadtest/main.go @@ -0,0 +1,173 @@ +// Command loadtest 是平台容量压测器：闭环加压、阶梯并发，输出吞吐/延迟分位曲线。 +// +// 它只通过网关 HTTP API 打真实流量（登录 → POST /tasks → SSE 流等收尾）， +// 故压的是「网关→NATS→调度→工具RTT→状态回流」全链路。配合 dispatcher 的 LLM_FORCE_STUB=1 +// + LLM_STUB_*_MS=0 可绕开真实 LLM 推理，量出平台自身吞吐天花板（而非被模型节奏掩盖）。 +// +// 用法： +// +// go run ./cmd/loadtest -url http://localhost:8080 -email keytest@local.dev -pass keytest123456 \ +// -levels 1,2,4,8,16,32,64 -dur 12s +package main + +import ( + "bytes" + "context" + "encoding/json" + "flag" + "fmt" + "io" + "net/http" + "os" + "sort" + "strconv" + "strings" + "sync" + "sync/atomic" + "time" +) + +func main() { + var ( + base = flag.String("url", "http://localhost:8080", "网关地址") + email = flag.String("email", "keytest@local.dev", "登录邮箱") + pass = flag.String("pass", "keytest123456", "登录密码") + levelStr = flag.String("levels", "1,2,4,8,16,32,64", "并发阶梯（逗号分隔）") + dur = flag.Duration("dur", 12*time.Second, "每个阶梯加压时长") + prompt = flag.String("prompt", "用一句话介绍杭州", "任务输入") + ) + flag.Parse() + + token, err := login(*base, *email, *pass) + if err != nil { + fmt.Printf("登录失败: %v\n", err) + os.Exit(1) + } + body := taskBody(*prompt) + + fmt.Printf("== 平台容量压测 == 目标 %s 每阶梯 %s (FORCE_STUB 下为平台天花板)\n", *base, *dur) + fmt.Printf("%-6s %-10s %-10s %-9s %-9s %-9s %-7s\n", "并发", "完成", "吞吐/s", "p50ms", "p95ms", "max ms", "错误") + for _, ls := range strings.Split(*levelStr, ",") { + c, err := strconv.Atoi(strings.TrimSpace(ls)) + if err != nil || c <= 0 { + continue + } + r := runLevel(*base, token, body, c, *dur) + fmt.Printf("%-6d %-10d %-10.1f %-9d %-9d %-9d %-7d\n", + c, r.done, float64(r.done)/dur.Seconds(), r.p50, r.p95, r.max, r.errs) + } +} + +type result struct { + done, errs int64 + p50, p95, max int64 +} + +// runLevel 在并发 c 下闭环加压 dur：c 个 worker 不停 submit→等终态→再来，统计完成数与 e2e 延迟。 +func runLevel(base, token string, body []byte, c int, dur time.Duration) result { + ctx, cancel := context.WithTimeout(context.Background(), dur) + defer cancel() + var done, errs int64 + var mu sync.Mutex + var lats []int64 + var wg sync.WaitGroup + for i := 0; i < c; i++ { + wg.Add(1) + go func() { + defer wg.Done() + cl := &http.Client{Timeout: 60 * time.Second} + for ctx.Err() == nil { + t0 := time.Now() + if runOne(ctx, cl, base, token, body) { + ms := time.Since(t0).Milliseconds() + mu.Lock() + lats = append(lats, ms) + mu.Unlock() + atomic.AddInt64(&done, 1) + } else { + atomic.AddInt64(&errs, 1) + } + } + }() + } + wg.Wait() + return result{done: done, errs: errs, p50: pct(lats, 50), p95: pct(lats, 95), max: pct(lats, 100)} +} + +// runOne 提交一个任务并经 SSE token 流等到收尾（流结束即任务完成）；返回是否成功。 +// 用流而非轮询：每任务 1 条推送连接，避免轮询放大把网关自身压成瓶颈，量出真实管线吞吐。 +func runOne(ctx context.Context, cl *http.Client, base, token string, body []byte) bool { + id, err := submit(ctx, cl, base, token, body) + if err != nil || id == "" { + return false + } + req, _ := http.NewRequestWithContext(ctx, "GET", base+"/api/v1/tasks/"+id+"/stream", nil) + req.Header.Set("Authorization", "Bearer "+token) + resp, err := cl.Do(req) + if err != nil { + return false + } + defer resp.Body.Close() + // 读到 EOF：网关在 token 流结束（任务收尾）时关闭 SSE。 + _, err = io.Copy(io.Discard, resp.Body) + return err == nil && ctx.Err() == nil +} + +func submit(ctx context.Context, cl *http.Client, base, token string, body []byte) (string, error) { + req, _ := http.NewRequestWithContext(ctx, "POST", base+"/api/v1/tasks", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Bearer "+token) + resp, err := cl.Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusAccepted { + io.Copy(io.Discard, resp.Body) + return "", fmt.Errorf("submit %d", resp.StatusCode) + } + var r struct { + TaskID string `json:"task_id"` + } + json.NewDecoder(resp.Body).Decode(&r) + return r.TaskID, nil +} + +func login(base, email, pass string) (string, error) { + b, _ := json.Marshal(map[string]string{"email": email, "password": pass}) + resp, err := http.Post(base+"/api/v1/auth/login", "application/json", bytes.NewReader(b)) + if err != nil { + return "", err + } + defer resp.Body.Close() + var r struct { + Token string `json:"token"` + } + json.NewDecoder(resp.Body).Decode(&r) + if r.Token == "" { + return "", fmt.Errorf("no token (status %d)", resp.StatusCode) + } + return r.Token, nil +} + +func taskBody(prompt string) []byte { + graph := map[string]any{ + "nodes": []any{ + map[string]any{"id": "i", "kind": "input", "config": map[string]any{"text": prompt}}, + map[string]any{"id": "a", "kind": "agent", "config": map[string]any{"system": "你是简洁的助手"}}, + }, + "edges": []any{map[string]any{"source": "i", "target": "a"}}, + } + b, _ := json.Marshal(map[string]any{"graph": graph, "meta": map[string]any{}}) + return b +} + +// pct 返回延迟切片的第 p 百分位（p=100 即 max）。 +func pct(v []int64, p int) int64 { + if len(v) == 0 { + return 0 + } + sort.Slice(v, func(i, j int) bool { return v[i] < v[j] }) + idx := (p * (len(v) - 1)) / 100 + return v[idx] +} diff --git a/sundynix-dispatcher/internal/llm/pool.go b/sundynix-dispatcher/internal/llm/pool.go index c659a84..9bfb031 100644 --- a/sundynix-dispatcher/internal/llm/pool.go +++ b/sundynix-dispatcher/internal/llm/pool.go @@ -7,6 +7,8 @@ import ( "context" "fmt" "io" + "os" + "strconv" "strings" "sync" "time" @@ -39,10 +41,14 @@ type Pool struct { func NewPool() *Pool { return &Pool{} } +// forceStub 报告是否强制走降级桩（LLM_FORCE_STUB=1）——压测平台自身吞吐时用， +// 绕开真实 LLM 推理与计费，只压全链路 plumbing（网关→NATS→调度→工具RTT→回流）。 +func forceStub() bool { return os.Getenv("LLM_FORCE_STUB") == "1" } + // SetConfig 热更新后端配置：重建 ChatModel 实例（控制面变更时调用）。 func (p *Pool) SetConfig(cfg *contract.ModelConfig) { var cm model.BaseChatModel - if cfg != nil && cfg.Ready() { + if cfg != nil && cfg.Ready() && !forceStub() { built, err := buildChatModel(cfg) if err != nil { fmt.Printf("[llm] 构建 ChatModel 失败（降级桩运行）: %v\n", err) @@ -222,17 +228,30 @@ func toSchema(msgs []ChatMessage) []*schema.Message { // ---- 占位降级（未配置后端时）---- // 占位参数：模拟真实后端的 TTFT(首 token 延迟) 与逐 token 间隔。 -const ( - timeToFirstToken = 700 * time.Millisecond - interTokenDelay = 60 * time.Millisecond +// 可经 env 调整（压测平台吞吐时设 0 → 近瞬时桩，测出 plumbing 天花板而非被桩节奏掩盖）。 +var ( + timeToFirstToken = envDuration("LLM_STUB_TTFT_MS", 700*time.Millisecond) + interTokenDelay = envDuration("LLM_STUB_INTERTOKEN_MS", 60*time.Millisecond) ) +// envDuration 读毫秒 env（允许 0），缺省回退 def。 +func envDuration(key string, def time.Duration) time.Duration { + if v := os.Getenv(key); v != "" { + if n, err := strconv.Atoi(v); err == nil && n >= 0 { + return time.Duration(n) * time.Millisecond + } + } + return def +} + // StreamText 按节奏把给定文本流式回调（未配置真实后端时的降级桩）。 func (p *Pool) StreamText(ctx context.Context, text string, onToken func([]byte)) error { - select { - case <-ctx.Done(): - return ctx.Err() - case <-time.After(timeToFirstToken): + if timeToFirstToken > 0 { + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(timeToFirstToken): + } } for _, tok := range tokenize(text) { select { @@ -241,7 +260,9 @@ func (p *Pool) StreamText(ctx context.Context, text string, onToken func([]byte) default: } onToken([]byte(tok)) - time.Sleep(interTokenDelay) + if interTokenDelay > 0 { + time.Sleep(interTokenDelay) + } } return nil } diff --git a/sundynix-gateway/internal/middleware/auth.go b/sundynix-gateway/internal/middleware/auth.go index 8ad1b47..53630f5 100644 --- a/sundynix-gateway/internal/middleware/auth.go +++ b/sundynix-gateway/internal/middleware/auth.go @@ -3,6 +3,7 @@ package middleware import ( "net/http" "os" + "strconv" "strings" "github.com/gin-gonic/gin" @@ -83,3 +84,13 @@ func splitEnv(key string) []string { } return out } + +// envInt 读正整数环境变量，缺省回退 def。 +func envInt(key string, def int) int { + if v := os.Getenv(key); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + return n + } + } + return def +} diff --git a/sundynix-gateway/internal/middleware/guardrail.go b/sundynix-gateway/internal/middleware/guardrail.go index d0f1fa8..a1f33c5 100644 --- a/sundynix-gateway/internal/middleware/guardrail.go +++ b/sundynix-gateway/internal/middleware/guardrail.go @@ -42,9 +42,9 @@ func Guardrail() gin.HandlerFunc { } // RateLimit 基于 Redis 的会话级限流（按客户端 IP，每分钟上限）。 -// Redis 降级时 Allow 始终放行，不阻断业务。 +// 上限经 RATE_LIMIT_PER_MIN 配置（缺省 120）；压测可调高。Redis 降级时始终放行，不阻断业务。 func RateLimit(cache *store.Redis) gin.HandlerFunc { - const perMinute = 120 + perMinute := int64(envInt("RATE_LIMIT_PER_MIN", 120)) return func(c *gin.Context) { ok, _ := cache.Allow(c.Request.Context(), c.ClientIP(), perMinute, time.Minute) if !ok {