kacerr/learning-go-for-sre-interview
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

body-drain: add -c concurrency and -idle-per-host to the client

Browse Source 
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GbnJj9ZqEx9nUvzkjdwfMq
This commit is contained in:
kacerr
2026-08-29 00:08:27 +02:00
co-authored by Claude Fable 5
parent f80ba127d8
commit 9230d0cd3d
2 changed files with 100 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
99-experiments/body-drain/README.md
+27
View File
@@ -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
99-experiments/body-drain/main.go
+73 -23
View File
@@ -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()