From 9230d0cd3d3df5828e510c915d53714c572887a2 Mon Sep 17 00:00:00 2001 From: Jan Novak Date: Sat, 29 Aug 2026 00:08:27 +0200 Subject: [PATCH] body-drain: add -c concurrency and -idle-per-host to the client Bounded concurrency via semaphore+WaitGroup, atomic counters, stop-on-first- error without cancelling in-flight requests (cancel would hide the leaks). Exposes MaxIdleConnsPerHost so pool thrash under concurrency is observable. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01GbnJj9ZqEx9nUvzkjdwfMq --- 99-experiments/body-drain/README.md | 27 ++++++++ 99-experiments/body-drain/main.go | 96 ++++++++++++++++++++++------- 2 files changed, 100 insertions(+), 23 deletions(-) diff --git a/99-experiments/body-drain/README.md b/99-experiments/body-drain/README.md index 68e2e20..958e743 100644 --- a/99-experiments/body-drain/README.md +++ b/99-experiments/body-drain/README.md @@ -8,6 +8,33 @@ go run ./99-experiments/body-drain -mode all go run ./99-experiments/body-drain -mode leak -n 200 ``` +## Concurrency: `-c` and `-idle-per-host` + +`-c N` keeps up to N requests in flight (semaphore + WaitGroup, the same +shape as exercise 01). `-idle-per-host` sets `Transport.MaxIdleConnsPerHost` +(default 10; Go's own default is only **2**). + +``` +go run ./99-experiments/body-drain -mode drain -n 200 -c 8 +# httptrace GotConn : reused=192 fresh=8 ← one conn per worker, then reused +# server new conns : 8 + +go run ./99-experiments/body-drain -mode drain -n 200 -c 8 -idle-per-host 2 +# server new conns : 13+ ← pool too small: conns returned +# while 2 are already idle get CLOSED +``` + +That second run is the trap in exercise 01: `limit` goroutines against one +host with the default transport means only 2 conns are ever kept, the rest are +dialed and thrown away on every round — even though you drained the body +correctly. On loopback the effect is small; against a TLS host each throwaway +is a full handshake. Raise `MaxIdleConnsPerHost` to your concurrency. + +With `-c` the leak modes also exhaust fds N times faster, and failures no +longer stop in-flight requests: cancelling a request context closes its body +and would clean up the very leaks we want to see, so the client only stops +*launching* new requests after the first error. + ## Two processes: watch each side run out of fds on its own In one process the client and server share one fd limit and each leaked diff --git a/99-experiments/body-drain/main.go b/99-experiments/body-drain/main.go index d4fd088..cb30c0d 100644 --- a/99-experiments/body-drain/main.go +++ b/99-experiments/body-drain/main.go @@ -24,6 +24,8 @@ // Run in one process: // // go run ./99-experiments/body-drain -mode all +// go run ./99-experiments/body-drain -mode drain -n 200 -c 8 # 8 in flight, pool of 10 → still 1 host conn per worker +// go run ./99-experiments/body-drain -mode drain -n 200 -c 8 -idle-per-host 2 # pool too small → fresh dials even when draining // // Run as two processes (two terminals): // @@ -56,6 +58,7 @@ import ( "os" "os/signal" "runtime" + "sync" "sync/atomic" "time" ) @@ -153,53 +156,73 @@ func serveForever(addr string) { // ---- client ---- -// runClient performs n GETs in the given mode and reports what it observed. -func runClient(m mode, n int, url string, headerTimeout time.Duration) { +// clientOpts are the knobs for one client run. +type clientOpts struct { + n int // total requests + concurrency int // max in-flight requests (1 = sequential) + idlePerHost int // Transport.MaxIdleConnsPerHost + headerTimeout time.Duration // Transport.ResponseHeaderTimeout +} + +// runClient performs opts.n GETs in the given mode and reports what it observed. +func runClient(m mode, url string, opts clientOpts) { // Fresh transport per run so runs don't share a pool. Cloning keeps sane defaults. tr := http.DefaultTransport.(*http.Transport).Clone() - tr.MaxIdleConnsPerHost = 10 + // With concurrency > MaxIdleConnsPerHost, even `drain` mode dials fresh + // connections: conns that come back while the idle pool is full are + // closed instead of pooled. Watch `fresh` climb as -c grows past -idle-per-host. + tr.MaxIdleConnsPerHost = opts.idlePerHost // Without this, a request whose connection the server can't accept (server // out of fds) blocks forever waiting for headers. - tr.ResponseHeaderTimeout = headerTimeout + tr.ResponseHeaderTimeout = opts.headerTimeout // NOTE: deliberately no Client.Timeout. With a timeout, a leaked body gets // force-closed when the timer fires, which hides the leak (try adding one // and watch the goroutine count drop after the timeout). client := &http.Client{Transport: tr} - var reused, fresh int + // Counters are touched from many goroutines when -c > 1 → atomics. + var reused, fresh, done, failed atomic.Int64 trace := &httptrace.ClientTrace{ GotConn: func(info httptrace.GotConnInfo) { if info.Reused { - reused++ + reused.Add(1) } else { - fresh++ + fresh.Add(1) } }, } - goroutinesBefore := runtime.NumGoroutine() - start := time.Now() + // stop is set on the first failure so we launch no further requests. + // Deliberately NOT a cancellable request context: cancelling a request's + // ctx closes its body/conn, which would clean up the leaks we're here + // to observe. In-flight requests are left alone. + var stop atomic.Bool + ctx := httptrace.WithClientTrace(context.Background(), trace) - done := 0 - for i := 0; i < n; i++ { - req, err := http.NewRequestWithContext(httptrace.WithClientTrace(context.Background(), trace), http.MethodGet, url, nil) + var firstErr sync.Once + doOne := func(i int) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) if err != nil { panic(err) } resp, err := client.Do(req) if err != nil { + failed.Add(1) // This is where `leak` eventually dies. Which error you get tells // you which side ran out of fds: // dial tcp ...: socket: too many open files → THIS process (client) // timeout awaiting response headers → the server can't accept (it's out of fds) - fmt.Printf(" request %d failed: %v\n", i, err) - var ne net.Error - if errors.As(err, &ne) && ne.Timeout() { - fmt.Println(" (timeout: the server probably can't accept — check its log for 'accept4: too many open files')") - } - break + firstErr.Do(func() { + fmt.Printf(" request %d failed: %v\n", i, err) + var ne net.Error + if errors.As(err, &ne) && ne.Timeout() { + fmt.Println(" (timeout: the server probably can't accept — check its log for 'accept: too many open files')") + } + stop.Store(true) + }) + return } - done++ + done.Add(1) switch m { case modeDrain: @@ -212,13 +235,33 @@ func runClient(m mode, n int, url string, headerTimeout time.Duration) { } } + goroutinesBefore := runtime.NumGoroutine() + start := time.Now() + + // Bounded concurrency: a buffered channel as a semaphore + WaitGroup. + // Same shape as exercise 01's CheckAll. + sem := make(chan struct{}, opts.concurrency) + var wg sync.WaitGroup + for i := 0; i < opts.n && !stop.Load(); i++ { + sem <- struct{}{} // blocks while `concurrency` requests are in flight + wg.Add(1) + go func(i int) { + defer wg.Done() + defer func() { <-sem }() + doOne(i) + }(i) + } + wg.Wait() + elapsed := time.Since(start) // Give the transport a moment to settle bookkeeping (put conn back to idle etc.). time.Sleep(50 * time.Millisecond) goroutinesAfter := runtime.NumGoroutine() - fmt.Printf(" requests : %d/%d ok in %v\n", done, n, elapsed.Round(time.Millisecond)) - fmt.Printf(" httptrace GotConn : reused=%d fresh=%d\n", reused, fresh) + fmt.Printf(" requests : %d/%d ok, %d failed, in %v (concurrency %d)\n", + done.Load(), opts.n, failed.Load(), elapsed.Round(time.Millisecond), opts.concurrency) + fmt.Printf(" httptrace GotConn : reused=%d fresh=%d (MaxIdleConnsPerHost=%d)\n", + reused.Load(), fresh.Load(), opts.idlePerHost) fmt.Printf(" goroutines : before=%d after=%d (delta %+d)\n", goroutinesBefore, goroutinesAfter, goroutinesAfter-goroutinesBefore) @@ -236,7 +279,14 @@ func main() { modeFlag := flag.String("mode", "all", "client: drain | close | leak | all") n := flag.Int("n", 20, "client: number of requests per mode") headerTimeout := flag.Duration("header-timeout", 0, "client: ResponseHeaderTimeout (0 = none; try 3s when the server is the one out of fds)") + concurrency := flag.Int("c", 1, "client: max requests in flight (1 = sequential)") + idlePerHost := flag.Int("idle-per-host", 10, "client: Transport.MaxIdleConnsPerHost (set below -c to watch drain mode stop reusing)") flag.Parse() + if *concurrency < 1 { + fmt.Fprintln(os.Stderr, "-c must be >= 1") + os.Exit(2) + } + opts := clientOpts{n: *n, concurrency: *concurrency, idlePerHost: *idlePerHost, headerTimeout: *headerTimeout} modes := []mode{modeDrain, modeClose, modeLeak} if *modeFlag != "all" { @@ -250,7 +300,7 @@ func main() { case "client": for _, m := range modes { fmt.Printf("== mode=%s (client only, target %s) ==\n", m, *url) - runClient(m, *n, *url, *headerTimeout) + runClient(m, *url, opts) fmt.Println() } if len(modes) == 1 && modes[0] == modeLeak { @@ -265,7 +315,7 @@ func main() { os.Exit(1) } fmt.Printf("== mode=%s ==\n", m) - runClient(m, *n, srv.URL(), *headerTimeout) + runClient(m, srv.URL(), opts) time.Sleep(50 * time.Millisecond) srv.report() fmt.Println()