Skip to content

Commit 2f23d14

Browse files
author
linkxzhou
committed
fix: 修复分布式压测的问题
1 parent 6d79ca3 commit 2f23d14

8 files changed

Lines changed: 239 additions & 11 deletions

File tree

‎cli_options.go‎

Lines changed: 44 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -84,7 +84,12 @@ func ParseConfig(args []string, getenv func(string) string, stderr io.Writer) (O
8484
fs.Var(stringSliceFlag{&opts.Headers}, "H", "Custom header (repeatable)")
8585
fs.Var(stringSliceFlag{&opts.WorkerAddrs}, "W", "Worker address (repeatable)")
8686
fs.Var(stringSliceFlag{&opts.WorkerAddrs}, "w", "Worker address alias")
87-
if err := fs.Parse(args); err != nil {
87+
// flag.Parse stops at the first non-flag argument, so a command like
88+
// "http_bench -n 100 -m POST <url> -body '{}' -W host:port" would
89+
// silently swallow every flag after the URL (the worker then never
90+
// receives a task). Reorder positional args to the end so flags and
91+
// the positional URL can be freely mixed (README documents this style).
92+
if err := fs.Parse(reorderArgs(args)); err != nil {
8893
return Options{}, err
8994
}
9095
// Detect whether -d was explicitly set on the command line. When the
@@ -133,6 +138,44 @@ func ParseConfig(args []string, getenv func(string) string, stderr io.Writer) (O
133138
return opts, nil
134139
}
135140

141+
// boolFlags lists the flags that take no value argument. reorderArgs uses it
142+
// to decide whether the token following a flag is its value or a positional
143+
// argument.
144+
var boolFlags = map[string]bool{
145+
"disable-compression": true,
146+
"disable-keepalive": true,
147+
"insecure": true,
148+
"example": true,
149+
"h": true,
150+
"help": true,
151+
}
152+
153+
// reorderArgs moves positional (non-flag) arguments behind all flags so that
154+
// flag.Parse — which stops consuming at the first non-flag token — parses
155+
// every flag. A flag consumes the next token as its value unless it is a
156+
// boolean flag or uses the -flag=value form.
157+
func reorderArgs(args []string) []string {
158+
var flags, positional []string
159+
for i := 0; i < len(args); i++ {
160+
a := args[i]
161+
if len(a) > 1 && a[0] == '-' {
162+
flags = append(flags, a)
163+
name := strings.TrimLeft(a, "-")
164+
if strings.Contains(name, "=") || boolFlags[name] {
165+
continue
166+
}
167+
// Value flag: consume the next token as its value.
168+
if i+1 < len(args) {
169+
i++
170+
flags = append(flags, args[i])
171+
}
172+
continue
173+
}
174+
positional = append(positional, a)
175+
}
176+
return append(flags, positional...)
177+
}
178+
136179
// stringSliceFlag implements flag.Value for repeatable string flags.
137180
type stringSliceFlag struct{ target *[]string }
138181

‎cli_test.go‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -122,3 +122,36 @@ func TestParseConfig_ConflictingURL(t *testing.T) {
122122
t.Errorf("unexpected error: %v", err)
123123
}
124124
}
125+
126+
// TestParseConfig_FlagsAfterPositionalURL covers the documented invocation
127+
// style where flags follow the positional URL (README distributed example).
128+
// flag.Parse would otherwise stop at the URL and silently drop -body/-W.
129+
func TestParseConfig_FlagsAfterPositionalURL(t *testing.T) {
130+
opts, err := ParseConfig([]string{
131+
"-n", "10000", "-c", "10", "-d", "30s", "-m", "POST",
132+
"http://www.baidu.com/api/test",
133+
"-body", `{"key":"value"}`, "-W", "127.0.0.1:12710", "-W", "127.0.0.1:12711",
134+
"-disable-keepalive", "-insecure=false",
135+
}, nil, &bytes.Buffer{})
136+
if err != nil {
137+
t.Fatalf("ParseConfig: %v", err)
138+
}
139+
if opts.URL != "http://www.baidu.com/api/test" {
140+
t.Errorf("URL = %q", opts.URL)
141+
}
142+
if opts.Body != `{"key":"value"}` {
143+
t.Errorf("Body = %q", opts.Body)
144+
}
145+
if len(opts.WorkerAddrs) != 2 || opts.WorkerAddrs[0] != "127.0.0.1:12710" || opts.WorkerAddrs[1] != "127.0.0.1:12711" {
146+
t.Errorf("WorkerAddrs = %v", opts.WorkerAddrs)
147+
}
148+
if !opts.DisableKeepAlives {
149+
t.Error("DisableKeepAlives should be true")
150+
}
151+
if opts.Insecure {
152+
t.Error("Insecure should be false")
153+
}
154+
if opts.Method != "POST" || opts.Count != 10000 || opts.Concurrency != 10 {
155+
t.Errorf("unexpected opts: %+v", opts)
156+
}
157+
}

‎http_bench.go‎

Lines changed: 19 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -84,9 +84,16 @@ func handleDistributedWorkers(params transport.HttpbenchParameters, dc distConfi
8484
result.ErrMsg = fmt.Sprintf("parameter marshaling failed: %v", err)
8585
return result, nil
8686
}
87-
distributedHTTPTimeout := params.Duration + 60*time.Second
87+
// With -n only (no -d) the run length is unpredictable, so leave the
88+
// timeout to PostAllWorkers' default instead of a too-tight 60s cap.
89+
var distributedHTTPTimeout time.Duration
90+
if params.Duration > 0 {
91+
distributedHTTPTimeout = params.Duration + 60*time.Second
92+
}
8893
distributed.APIKey = dc.authKey
89-
distributedResult, err := distributed.PostAllWorkers(dc.workerAddrs, jsonBody, distributedHTTPTimeout)
94+
workerURLs := normalizeWorkerAddrs(dc.workerAddrs, dc.workerAPIPath)
95+
logging.Info(seqId, "dispatching task to workers: %v", workerURLs)
96+
distributedResult, err := distributed.PostAllWorkers(workerURLs, jsonBody, distributedHTTPTimeout)
9097
if err != nil {
9198
logging.Error(seqId, "distributed workers execution failed: %v", err)
9299
result := metrics.NewCollectResult()
@@ -235,7 +242,16 @@ func (r *defaultRunner) RunWorker(ctx context.Context, params transport.Httpbenc
235242
default: // transport.CmdStart
236243
if params.From == "" {
237244
worker := NewWorker(params.SequenceId)
238-
return worker.Run(ctx, params)
245+
result, err := worker.Run(ctx, params)
246+
if err != nil {
247+
return nil, err
248+
}
249+
// RPS/Average 是 Merge 计算的派生指标;先 Merge 与
250+
// handleStartup 的本地 CLI 路径保持一致,再在 worker 节点
251+
// 本地打印压测 Summary,结果同时经 HTTP 响应返回控制器汇总。
252+
result = metrics.Merge(nil, result)
253+
result.Print()
254+
return result, nil
239255
}
240256
// 浏览器 dashboard:异步执行,立即返回。压测生命周期由
241257
// params.Duration 与 CmdStop 控制,不能使用 HTTP 请求的 ctx

‎index.html‎

Lines changed: 10 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -57,8 +57,9 @@
5757
.main-container { max-width: 1400px; margin: 0 auto; position: relative; z-index: 1; }
5858
.control-panel { position: relative; }
5959
.language-selector-wrap { position: absolute; top: 20px; right: 20px; width: 130px; z-index: 2; }
60-
.language-selector .el-input__inner { color: white; }
61-
.language-selector .el-input__suffix { color: white; }
60+
.language-selector { width: 100%; }
61+
.language-selector .el-input__inner { color: var(--text-main); }
62+
.language-selector .el-input__suffix { color: var(--text-muted); }
6263
.chart-card { min-width: 0; }
6364
.el-form-item { margin-bottom: 12px; }
6465
.status-line { min-height: 24px; margin-top: 8px; color: var(--text-muted); font-size: 0.9rem; }
@@ -163,8 +164,10 @@
163164
background: var(--gradient); color: white; border: none; font-weight: 600; border-radius: 8px 0 0 8px;
164165
}
165166
.language-selector .el-input__inner {
166-
background: rgba(255,255,255,0.2); border: 2px solid rgba(255,255,255,0.3); color: white; font-weight: 600;
167+
background: #f7f9fc; border: 1px solid #dcdfe6; color: var(--text-main); font-weight: 600; border-radius: 8px;
167168
}
169+
.language-selector .el-input__inner:hover { border-color: var(--primary); }
170+
.language-selector .el-select__caret { color: var(--text-muted); }
168171

169172
.charts-row { display: grid; grid-template-columns: 1fr 1fr; gap: 16px; margin-bottom: 16px; animation-fill-mode: both; }
170173
.charts-row .chart-card { margin-bottom: 0; }
@@ -208,7 +211,7 @@
208211
qps: 'QPS',
209212
statusCode: '状态码',
210213
latency: '⏱️ 响应耗时分布',
211-
errorDist: '🚫 错误分布'
214+
errorDist: '⚠️ 错误分布'
212215
},
213216
stats: {
214217
avg: '平均耗时',
@@ -288,10 +291,10 @@
288291
start: 'Start Test',
289292
stop: 'Stop Test',
290293
chart: {
291-
title: '📊 Real-time Performance Metrics',
294+
title: '⚡️ Real-time Performance Metrics',
292295
qps: 'QPS',
293296
statusCode: 'Status Code',
294-
latency: '📊 Latency Distribution',
297+
latency: '⏱️ Latency Distribution',
295298
errorDist: '⚠️ Error Distribution'
296299
},
297300
stats: {
@@ -382,7 +385,7 @@ <h2 id="page-title">🚀 HTTP Benchmark Tool</h2>
382385
<!-- Chart Card -->
383386
<div class="chart-card">
384387
<div class="chart-card-header">
385-
<div class="chart-card-title" id="chart-title">📊 Real-time Performance Metrics</div>
388+
<div class="chart-card-title" id="chart-title">⚡️ Real-time Performance Metrics</div>
386389
</div>
387390
<div id="chart-container" style="height: 400px; width: 100%;"></div>
388391
</div>

‎internal/distributed/distributed_test.go‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ import (
77
"net/http/httptest"
88
"strings"
99
"testing"
10+
"time"
1011

1112
"github.com/linkxzhou/http_bench/internal/transport"
1213
)
@@ -98,3 +99,40 @@ func TestPostAllWorkers_Merges(t *testing.T) {
9899
t.Errorf("merged.BytesReceived = %d, want 30", res.Merged.BytesReceived)
99100
}
100101
}
102+
103+
// TestServeRequest_LongRunSurvivesWriteTimeout reproduces the controller-side
104+
// "all worker(s) failed" error: a benchmark run longer than the server
105+
// WriteTimeout must still deliver its response (the handler clears the write
106+
// deadline), instead of having the connection force-closed mid-run.
107+
func TestServeRequest_LongRunSurvivesWriteTimeout(t *testing.T) {
108+
slow := WorkerServiceFunc(func(ctx context.Context, _ WorkerRequest) (*WorkerResponse, error) {
109+
select {
110+
case <-time.After(300 * time.Millisecond):
111+
return &WorkerResponse{RPS: 42}, nil
112+
case <-ctx.Done():
113+
return nil, ctx.Err()
114+
}
115+
})
116+
mux := http.NewServeMux()
117+
mux.HandleFunc("/api", func(w http.ResponseWriter, r *http.Request) {
118+
ServeRequest(slow, nil, w, r)
119+
})
120+
srv := httptest.NewUnstartedServer(mux)
121+
srv.Config.WriteTimeout = 100 * time.Millisecond
122+
srv.Start()
123+
defer srv.Close()
124+
125+
res, err := PostWorker(srv.URL+"/api", []byte(`{"c":1,"n":1}`), 5*time.Second)
126+
if err != nil {
127+
t.Fatalf("PostWorker failed (write deadline not cleared): %v", err)
128+
}
129+
if res.RPS != 42 {
130+
t.Errorf("RPS = %d, want 42", res.RPS)
131+
}
132+
}
133+
134+
type WorkerServiceFunc func(ctx context.Context, req WorkerRequest) (*WorkerResponse, error)
135+
136+
func (f WorkerServiceFunc) Execute(ctx context.Context, req WorkerRequest) (*WorkerResponse, error) {
137+
return f(ctx, req)
138+
}

‎internal/distributed/handler.go‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -93,6 +93,14 @@ func DefaultValidator(p transport.HttpbenchParameters) error {
9393
// ServeRequest is the standard worker API entry point. The default validation
9494
// can be replaced by passing a custom RequestValidator.
9595
func ServeRequest(service WorkerService, validator RequestValidator, w http.ResponseWriter, r *http.Request) {
96+
// A benchmark run may last far longer than the dashboard server's
97+
// WriteTimeout (30s default): the handler stays busy for the whole run
98+
// and only then writes the response. Without this, the connection is
99+
// force-closed mid-run, the response is truncated, and the controller
100+
// reports the worker as failed even though it completed the benchmark.
101+
// SetWriteDeadline is unsupported on some protocols (e.g. HTTP/2);
102+
// the error is safely ignorable there.
103+
_ = http.NewResponseController(w).SetWriteDeadline(time.Time{})
96104
setCORSHeaders(w, r)
97105
if r.Method == http.MethodOptions {
98106
w.WriteHeader(http.StatusOK)

‎util.go‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import (
44
"errors"
55
"flag"
66
"fmt"
7+
"net/url"
78
"os"
89
"strconv"
910
"strings"
@@ -83,6 +84,32 @@ func parseDuration(timeStr string) (time.Duration, error) {
8384
return d, nil
8485
}
8586

87+
// normalizeWorkerAddrs rewrites bare worker addresses (host:port) into full
88+
// worker API URLs. A missing scheme defaults to "http"; a missing path
89+
// defaults to the worker API mount point ("/api" + workerAPIPath), matching
90+
// the route registered by the dashboard server. Without this, a bare
91+
// "host:port" address fails in http.NewRequest ("first path segment in URL
92+
// cannot contain colon") and the task never reaches the worker.
93+
func normalizeWorkerAddrs(addrs []string, workerAPIPath string) []string {
94+
apiPath := "/api" + workerAPIPath
95+
out := make([]string, 0, len(addrs))
96+
for _, addr := range addrs {
97+
addr = strings.TrimSpace(addr)
98+
if addr == "" {
99+
continue
100+
}
101+
if !strings.Contains(addr, "://") {
102+
addr = "http://" + addr
103+
}
104+
if u, err := url.Parse(addr); err == nil && (u.Path == "" || u.Path == "/") {
105+
u.Path = apiPath
106+
addr = u.String()
107+
}
108+
out = append(out, addr)
109+
}
110+
return out
111+
}
112+
86113
func normalizeCaseInsensitive(s string) string {
87114
if !strings.ContainsAny(s, "smhdMSMHD") {
88115
return s

‎util_test.go‎

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -114,3 +114,63 @@ func TestGenSequenceId_MonotonicWithinNanosecond(t *testing.T) {
114114
t.Fatalf("two consecutive calls produced identical ID %d", a)
115115
}
116116
}
117+
118+
// TestNormalizeWorkerAddrs verifies bare host:port addresses are rewritten
119+
// into full worker API URLs while already-qualified URLs are preserved.
120+
func TestNormalizeWorkerAddrs(t *testing.T) {
121+
tests := []struct {
122+
name string
123+
in []string
124+
api string
125+
want []string
126+
}{
127+
{
128+
name: "bare host:port gets scheme and default api path",
129+
in: []string{"127.0.0.1:12710"},
130+
api: "",
131+
want: []string{"http://127.0.0.1:12710/api"},
132+
},
133+
{
134+
name: "custom worker api path",
135+
in: []string{"127.0.0.1:12710"},
136+
api: "cb9ab101f9f725cb7c3a355bd5631184",
137+
want: []string{"http://127.0.0.1:12710/apicb9ab101f9f725cb7c3a355bd5631184"},
138+
},
139+
{
140+
name: "scheme preserved, path appended",
141+
in: []string{"http://192.168.1.10:12710"},
142+
api: "",
143+
want: []string{"http://192.168.1.10:12710/api"},
144+
},
145+
{
146+
name: "explicit path preserved",
147+
in: []string{"http://192.168.1.10:12710/custom"},
148+
api: "",
149+
want: []string{"http://192.168.1.10:12710/custom"},
150+
},
151+
{
152+
name: "trailing slash treated as empty path",
153+
in: []string{"192.168.1.10:12710/"},
154+
api: "",
155+
want: []string{"http://192.168.1.10:12710/api"},
156+
},
157+
{
158+
name: "empty entries dropped",
159+
in: []string{"", " ", "127.0.0.1:12710"},
160+
api: "",
161+
want: []string{"http://127.0.0.1:12710/api"},
162+
},
163+
}
164+
for _, tt := range tests {
165+
got := normalizeWorkerAddrs(tt.in, tt.api)
166+
if len(got) != len(tt.want) {
167+
t.Errorf("%s: normalizeWorkerAddrs(%v) = %v, want %v", tt.name, tt.in, got, tt.want)
168+
continue
169+
}
170+
for i := range got {
171+
if got[i] != tt.want[i] {
172+
t.Errorf("%s: normalizeWorkerAddrs(%v)[%d] = %q, want %q", tt.name, tt.in, i, got[i], tt.want[i])
173+
}
174+
}
175+
}
176+
}

0 commit comments

Comments
 (0)