diff --git a/.cz.toml b/.cz.toml index ecd78b4..2cc9612 100644 --- a/.cz.toml +++ b/.cz.toml @@ -2,6 +2,6 @@ name = "cz_conventional_commits" tag_format = "$version" version_scheme = "semver" -version = "0.15.4" +version = "0.16.0" update_changelog_on_bump = false major_version_zero = true diff --git a/.gitignore b/.gitignore index 03e07c4..36a8c9a 100644 --- a/.gitignore +++ b/.gitignore @@ -21,6 +21,8 @@ manpages # Output of the go coverage tool, specifically when used with LiteIDE *.out +*.prof +profiles/ # Dependency directories (remove the comment below to include it) # vendor/ diff --git a/CHANGELOG.md b/CHANGELOG.md index 8155d8b..482018c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,44 @@ # https-wrench - changelog +## 0.16.0 (2026-09-17) + +### Dependencies + + Deps: upgrade golang.org/x/sync to v0.23.0 and promote to direct dependency. + +### Feat + + Requests: support concurrent HTTP requests execution via bounded worker pool (`--concurrency` / `-c` flag, default: 10). + + Requests: add `concurrency` property to configuration schema and examples, defaulting to 10 if omitted or <= 0. + + Requests: extend concurrency boundaries down to individual single requests, enabling multiple hosts and URIs within a single `RequestConfig` to execute concurrently. + + Requests: bound active host goroutines in `processHostsConcurrently` using `errgroup.Group.SetLimit`, keeping the HTTP request limiter exclusively for URI requests. + + Requests: acquire limiter slot before launching goroutines in `processURIsConcurrently` to eliminate unbounded goroutine allocation. + + Requests: serialize debug output writing (`PrintRequestDebug`, `PrintResponseDebug`) via mutex to eliminate log tearing and data races across concurrent requests. + + Requests: preserve deterministic slice ordering for hosts and URIs, aggregating response data by request name. + +### Fix + + Devenv: update profiling cleanup traps to remove generated `requests.test` binary from the invocation directory instead of `internal/requests/requests.test`. + + Devenv: add `EXIT` trap to aggregate profiling script to guarantee test binary cleanup on command failures, and remove redundant test binary cleanup from trace script. + + Schema: remove `minimum: 0` constraint on `concurrency` in both `https-wrench.schema.json` and `internal/mcp/assets/schema.json` so negative values are accepted and fall back cleanly to 10. + +### Tests + + Requests: add `TestRequests_ExecuteWithWriter_PeakInFlight` regression test measuring peak in-flight requests with barrier synchronization to eliminate timing jitter under race detection and coverage instrumentation. + + Requests: add `BenchmarkExecuteWithWriter` benchmarking performance across concurrency tiers (`1`, `2`, `5`, `10`, `20`). + + Devenv: add isolated profiling scripts (`profile-requests-concurrency-cpu`, `profile-requests-concurrency-mem`, `profile-requests-concurrency-block`, `profile-requests-concurrency-mutex`, `profile-requests-concurrency-all`, and `trace-requests-concurrency-goroutines`) to avoid measurement cross-talk. + ## 0.15.4 (2026-09-15) ### Feat diff --git a/devenv.nix b/devenv.nix index baff8a3..ff74c95 100644 --- a/devenv.nix +++ b/devenv.nix @@ -1027,6 +1027,90 @@ in go tool trace -http=:3111 /tmp/BenchmarkProbeCiphersConcurrently.trace.out ''; + scripts.bench-requests-concurrency-tiers.exec = '' + set -e + gum format "## BenchmarkExecuteWithWriter concurrency tiers (1, 2, 5, 10, 20)" + + go test ./internal/requests/ -run '^$' \ + -bench BenchmarkExecuteWithWriter -benchmem -count=3 + ''; + + scripts.profile-requests-concurrency-cpu.exec = '' + set -e + trap 'rm -f /tmp/BenchmarkExecuteWithWriter.cpu.out requests.test' EXIT + gum format "## BenchmarkExecuteWithWriter CPU profile (pprof :3112)" + + go test ./internal/requests/ -run '^$' \ + -bench BenchmarkExecuteWithWriter/concurrency-10 -benchtime 2s -benchmem \ + -cpuprofile BenchmarkExecuteWithWriter.cpu.out \ + -outputdir /tmp + go tool pprof -http=:3112 /tmp/BenchmarkExecuteWithWriter.cpu.out + ''; + + scripts.profile-requests-concurrency-mem.exec = '' + set -e + trap 'rm -f /tmp/BenchmarkExecuteWithWriter.mem.out requests.test' EXIT + gum format "## BenchmarkExecuteWithWriter heap profile (pprof :3112, -alloc_objects)" + + go test ./internal/requests/ -run '^$' \ + -bench BenchmarkExecuteWithWriter/concurrency-10 -benchtime 2s -benchmem \ + -memprofile BenchmarkExecuteWithWriter.mem.out \ + -outputdir /tmp + go tool pprof -http=:3112 -alloc_objects /tmp/BenchmarkExecuteWithWriter.mem.out + ''; + + scripts.profile-requests-concurrency-block.exec = '' + set -e + trap 'rm -f /tmp/BenchmarkExecuteWithWriter.block.out requests.test' EXIT + gum format "## BenchmarkExecuteWithWriter block profile (pprof :3112)" + + go test ./internal/requests/ -run '^$' \ + -bench BenchmarkExecuteWithWriter/concurrency-10 -benchtime 2s \ + -blockprofile BenchmarkExecuteWithWriter.block.out \ + -outputdir /tmp + go tool pprof -http=:3112 /tmp/BenchmarkExecuteWithWriter.block.out + ''; + + scripts.profile-requests-concurrency-mutex.exec = '' + set -e + trap 'rm -f /tmp/BenchmarkExecuteWithWriter.mutex.out requests.test' EXIT + gum format "## BenchmarkExecuteWithWriter mutex profile (pprof :3112)" + + go test ./internal/requests/ -run '^$' \ + -bench BenchmarkExecuteWithWriter/concurrency-10 -benchtime 2s \ + -mutexprofile BenchmarkExecuteWithWriter.mutex.out \ + -outputdir /tmp + go tool pprof -http=:3112 /tmp/BenchmarkExecuteWithWriter.mutex.out + ''; + + scripts.trace-requests-concurrency-goroutines.exec = '' + set -e + trap 'rm -f /tmp/BenchmarkExecuteWithWriter.trace.out' EXIT + gum format "## BenchmarkExecuteWithWriter execution trace (:3112)" + + go test ./internal/requests/ -run '^$' \ + -bench BenchmarkExecuteWithWriter/concurrency-10 -benchtime 1s \ + -trace /tmp/BenchmarkExecuteWithWriter.trace.out \ + -outputdir /tmp + go tool trace -http=:3112 /tmp/BenchmarkExecuteWithWriter.trace.out + ''; + + scripts.profile-requests-concurrency-all.exec = '' + set -e + trap 'rm -f requests.test' EXIT + gum format "## Running isolated profiles sequentially to avoid measurement cross-talk" + mkdir -p profiles + echo "1/4 Collecting CPU profile..." + go test ./internal/requests/ -run '^$' -bench BenchmarkExecuteWithWriter/concurrency-10 -benchtime 2s -cpuprofile profiles/requests-cpu.prof + echo "2/4 Collecting Memory profile..." + go test ./internal/requests/ -run '^$' -bench BenchmarkExecuteWithWriter/concurrency-10 -benchtime 2s -memprofile profiles/requests-mem.prof + echo "3/4 Collecting Block profile..." + go test ./internal/requests/ -run '^$' -bench BenchmarkExecuteWithWriter/concurrency-10 -benchtime 2s -blockprofile profiles/requests-block.prof + echo "4/4 Collecting Mutex profile..." + go test ./internal/requests/ -run '^$' -bench BenchmarkExecuteWithWriter/concurrency-10 -benchtime 2s -mutexprofile profiles/requests-mutex.prof + gum format "### All profiles saved in profiles/. Inspect with: go tool pprof -http=:3112 profiles/.prof" + ''; + enterShell = '' echo "https-wrench devenv ready" go version diff --git a/go.mod b/go.mod index b4eb183..c65fc08 100644 --- a/go.mod +++ b/go.mod @@ -21,6 +21,7 @@ require ( github.com/spf13/viper v1.21.0 github.com/stretchr/testify v1.11.1 github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 + golang.org/x/sync v0.23.0 golang.org/x/term v0.45.0 ) @@ -60,7 +61,6 @@ require ( go.yaml.in/yaml/v3 v3.0.5 // indirect golang.org/x/crypto v0.55.0 // indirect golang.org/x/oauth2 v0.36.0 // indirect - golang.org/x/sync v0.22.0 // indirect golang.org/x/sys v0.47.0 // indirect golang.org/x/text v0.41.0 // indirect golang.org/x/time v0.15.0 // indirect diff --git a/go.sum b/go.sum index e01694c..6fcefe6 100644 --- a/go.sum +++ b/go.sum @@ -121,8 +121,8 @@ golang.org/x/exp v0.0.0-20231006140011-7918f672742d h1:jtJma62tbqLibJ5sFQz8bKtEM golang.org/x/exp v0.0.0-20231006140011-7918f672742d/go.mod h1:ldy0pHrwJyGW56pPQzzkH36rKxoZW1tw7ZJpeKx+hdo= golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs= golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q= -golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= -golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sync v0.23.0 h1:KameEIfc1IkluZyXWLn39Wd4tURc6GbCiISGiZm2bQk= +golang.org/x/sync v0.23.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0= golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/term v0.45.0 h1:NwWyBmoJCbfTHpxrWoZ9C6/VxOf7ic219I8xZZFdrf0= diff --git a/https-wrench.schema.json b/https-wrench.schema.json index 5373300..3f38b54 100644 --- a/https-wrench.schema.json +++ b/https-wrench.schema.json @@ -19,6 +19,11 @@ "type": "string", "description": "PEM-encoded CA certificate bundle used to verify server certificates. When omitted, the system trust store is used." }, + "concurrency": { + "type": "integer", + "default": 10, + "description": "Maximum number of concurrent HTTP requests (1 for sequential; defaults to 10 if omitted or <= 0)." + }, "baseRequest": { "$ref": "#/definitions/RequestDefaults", "description": "YAML-only shared request template. Define an anchor (e.g. baseRequest: &base) and merge into requests with <<: *base. Ignored by https-wrench at runtime." diff --git a/internal/cmd/embedded/config-example.yaml b/internal/cmd/embedded/config-example.yaml index 73c6a3f..d3aa961 100644 --- a/internal/cmd/embedded/config-example.yaml +++ b/internal/cmd/embedded/config-example.yaml @@ -10,6 +10,9 @@ debug: false ## verbose - Enables verbose output, showing more details during execution. Required option. verbose: true +## concurrency - Maximum number of concurrent HTTP requests (1 for sequential, defaults to 10 if omitted or <= 0). +concurrency: 10 + ## caBundle - A PEM-encoded CA certificate bundle as a multiline string to be used for verifying server certificates. ## When testing inside the devenv environment, the 'devenv up' command will create new self-signed certificates and ## start a local, HTTPS-enabled Nginx server. diff --git a/internal/cmd/requests.go b/internal/cmd/requests.go index e6a2173..1d72f34 100644 --- a/internal/cmd/requests.go +++ b/internal/cmd/requests.go @@ -20,9 +20,10 @@ import ( var ( //go:embed embedded/config-example.yaml - sampleYamlConfig string - showSampleConfig bool - requestsFmt string + sampleYamlConfig string + showSampleConfig bool + requestsFmt string + requestsConcurrency int ) var requestsCmd = &cobra.Command{ @@ -95,8 +96,14 @@ Examples: return } + concurrency := requestsConcurrency + if !cmd.Flags().Changed("concurrency") && cfg.Concurrency > 0 { + concurrency = cfg.Concurrency + } + requestsCfg.SetVerbose(cfg.Verbose). SetDebug(cfg.Debug). + SetConcurrency(concurrency). SetRequests(cfg.Requests) if err := requestsCfg.SetCaPoolFromYAML(cfg.CaBundle); err != nil { @@ -166,5 +173,12 @@ func init() { "text", "Output format: text (human-readable) or json (machine-readable, no ANSI)", ) + requestsCmd.Flags().IntVarP( + &requestsConcurrency, + "concurrency", + "c", + requests.DefaultRequestsConcurrency, + "Maximum number of concurrent HTTP requests (1 for sequential)", + ) rootCmd.AddCommand(requestsCmd) } diff --git a/internal/cmd/requests_test.go b/internal/cmd/requests_test.go index 32d1831..e01d41c 100644 --- a/internal/cmd/requests_test.go +++ b/internal/cmd/requests_test.go @@ -33,6 +33,7 @@ func TestRequestsCmd(t *testing.T) { "Global Flags:", "--config", "--ca-bundle", + "--concurrency", "--format", "--show-sample-config", "--version", @@ -45,6 +46,7 @@ func TestRequestsCmd(t *testing.T) { expectError: false, expected: []string{ "https-wrench.schema.json", + "concurrency:", "requests:", "transportOverrideUrl:", "requestHeaders:", diff --git a/internal/cmd/root_test.go b/internal/cmd/root_test.go index 167736b..ba55052 100644 --- a/internal/cmd/root_test.go +++ b/internal/cmd/root_test.go @@ -84,6 +84,7 @@ func TestRootCmd_LoadConfig(t *testing.T) { require.IsType(t, expectedRequestsConfigs, config.Requests) // testing against the current values of the embedded config + require.Equal(t, 10, config.Concurrency) require.Equal(t, "SampleRequestAgainstLocalWebserver", config.Requests[0].Name) require.Equal(t, "https://127.0.0.1:9443", config.Requests[0].TransportOverrideURL) }) diff --git a/internal/mcp/assets/sample-config.yaml b/internal/mcp/assets/sample-config.yaml index f3a82ee..16f1c54 100644 --- a/internal/mcp/assets/sample-config.yaml +++ b/internal/mcp/assets/sample-config.yaml @@ -10,6 +10,9 @@ debug: false ## verbose - Enables verbose output, showing more details during execution. Required option. verbose: true +## concurrency - Maximum number of concurrent HTTP requests (1 for sequential, defaults to 10 if omitted or <= 0). +concurrency: 10 + ## caBundle - A PEM-encoded CA certificate bundle as a multiline string to be used for verifying server certificates. ## When testing inside the devenv environment, the 'devenv up' command will create new self-signed certificates and ## start a local, HTTPS-enabled Nginx server. diff --git a/internal/mcp/assets/schema.json b/internal/mcp/assets/schema.json index 4b4f18d..4be7280 100644 --- a/internal/mcp/assets/schema.json +++ b/internal/mcp/assets/schema.json @@ -19,6 +19,11 @@ "type": "string", "description": "PEM-encoded CA certificate bundle used to verify server certificates. When omitted, the system trust store is used." }, + "concurrency": { + "type": "integer", + "default": 10, + "description": "Maximum number of concurrent HTTP requests (1 for sequential; defaults to 10 if omitted or <= 0)." + }, "baseRequest": { "$ref": "#/definitions/RequestDefaults", "description": "YAML-only shared request template. Define an anchor (e.g. baseRequest: &base) and merge into requests with <<: *base. Ignored by https-wrench at runtime." diff --git a/internal/requests/requests.go b/internal/requests/requests.go index dbd9f57..b98a63c 100644 --- a/internal/requests/requests.go +++ b/internal/requests/requests.go @@ -15,11 +15,13 @@ import ( "os" "slices" "strings" + "sync" "time" "github.com/pires/go-proxyproto" "github.com/xenos76/https-wrench/internal/certinfo" "github.com/xenos76/https-wrench/internal/view" + "golang.org/x/sync/errgroup" ) const ( @@ -40,6 +42,9 @@ const ( proxyProtoDefaultSrcPort = 54321 emptyString = "" + + // DefaultRequestsConcurrency is the default number of concurrent requests. + DefaultRequestsConcurrency = 10 ) // defaultCurvePreferences lists Go 1.27 TLS hybrids plus classical fallbacks. @@ -180,6 +185,8 @@ type RequestsMetaConfig struct { RequestVerbose bool // CACertsPool is the certificate pool used for validating server certificates. CACertsPool *x509.CertPool + // Concurrency limits the number of concurrent requests executing simultaneously. + Concurrency int `mapstructure:"concurrency"` // Requests is the list of request configurations to execute. Requests []RequestConfig `mapstructure:"requests"` } @@ -210,6 +217,12 @@ func (r *RequestsMetaConfig) SetDebug(b bool) *RequestsMetaConfig { return r } +// SetConcurrency sets the maximum concurrency level for executing requests. +func (r *RequestsMetaConfig) SetConcurrency(n int) *RequestsMetaConfig { + r.Concurrency = n + return r +} + // SetCaPoolFromYAML loads a CA certificate pool from a PEM-encoded string. func (r *RequestsMetaConfig) SetCaPoolFromYAML(s string) error { if s != "" { @@ -252,36 +265,177 @@ func (r *RequestsMetaConfig) Execute(ctx context.Context) (*Result, map[string][ return r.ExecuteWithWriter(ctx, io.Discard) } -// ExecuteWithWriter runs all configured HTTP requests, writing debug output to w. -func (r *RequestsMetaConfig) ExecuteWithWriter(ctx context.Context, w io.Writer) (*Result, map[string][]ResponseData, error) { - if w == nil { - w = io.Discard - } - - seenNames := make(map[string]struct{}, len(r.Requests)) - for _, reqCfg := range r.Requests { +func validateUniqueRequestNames(requests []RequestConfig) error { + seenNames := make(map[string]struct{}, len(requests)) + for _, reqCfg := range requests { if _, exists := seenNames[reqCfg.Name]; exists { - return nil, nil, &DuplicateRequestNameError{Name: reqCfg.Name} + return &DuplicateRequestNameError{Name: reqCfg.Name} } seenNames[reqCfg.Name] = struct{}{} } - responseDataMap := make(map[string][]ResponseData) + return nil +} + +type requestLimiter struct { + sem chan struct{} + writerMu sync.Mutex +} + +func newRequestLimiter(limit int) *requestLimiter { + if limit <= 0 { + limit = DefaultRequestsConcurrency + } + + return &requestLimiter{ + sem: make(chan struct{}, limit), + } +} + +func (l *requestLimiter) acquire(ctx context.Context) error { + if l == nil { + return nil + } + + select { + case l.sem <- struct{}{}: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +func (l *requestLimiter) release() { + if l == nil { + return + } + + <-l.sem +} + +func (l *requestLimiter) writeOutput(w io.Writer, data []byte) { + if l == nil { + _, _ = w.Write(data) + return + } + + l.writerMu.Lock() + defer l.writerMu.Unlock() + + _, _ = w.Write(data) +} + +func (l *requestLimiter) limit() int { + if l == nil { + return 0 + } + + return cap(l.sem) +} + +type indexedResult struct { + name string + data []ResponseData +} + +func (r *RequestsMetaConfig) executeRequestsSequentially( + ctx context.Context, + w io.Writer, +) (map[string][]ResponseData, error) { + responseDataMap := make(map[string][]ResponseData, len(r.Requests)) for _, reqCfg := range r.Requests { if err := ctx.Err(); err != nil { - return nil, nil, err + return nil, err } responseDataList, err := processHTTPRequestsByHost(ctx, w, reqCfg, r.CACertsPool, r.RequestVerbose) if err != nil { - return nil, nil, err + return nil, err } responseDataMap[reqCfg.Name] = responseDataList } + return responseDataMap, nil +} + +func (r *RequestsMetaConfig) executeRequestsConcurrently( + ctx context.Context, + w io.Writer, + concurrency int, +) (map[string][]ResponseData, error) { + limiter := newRequestLimiter(concurrency) + results := make([]indexedResult, len(r.Requests)) + + g, gCtx := errgroup.WithContext(ctx) + + for i, reqCfg := range r.Requests { + g.Go(func() error { + data, err := processHTTPRequestsByHost(gCtx, w, reqCfg, r.CACertsPool, r.RequestVerbose, limiter) + if err != nil { + return err + } + + results[i] = indexedResult{ + name: reqCfg.Name, + data: data, + } + + return nil + }) + } + + if err := g.Wait(); err != nil { + return nil, err + } + + if err := ctx.Err(); err != nil { + return nil, err + } + + responseDataMap := make(map[string][]ResponseData, len(r.Requests)) + + for _, res := range results { + if res.name != "" { + responseDataMap[res.name] = res.data + } + } + + return responseDataMap, nil +} + +// ExecuteWithWriter runs all configured HTTP requests, writing debug output to w. +func (r *RequestsMetaConfig) ExecuteWithWriter(ctx context.Context, w io.Writer) (*Result, map[string][]ResponseData, error) { + if w == nil { + w = io.Discard + } + + if err := validateUniqueRequestNames(r.Requests); err != nil { + return nil, nil, err + } + + concurrency := r.Concurrency + if concurrency <= 0 { + concurrency = DefaultRequestsConcurrency + } + + var ( + responseDataMap map[string][]ResponseData + err error + ) + + if concurrency == 1 || len(r.Requests) == 0 { + responseDataMap, err = r.executeRequestsSequentially(ctx, w) + } else { + responseDataMap, err = r.executeRequestsConcurrently(ctx, w, concurrency) + } + + if err != nil { + return nil, nil, err + } + result, err := BuildResult(responseDataMap, r) if err != nil { return nil, nil, err @@ -726,6 +880,28 @@ func processHTTPRequestsByHost( r RequestConfig, caPool *x509.CertPool, isVerbose bool, + limiter ...*requestLimiter, +) ([]ResponseData, error) { + var lim *requestLimiter + if len(limiter) > 0 { + lim = limiter[0] + } + + if lim == nil || len(r.Hosts) <= 1 { + return processHostsSequentially(ctx, w, r, caPool, isVerbose, lim) + } + + return processHostsConcurrently(ctx, w, r, caPool, isVerbose, lim) +} + +//nolint:revive +func processHostsSequentially( + ctx context.Context, + w io.Writer, + r RequestConfig, + caPool *x509.CertPool, + isVerbose bool, + limiter *requestLimiter, ) ([]ResponseData, error) { var responseDataList []ResponseData @@ -734,7 +910,7 @@ func processHTTPRequestsByHost( return nil, err } - hostResults, err := processRequestsForHost(ctx, w, r, host, caPool, isVerbose) + hostResults, err := processRequestsForHost(ctx, w, r, host, caPool, isVerbose, limiter) if err != nil { return nil, err } @@ -745,7 +921,54 @@ func processHTTPRequestsByHost( return responseDataList, nil } +//nolint:revive +func processHostsConcurrently( + ctx context.Context, + w io.Writer, + r RequestConfig, + caPool *x509.CertPool, + isVerbose bool, + limiter *requestLimiter, +) ([]ResponseData, error) { + hostResultsList := make([][]ResponseData, len(r.Hosts)) + + g, gCtx := errgroup.WithContext(ctx) + if limiter != nil && limiter.limit() > 0 { + g.SetLimit(limiter.limit()) + } + + for i, host := range r.Hosts { + if err := gCtx.Err(); err != nil { + break + } + + g.Go(func() error { + results, err := processRequestsForHost(gCtx, w, r, host, caPool, isVerbose, limiter) + if err != nil { + return err + } + + hostResultsList[i] = results + + return nil + }) + } + + if err := g.Wait(); err != nil { + return nil, err + } + + var responseDataList []ResponseData + for _, res := range hostResultsList { + responseDataList = append(responseDataList, res...) + } + + return responseDataList, nil +} + // processRequestsForHost initializes the HTTP client and executes all configured URIs for a single host. +// +//nolint:revive func processRequestsForHost( ctx context.Context, w io.Writer, @@ -753,9 +976,8 @@ func processRequestsForHost( host Host, caPool *x509.CertPool, isVerbose bool, + limiter *requestLimiter, ) ([]ResponseData, error) { - var responseDataList []ResponseData - reqClient, err := NewHTTPClientFromRequestConfig(r, host.Name, caPool) if err != nil { return nil, err @@ -766,20 +988,117 @@ func processRequestsForHost( return nil, err } + if limiter == nil || len(urlList) <= 1 { + return processURIsSequentially(ctx, w, r, reqClient, urlList, isVerbose, limiter) + } + + return processURIsConcurrently(ctx, w, r, reqClient, urlList, isVerbose, limiter) +} + +//nolint:revive +func processURIsSequentially( + ctx context.Context, + w io.Writer, + r RequestConfig, + reqClient *RequestHTTPClient, + urlList []string, + isVerbose bool, + limiter *requestLimiter, +) ([]ResponseData, error) { requestBodyBytes := []byte(r.RequestBody) + responseDataList := make([]ResponseData, len(urlList)) - for _, reqURL := range urlList { + for i, reqURL := range urlList { if err := ctx.Err(); err != nil { return nil, err } - responseData := executeSingleRequest(ctx, w, r, reqClient, reqURL, requestBodyBytes, isVerbose) - responseDataList = append(responseDataList, responseData) + if err := limiter.acquire(ctx); err != nil { + return nil, err + } + + responseDataList[i] = executeSingleRequestWithLimiterOutput( + ctx, w, r, reqClient, reqURL, requestBodyBytes, isVerbose, limiter, + ) + limiter.release() + } + + return responseDataList, nil +} + +//nolint:revive +func processURIsConcurrently( + ctx context.Context, + w io.Writer, + r RequestConfig, + reqClient *RequestHTTPClient, + urlList []string, + isVerbose bool, + limiter *requestLimiter, +) ([]ResponseData, error) { + requestBodyBytes := []byte(r.RequestBody) + responseDataList := make([]ResponseData, len(urlList)) + + g, gCtx := errgroup.WithContext(ctx) + + for i, reqURL := range urlList { + if err := limiter.acquire(gCtx); err != nil { + _ = g.Wait() + + return nil, err + } + + g.Go(func() error { + defer limiter.release() + + responseDataList[i] = executeSingleRequestWithLimiterOutput( + gCtx, w, r, reqClient, reqURL, requestBodyBytes, isVerbose, limiter, + ) + + return nil + }) + } + + if err := g.Wait(); err != nil { + return nil, err + } + + if err := ctx.Err(); err != nil { + return nil, err } return responseDataList, nil } +//nolint:revive +func executeSingleRequestWithLimiterOutput( + ctx context.Context, + w io.Writer, + r RequestConfig, + reqClient *RequestHTTPClient, + reqURL string, + requestBodyBytes []byte, + isVerbose bool, + limiter *requestLimiter, +) ResponseData { + if w == nil { + w = io.Discard + } + + if limiter != nil && w != io.Discard { + var buf bytes.Buffer + + resp := executeSingleRequest(ctx, &buf, r, reqClient, reqURL, requestBodyBytes, isVerbose) + if buf.Len() > 0 { + limiter.writeOutput(w, buf.Bytes()) + } + + return resp + } + + return executeSingleRequest(ctx, w, r, reqClient, reqURL, requestBodyBytes, isVerbose) +} + // executeSingleRequest performs a single HTTP request and returns the collected response data. func executeSingleRequest( ctx context.Context, diff --git a/internal/requests/requests_test.go b/internal/requests/requests_test.go index 37434cf..c0f4a4f 100644 --- a/internal/requests/requests_test.go +++ b/internal/requests/requests_test.go @@ -11,6 +11,8 @@ import ( "net/http" "net/http/httptest" "net/url" + "sync" + "sync/atomic" "testing" "time" @@ -1742,3 +1744,412 @@ func verifyProcessHTTPRequestsResults( } } } + +func TestRequests_ExecuteWithWriter_Concurrent(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("ok: " + r.URL.Path)) + })) + defer ts.Close() + + u, err := url.Parse(ts.URL) + require.NoError(t, err) + + const numRequests = 8 + + reqs := make([]RequestConfig, numRequests) + for i := 0; i < numRequests; i++ { + reqs[i] = RequestConfig{ + Name: fmt.Sprintf("req-%d", i), + Hosts: []Host{ + { + Name: u.Host, + URIList: []URI{URI(fmt.Sprintf("/path-%d", i))}, + }, + }, + TransportOverrideURL: ts.URL, + RequestDebug: true, + ResponseDebug: true, + } + } + + rmc := &RequestsMetaConfig{ + Requests: reqs, + } + + // Sequential run (concurrency = 1) + rmc.SetConcurrency(1) + + seqBuf := &bytes.Buffer{} + seqResult, seqMap, err := rmc.ExecuteWithWriter(context.Background(), seqBuf) + require.NoError(t, err) + require.Len(t, seqMap, numRequests) + require.Len(t, seqResult.Requests, numRequests) + + // Concurrent run (concurrency = 4) + rmc.SetConcurrency(4) + + concBuf := &bytes.Buffer{} + concResult, concMap, err := rmc.ExecuteWithWriter(context.Background(), concBuf) + require.NoError(t, err) + require.Len(t, concMap, numRequests) + require.Len(t, concResult.Requests, numRequests) + + // Verify both produce identical result structures and response mappings + for i := 0; i < numRequests; i++ { + name := fmt.Sprintf("req-%d", i) + require.Contains(t, concMap, name) + require.Len(t, concMap[name], 1) + assert.Equal(t, seqMap[name][0].URL, concMap[name][0].URL) + assert.Equal(t, seqResult.Requests[i].Name, concResult.Requests[i].Name) + } + + // Verify debug output was written to buffer + assert.NotEmpty(t, concBuf.String()) +} + +func TestRequests_ExecuteWithWriter_ContextCancel(t *testing.T) { + ts := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + time.Sleep(100 * time.Millisecond) + w.WriteHeader(http.StatusOK) + })) + defer ts.Close() + + u, err := url.Parse(ts.URL) + require.NoError(t, err) + + reqs := make([]RequestConfig, 4) + for i := 0; i < 4; i++ { + reqs[i] = RequestConfig{ + Name: fmt.Sprintf("cancel-req-%d", i), + Hosts: []Host{ + { + Name: u.Host, + URIList: []URI{"/sleep"}, + }, + }, + TransportOverrideURL: ts.URL, + Insecure: true, + } + } + + rmc := &RequestsMetaConfig{ + Concurrency: 4, + Requests: reqs, + } + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + + _, _, err = rmc.ExecuteWithWriter(ctx, io.Discard) + require.Error(t, err) + assert.ErrorIs(t, err, context.DeadlineExceeded) +} + +func TestRequests_ExecuteWithWriter_SetupError(t *testing.T) { + reqs := []RequestConfig{ + { + Name: "valid-req", + Hosts: []Host{ + { + Name: "localhost", + URIList: []URI{"/ok"}, + }, + }, + }, + { + Name: "invalid-req", + Hosts: []Host{ + { + Name: "localhost", + URIList: []URI{"invalid-uri-no-slash"}, + }, + }, + }, + } + + rmc := &RequestsMetaConfig{ + Concurrency: 2, + Requests: reqs, + } + + _, _, err := rmc.ExecuteWithWriter(context.Background(), io.Discard) + require.Error(t, err) + assert.ErrorIs(t, err, ErrInvalidURI) +} + +func newBarrierTLSServer(targetConcurrency int64) (*httptest.Server, *atomic.Int64) { + var ( + currentInFlight atomic.Int64 + peakInFlight atomic.Int64 + releaseOnce sync.Once + ) + + releaseCh := make(chan struct{}) + + ts := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + cur := currentInFlight.Add(1) + defer currentInFlight.Add(-1) + + for { + peak := peakInFlight.Load() + if cur <= peak { + break + } + + if peakInFlight.CompareAndSwap(peak, cur) { + break + } + } + + if cur == targetConcurrency { + releaseOnce.Do(func() { + close(releaseCh) + }) + } + + select { + case <-releaseCh: + case <-time.After(5 * time.Second): + } + + w.WriteHeader(http.StatusOK) + })) + + return ts, &peakInFlight +} + +func testPeakInFlightSingleRequestConfig(t *testing.T) { + t.Parallel() + + const targetConcurrency = 3 + + ts, peakInFlight := newBarrierTLSServer(targetConcurrency) + defer ts.Close() + + u, err := url.Parse(ts.URL) + require.NoError(t, err) + + const numURIs = 10 + + uris := make([]URI, numURIs) + for i := 0; i < numURIs; i++ { + uris[i] = URI(fmt.Sprintf("/path-%d", i)) + } + + rmc := &RequestsMetaConfig{ + Concurrency: targetConcurrency, + Requests: []RequestConfig{ + { + Name: "multi-uri-request", + Hosts: []Host{ + { + Name: u.Host, + URIList: uris, + }, + }, + TransportOverrideURL: ts.URL, + Insecure: true, + }, + }, + } + + result, respMap, err := rmc.ExecuteWithWriter(context.Background(), io.Discard) + require.NoError(t, err) + require.NotNil(t, result) + + assert.Equal( + t, int64(targetConcurrency), peakInFlight.Load(), + "peak in-flight requests should reach concurrency limit", + ) + require.Contains(t, respMap, "multi-uri-request") + require.Len(t, respMap["multi-uri-request"], numURIs) + + for i := 0; i < numURIs; i++ { + expectedURL := fmt.Sprintf("https://%s/path-%d", u.Host, i) + assert.Equal(t, expectedURL, respMap["multi-uri-request"][i].URL) + } +} + +func testPeakInFlightMultipleRequestConfigs(t *testing.T) { + t.Parallel() + + const targetConcurrency = 4 + + ts, peakInFlight := newBarrierTLSServer(targetConcurrency) + defer ts.Close() + + u, err := url.Parse(ts.URL) + require.NoError(t, err) + + const ( + numRequests = 4 + urisPerReq = 3 + ) + + reqs := make([]RequestConfig, numRequests) + for i := 0; i < numRequests; i++ { + uris := make([]URI, urisPerReq) + for j := 0; j < urisPerReq; j++ { + uris[j] = URI(fmt.Sprintf("/req-%d-path-%d", i, j)) + } + + reqs[i] = RequestConfig{ + Name: fmt.Sprintf("req-%d", i), + Hosts: []Host{ + { + Name: u.Host, + URIList: uris, + }, + }, + TransportOverrideURL: ts.URL, + Insecure: true, + } + } + + rmc := &RequestsMetaConfig{ + Concurrency: targetConcurrency, + Requests: reqs, + } + + result, respMap, err := rmc.ExecuteWithWriter(context.Background(), io.Discard) + require.NoError(t, err) + require.NotNil(t, result) + + assert.Equal( + t, int64(targetConcurrency), peakInFlight.Load(), + "peak in-flight requests should reach concurrency limit", + ) + assert.Len(t, respMap, numRequests) + + for i := 0; i < numRequests; i++ { + reqName := fmt.Sprintf("req-%d", i) + require.Contains(t, respMap, reqName) + require.Len(t, respMap[reqName], urisPerReq) + + for j := 0; j < urisPerReq; j++ { + expectedURL := fmt.Sprintf("https://%s/req-%d-path-%d", u.Host, i, j) + assert.Equal(t, expectedURL, respMap[reqName][j].URL) + } + } +} + +func testPeakInFlightSequential(t *testing.T) { + t.Parallel() + + var ( + currentInFlight atomic.Int64 + peakInFlight atomic.Int64 + ) + + ts := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + cur := currentInFlight.Add(1) + defer currentInFlight.Add(-1) + + for { + peak := peakInFlight.Load() + if cur <= peak { + break + } + + if peakInFlight.CompareAndSwap(peak, cur) { + break + } + } + + time.Sleep(10 * time.Millisecond) + w.WriteHeader(http.StatusOK) + })) + defer ts.Close() + + u, err := url.Parse(ts.URL) + require.NoError(t, err) + + uris := []URI{"/p1", "/p2", "/p3", "/p4"} + rmc := &RequestsMetaConfig{ + Concurrency: 1, + Requests: []RequestConfig{ + { + Name: "seq-req", + Hosts: []Host{ + { + Name: u.Host, + URIList: uris, + }, + }, + TransportOverrideURL: ts.URL, + Insecure: true, + }, + }, + } + + result, respMap, err := rmc.ExecuteWithWriter(context.Background(), io.Discard) + require.NoError(t, err) + require.NotNil(t, result) + + assert.Equal( + t, int64(1), peakInFlight.Load(), + "concurrency 1 must never exceed 1 in-flight", + ) + require.Len(t, respMap["seq-req"], 4) +} + +func TestRequests_ExecuteWithWriter_PeakInFlight(t *testing.T) { + t.Parallel() + + t.Run("single request config with multiple URIs", testPeakInFlightSingleRequestConfig) + t.Run("multiple request configs with multiple URIs", testPeakInFlightMultipleRequestConfigs) + t.Run("concurrency 1 executes strictly sequentially", testPeakInFlightSequential) +} + +func BenchmarkExecuteWithWriter(b *testing.B) { + ts := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + time.Sleep(2 * time.Millisecond) // simulate small network I/O latency + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("benchmark-ok")) + })) + defer ts.Close() + + u, err := url.Parse(ts.URL) + if err != nil { + b.Fatal(err) + } + + const numRequests = 20 + + reqs := make([]RequestConfig, numRequests) + for i := 0; i < numRequests; i++ { + reqs[i] = RequestConfig{ + Name: fmt.Sprintf("bench-req-%d", i), + Hosts: []Host{ + { + Name: u.Host, + URIList: []URI{URI(fmt.Sprintf("/path-%d", i))}, + }, + }, + TransportOverrideURL: ts.URL, + Insecure: true, + } + } + + concurrencyTiers := []int{1, 2, 5, 10, 20} + + for _, tier := range concurrencyTiers { + b.Run(fmt.Sprintf("concurrency-%d", tier), func(b *testing.B) { + rmc := &RequestsMetaConfig{ + Concurrency: tier, + Requests: reqs, + } + + b.ResetTimer() + + for i := 0; i < b.N; i++ { + _, _, err := rmc.ExecuteWithWriter(context.Background(), io.Discard) + if err != nil { + b.Fatalf("benchmark failed: %v", err) + } + } + }) + } +} diff --git a/vendor/modules.txt b/vendor/modules.txt index 4d00d2a..74b2c7d 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -222,8 +222,8 @@ golang.org/x/crypto/scrypt ## explicit; go 1.25.0 golang.org/x/oauth2 golang.org/x/oauth2/internal -# golang.org/x/sync v0.22.0 -## explicit; go 1.25.0 +# golang.org/x/sync v0.23.0 +## explicit; go 1.26.0 golang.org/x/sync/errgroup # golang.org/x/sys v0.47.0 ## explicit; go 1.25.0