From dfe8a1a9af983b42cef3024e23b9c3219a169330 Mon Sep 17 00:00:00 2001 From: k2safe Date: Wed, 23 Sep 2026 10:18:49 +0800 Subject: [PATCH] test: add TLS mixed soak with paced clients --- CHANGELOG.md | 1 + README.md | 2 + scripts/verifyHTTP2.py | 5 +- transport/http2/mixedSoak_test.go | 304 ++++++++++++++++++++++++++++++ transport/http2/soak_test.go | 21 ++- 5 files changed, 324 insertions(+), 9 deletions(-) create mode 100644 transport/http2/mixedSoak_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index c32987a..0a8d1a4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## Unreleased +- 新增 `TestTLSMixedSoak` 与 `verifyHTTP2.py --scenario mixed`：8 条 TLS 连接、128 个 worker 持续混跑普通请求、慢上传和慢下载，校验 TLS/HTTP2、requestId、响应内容及传输摘要；分别统计三类请求与每条连接覆盖情况，并记录退出/GC 后的回收数据。 - 新增独立 `transport/http2` 实验 Runner：支持 h2c/TLS h2、连接上限、全局请求并发闸门、GOAWAY 排空、RST_STREAM 单流取消和生命周期统计；不改变 native HTTP/1.1 默认路径。 - HTTP/2 实验 Runner 增加标准 1 MiB 头部上限、默认 8 MiB 请求体上限、128 路同连接多路复用、2 MiB 慢读取流控验收、协议错误、RST_STREAM 和 GOAWAY 帧计数；已声明超限请求在进入业务 handler 前返回 413。 - 新增可配置时长的 HTTP/2 TLS soak：默认 3 秒，Linux 验收可通过 `GO_OMNI_HTTP2_SOAK_DURATION=30m` 覆盖，混合验证 128 路 stream、慢 handler、请求体、Keep-Alive 和退出回收。 diff --git a/README.md b/README.md index 4a14c8f..01f24ae 100644 --- a/README.md +++ b/README.md @@ -230,6 +230,8 @@ python3 scripts/verifyBusiness.py --version v0.6.0-preview HTTP/2 仍是显式 opt-in 实验适配器，不会改变 native 的 HTTP/1.1 默认路径。Linux TLS 长测使用 `python3 scripts/verifyHTTP2.py --duration 30m`，默认使用普通 Go 运行模式；`--race` 默认改为 60 秒，用于并发检查，其额外内存开销不能当成生产运行基线。 +多连接混合场景使用 `python3 scripts/verifyHTTP2.py --scenario mixed --duration 30m`：8 条独立 TLS 连接，每条连接 14 个普通请求 worker、1 个慢上传 worker、1 个慢下载 worker，共 128 个 worker。上传为 256 KiB、每 4 KiB 等待 2 ms；下载为 8 MiB、每次最多读取 64 KiB 后等待 20 ms。日志分别记录三类成功数、传输字节数、每条连接的覆盖情况，以及停止并释放测试数据后两次 GC 的采样。 + 该脚本默认把测试容器限制为 2 核、2 GiB 内存（禁用 swap）、256 个进程/线程，整个命令最多运行 40 分钟，中断和超时会清理本次容器。这些限制可用 `--cpus`、`--memory-mb`、`--pids-limit`、`--wall-timeout` 调整；共享服务器应先用 `--duration 60s` 短测，再逐步延长。每 15 秒输出的堆、RSS 和 FD 统计包含同一进程内的客户端与服务端。 Linux amd64 已完成 30 分钟、单 TLS 连接、128 个 worker 的普通模式长测：58,102,476 次请求，0 失败、0 协议错误，采样 RSS 峰值约 22.6 MiB，连接/活动请求归零且 FD 回到基线。完整采样和原始日志见 [HTTP/2 长测报告](performance/http2SoakLinux.json) 与 [原始日志](performance/http2SoakLinux.txt)，此前的 race 对照见 [短测记录](performance/http2SmokeLinux.json)。这项结果覆盖单连接复用；多连接、持续慢客户端、连接级流控和代理 GOAWAY/排空的完整矩阵仍未完成，HTTP/2 保持实验状态。 diff --git a/scripts/verifyHTTP2.py b/scripts/verifyHTTP2.py index 078d9cb..7599ab1 100755 --- a/scripts/verifyHTTP2.py +++ b/scripts/verifyHTTP2.py @@ -13,6 +13,7 @@ import uuid def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--image", default="golang:1.25") + parser.add_argument("--scenario", choices=("single", "mixed"), default="single", help="single TLS connection or eight connections with paced upload/download clients") parser.add_argument("--goproxy", default="", help="optional Go module proxy passed to the container") parser.add_argument("--duration", help="soak duration (default: 30m, or 60s with --race)") parser.add_argument("--timeout", default="35m", help="go test timeout (default: 35m)") @@ -54,11 +55,13 @@ def main() -> None: if args.race: command.append("-race") command += [ - "-v", "-timeout", args.timeout, "./transport/http2", "-run", "^TestTLSSoak$", "-count=1", + "-v", "-timeout", args.timeout, "./transport/http2", "-run", + "^TestTLSMixedSoak$" if args.scenario == "mixed" else "^TestTLSSoak$", "-count=1", ] print(f"HTTP/2 Linux soak image: {args.image}", flush=True) print(f"HTTP/2 TLS soak duration: {args.duration}", flush=True) print(f"HTTP/2 TLS soak mode: {'race' if args.race else 'regular'}", flush=True) + print(f"HTTP/2 TLS soak scenario: {args.scenario}", flush=True) print(f"HTTP/2 soak container: {name}; limits: {args.cpus:g} CPUs, {args.memory_mb} MiB, {args.pids_limit} tasks", flush=True) def interrupt(signum, frame): diff --git a/transport/http2/mixedSoak_test.go b/transport/http2/mixedSoak_test.go new file mode 100644 index 0000000..98d3cfa --- /dev/null +++ b/transport/http2/mixedSoak_test.go @@ -0,0 +1,304 @@ +package http2 + +import ( + "bytes" + "context" + "crypto/sha256" + "crypto/tls" + "fmt" + "io" + "net" + "net/http" + "runtime" + "strconv" + "sync" + "sync/atomic" + "testing" + "time" + + xhttp2 "golang.org/x/net/http2" +) + +type mixedConnectionCounts struct { + health atomic.Uint64 + slowUploads atomic.Uint64 + slowDownloads atomic.Uint64 +} + +// TestTLSMixedSoak shares each of eight connections between fourteen fast +// workers, one paced uploader and one paced downloader. The 8 MiB download is +// larger than x/net v0.25.0's 4 MiB initial stream window, requiring window +// updates to finish. This does not exhaust the connection-level receive window. +func TestTLSMixedSoak(t *testing.T) { + const connectionCount, workersPerConnection = 8, 16 + const uploadBytes, downloadBytes = 256 << 10, 8 << 20 + started := time.Now() + duration := soakDuration(t) + logSoakResources(t, started, "before", 0, 0, Stats{}) + t.Logf("scenario=mixed connections=%d workersPerConnection=%d uploadBytes=%d uploadChunkBytes=4096 uploadDelay=2ms downloadBytes=%d downloadChunkBytes=65536 downloadDelay=20ms", + connectionCount, workersPerConnection, uploadBytes, downloadBytes) + downloadPayload := bytes.Repeat([]byte("x"), downloadBytes) + expectedDownload := sha256.Sum256(downloadPayload) + expectedUpload := sha256.Sum256(downloadPayload[:uploadBytes]) + certificate, roots := testCertificate(t) + var serverFailures atomic.Uint64 + server := New(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + writer.Header().Set("X-Soak-Request-Id", request.Header.Get("X-Soak-Request-Id")) + switch request.URL.Path { + case "/slowUpload": + digest := sha256.New() + count, err := io.Copy(digest, request.Body) + if err != nil || count != uploadBytes || !bytes.Equal(digest.Sum(nil), expectedUpload[:]) { + serverFailures.Add(1) + http.Error(writer, "incomplete upload", http.StatusBadRequest) + return + } + _, _ = io.WriteString(writer, "uploaded") + case "/slowDownload": + writer.Header().Set("Content-Length", strconv.Itoa(downloadBytes)) + if n, err := writer.Write(downloadPayload); err != nil || n != downloadBytes { + serverFailures.Add(1) + } + default: + _, _ = io.WriteString(writer, "healthy") + } + })) + server.MaxConnections = connectionCount + server.MaxConcurrentRequests = connectionCount * workersPerConnection + server.MaxConcurrentStreams = workersPerConnection + server.MaxBodyBytes = 1 << 20 + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + done := make(chan error, 1) + go func() { + done <- server.ServeTLS(context.Background(), listener, &tls.Config{ + Certificates: []tls.Certificate{certificate}, MinVersion: tls.VersionTLS12, + }) + }() + var transports []*xhttp2.Transport + var clients []*http.Client + for index := 0; index < connectionCount; index++ { + transport := &xhttp2.Transport{ + TLSClientConfig: &tls.Config{RootCAs: roots, ServerName: "localhost", MinVersion: tls.VersionTLS12, NextProtos: []string{"h2"}}, + StrictMaxConcurrentStreams: true, + } + transports = append(transports, transport) + clients = append(clients, &http.Client{Transport: transport, Timeout: 20 * time.Second}) + } + var stopOnce sync.Once + stop := func() { + stopOnce.Do(func() { + for _, transport := range transports { + transport.CloseIdleConnections() + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := server.Shutdown(ctx); err != nil { + t.Errorf("mixed soak shutdown: %v", err) + } + _ = listener.Close() + if err := serveResult(t, done); err != nil { + t.Errorf("mixed soak serve: %v", err) + } + }) + } + defer stop() + + var counts [connectionCount]mixedConnectionCounts + var requests, failures atomic.Uint64 + var failureOnce sync.Once + var firstFailure string + recordFailure := func(err error) { + failures.Add(1) + failureOnce.Do(func() { firstFailure = err.Error() }) + } + logCounts := func(phase string) { + var health, uploads, downloads uint64 + for i := range counts { + health += counts[i].health.Load() + uploads += counts[i].slowUploads.Load() + downloads += counts[i].slowDownloads.Load() + } + t.Logf("mixedTotals phase=%s health=%d slowUploads=%d slowDownloads=%d uploadedBytes=%d downloadedBytes=%d serverFailures=%d", + phase, health, uploads, downloads, uploads*uploadBytes, downloads*downloadBytes, serverFailures.Load()) + } + monitorStop, monitorDone := make(chan struct{}), make(chan struct{}) + go func() { + defer close(monitorDone) + ticker := time.NewTicker(15 * time.Second) + defer ticker.Stop() + for { + select { + case <-ticker.C: + logSoakResources(t, started, "running", requests.Load(), failures.Load(), server.Stats()) + logCounts("running") + case <-monitorStop: + return + } + } + }() + var monitorOnce sync.Once + stopMonitor := func() { monitorOnce.Do(func() { close(monitorStop); <-monitorDone }) } + defer stopMonitor() + + deadline := time.Now().Add(duration) + var workers sync.WaitGroup + for worker := 0; worker < connectionCount*workersPerConnection; worker++ { + workers.Add(1) + go func(worker int) { + defer workers.Done() + connection := worker / workersPerConnection + kind := worker % workersPerConnection + for sequence := uint64(0); time.Now().Before(deadline) && failures.Load() == 0; sequence++ { + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + path, method := "/health", http.MethodGet + var body io.Reader + if kind == 0 { + path, method = "/slowUpload", http.MethodPost + body = &pacedUpload{ctx: ctx, remaining: uploadBytes} + } else if kind == 1 { + path = "/slowDownload" + } + requestID := strconv.Itoa(worker) + ":" + strconv.FormatUint(sequence, 10) + request, err := http.NewRequestWithContext(ctx, method, "https://"+listener.Addr().String()+path, body) + if err == nil { + request.Header.Set("X-Soak-Request-Id", requestID) + err = checkMixedResponse(ctx, clients[connection], request, kind, downloadBytes, expectedDownload) + } + cancel() + if err != nil { + recordFailure(fmt.Errorf("connection=%d path=%s: %w", connection, path, err)) + continue + } + requests.Add(1) + switch kind { + case 0: + counts[connection].slowUploads.Add(1) + case 1: + counts[connection].slowDownloads.Add(1) + default: + counts[connection].health.Add(1) + } + } + }(worker) + } + workers.Wait() + stopMonitor() + stop() + logSoakResources(t, started, "after", requests.Load(), failures.Load(), server.Stats()) + logCounts("after") + var completed uint64 + for index := range counts { + c := &counts[index] + completed += c.health.Load() + c.slowUploads.Load() + c.slowDownloads.Load() + t.Logf("connection=%d health=%d slowUploads=%d slowDownloads=%d", index, c.health.Load(), c.slowUploads.Load(), c.slowDownloads.Load()) + if c.health.Load() == 0 || c.slowUploads.Load() == 0 || c.slowDownloads.Load() == 0 { + t.Errorf("connection %d did not complete all workload types", index) + } + } + if completed != requests.Load() { + t.Errorf("mixed workload accounting: completed=%d requests=%d", completed, requests.Load()) + } + // Release workload fixtures and idle client pools before measuring retained + // objects. Two GCs also allow sync.Pool's previous generation to be collected. + downloadPayload, clients, transports = nil, nil, nil + runtime.GC() + runtime.GC() + logSoakResources(t, started, "afterReclaim", requests.Load(), failures.Load(), server.Stats()) + if failures.Load() != 0 || serverFailures.Load() != 0 { + t.Fatalf("mixed soak failures=%d serverFailures=%d firstFailure=%q", failures.Load(), serverFailures.Load(), firstFailure) + } + if stats := server.Stats(); stats.AcceptedConnections != connectionCount || stats.ActiveConnections != 0 || stats.ActiveRequests != 0 || stats.ProtocolErrors != 0 || stats.RejectedRequests != 0 || stats.RejectedConnections != 0 { + t.Fatalf("mixed soak connections, errors or recovery: %+v", stats) + } +} + +func checkMixedResponse(ctx context.Context, client *http.Client, request *http.Request, kind, downloadBytes int, expectedHash [sha256.Size]byte) error { + response, err := client.Do(request) + if err != nil { + return err + } + defer response.Body.Close() + if response.StatusCode != http.StatusOK || response.ProtoMajor != 2 || response.TLS == nil || response.TLS.NegotiatedProtocol != "h2" || response.Header.Get("X-Soak-Request-Id") != request.Header.Get("X-Soak-Request-Id") { + return fmt.Errorf("unexpected status, protocol or request ID: %s", response.Status) + } + if kind != 1 { + body, err := io.ReadAll(io.LimitReader(response.Body, 1024)) + expected := "healthy" + if kind == 0 { + expected = "uploaded" + } + if err != nil || string(body) != expected { + return fmt.Errorf("invalid response %q: %v", body, err) + } + return nil + } + digest := sha256.New() + buffer := make([]byte, 64<<10) + total := 0 + for { + n, err := response.Body.Read(buffer) + total += n + if total > downloadBytes { + return fmt.Errorf("download exceeds %d bytes", downloadBytes) + } + if n > 0 { + _, _ = digest.Write(buffer[:n]) + if pauseErr := pauseMixedClient(ctx, 20*time.Millisecond); pauseErr != nil { + return pauseErr + } + } + if err == io.EOF { + break + } + if err != nil { + return err + } + } + if total != downloadBytes || !bytes.Equal(digest.Sum(nil), expectedHash[:]) { + return fmt.Errorf("download length or digest mismatch: %d bytes", total) + } + return nil +} + +type pacedUpload struct { + ctx context.Context + remaining int +} + +func (r *pacedUpload) Read(buffer []byte) (int, error) { + if r.remaining == 0 { + return 0, io.EOF + } + if len(buffer) == 0 { + return 0, nil + } + if err := pauseMixedClient(r.ctx, 2*time.Millisecond); err != nil { + return 0, err + } + n := len(buffer) + if n > 4096 { + n = 4096 + } + if n > r.remaining { + n = r.remaining + } + for index := 0; index < n; index++ { + buffer[index] = 'x' + } + r.remaining -= n + return n, nil +} + +func pauseMixedClient(ctx context.Context, delay time.Duration) error { + timer := time.NewTimer(delay) + defer timer.Stop() + select { + case <-timer.C: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} diff --git a/transport/http2/soak_test.go b/transport/http2/soak_test.go index b11ef0d..2dd5d30 100644 --- a/transport/http2/soak_test.go +++ b/transport/http2/soak_test.go @@ -22,14 +22,7 @@ import ( // verification. Linux acceptance runs set GO_OMNI_HTTP2_SOAK_DURATION=30m. func TestTLSSoak(t *testing.T) { started := time.Now() - duration := 3 * time.Second - if value := os.Getenv("GO_OMNI_HTTP2_SOAK_DURATION"); value != "" { - parsed, err := time.ParseDuration(value) - if err != nil || parsed <= 0 { - t.Fatalf("invalid GO_OMNI_HTTP2_SOAK_DURATION %q: %v", value, err) - } - duration = parsed - } + duration := soakDuration(t) logSoakResources(t, started, "before", 0, 0, Stats{}) certificate, roots := testCertificate(t) @@ -172,6 +165,18 @@ func TestTLSSoak(t *testing.T) { } } +func soakDuration(t *testing.T) time.Duration { + t.Helper() + if value := os.Getenv("GO_OMNI_HTTP2_SOAK_DURATION"); value != "" { + duration, err := time.ParseDuration(value) + if err != nil || duration <= 0 { + t.Fatalf("invalid GO_OMNI_HTTP2_SOAK_DURATION %q: %v", value, err) + } + return duration + } + return 3 * time.Second +} + // These measurements cover the client and server in the same test process. // RSS includes race runtime allocations that do not appear in Go heap stats. func logSoakResources(t *testing.T, started time.Time, phase string, requests, failures uint64, stats Stats) {