Compare commits
9
Commits
8dd8381b8c
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f08d4f44d4 | ||
|
|
b290957fc9 | ||
|
|
86a61da36f | ||
|
|
47ced49d6a | ||
|
|
5e2d9400ac | ||
|
|
1610cd69e0 | ||
|
|
9230d0cd3d | ||
|
|
f80ba127d8 | ||
|
|
57349285b4 |
@@ -1,2 +1,4 @@
|
||||
.idea/
|
||||
.junie/
|
||||
.claude/
|
||||
/server
|
||||
|
||||
+30
-1
@@ -17,22 +17,51 @@ package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
type Limiter struct {
|
||||
|
||||
// TODO: fields (mutex, current tokens, max, rate, last refill time)
|
||||
mu sync.Mutex
|
||||
tokens float64
|
||||
max float64
|
||||
rate float64
|
||||
last time.Time
|
||||
}
|
||||
|
||||
// NewLimiter returns a limiter allowing `rate` ops/sec with bursts up to `burst`.
|
||||
// TODO: implement me.
|
||||
func NewLimiter(rate float64, burst int) *Limiter {
|
||||
return &Limiter{}
|
||||
return &Limiter{tokens: float64(burst), rate: rate, max: float64(burst), last: time.Now()}
|
||||
}
|
||||
|
||||
// Allow reports whether one operation may proceed now.
|
||||
// TODO: implement me.
|
||||
func (l *Limiter) Allow() bool {
|
||||
/*
|
||||
1. Lock — refill + consume is a read-modify-write on two fields;
|
||||
without the lock two goroutines can both see 1 token and both take it.
|
||||
2. Lazy refill — tokens earned since the last call = elapsed * rate.
|
||||
float64 so fractions (0.6 of a token) carry over instead of truncating to 0.
|
||||
3. Cap at max — idle time must not build unlimited credit (burst is a ceiling).
|
||||
4. Update last on EVERY call, granted or not — otherwise the same interval
|
||||
is credited again on the next call.
|
||||
5. If at least one whole token is available, take it and allow; else deny.
|
||||
Non-blocking: never sleeps or waits.
|
||||
*/
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
l.tokens += l.rate * time.Since(l.last).Seconds()
|
||||
if l.tokens > l.max {
|
||||
l.tokens = l.max
|
||||
}
|
||||
l.last = time.Now()
|
||||
if l.tokens >= 1 {
|
||||
l.tokens--
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
|
||||
+36
-2
@@ -4,7 +4,7 @@
|
||||
// ----------
|
||||
// Log lines look like:
|
||||
//
|
||||
// 2026-08-28T10:00:01Z GET /api/users 200 123ms
|
||||
// 2026-08-28T10:00:01Z GET /api/users 200 123ms
|
||||
//
|
||||
// Implement TopN(r, n) so that it:
|
||||
//
|
||||
@@ -20,8 +20,11 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"cmp"
|
||||
"fmt"
|
||||
"io"
|
||||
"slices"
|
||||
"strings"
|
||||
)
|
||||
|
||||
@@ -38,7 +41,38 @@ func TopN(r io.Reader, n int) ([]KV, error) {
|
||||
// - map[string]int for counting
|
||||
// - copy map into []KV and sort by count desc (maps can't be sorted)
|
||||
// - truncate to n
|
||||
return nil, nil
|
||||
|
||||
m := make(map[string]int)
|
||||
sc := bufio.NewScanner(r)
|
||||
sc.Buffer(make([]byte, 0, 64*1024), 1024*1024)
|
||||
for sc.Scan() {
|
||||
//fmt.Println(sc.Text())
|
||||
fields := strings.Fields(sc.Text())
|
||||
//fmt.Printf("len: %v, fields: %v\n", len(fields), fields)
|
||||
if len(fields) < 3 {
|
||||
continue
|
||||
}
|
||||
m[fields[2]]++
|
||||
}
|
||||
|
||||
// copy over map into list of KV
|
||||
l := make([]KV, len(m))
|
||||
i := 0
|
||||
for k, v := range m {
|
||||
// println(i, k, v, len(l), len(m))
|
||||
l[i] = KV{k, v}
|
||||
i++
|
||||
}
|
||||
|
||||
// sort by count desc
|
||||
slices.SortFunc(l, func(a, b KV) int {
|
||||
return cmp.Compare(b.Count, a.Count)
|
||||
})
|
||||
if len(l) > n {
|
||||
l = l[:n]
|
||||
}
|
||||
|
||||
return l, nil
|
||||
}
|
||||
|
||||
func main() {
|
||||
|
||||
+50
-5
@@ -16,43 +16,88 @@
|
||||
// Run the tests: go test -race ./05-lru-cache/
|
||||
package main
|
||||
|
||||
import "fmt"
|
||||
import (
|
||||
"container/list"
|
||||
"fmt"
|
||||
"sync"
|
||||
)
|
||||
|
||||
type entry struct {
|
||||
key string
|
||||
val any
|
||||
}
|
||||
type LRU struct {
|
||||
// TODO: fields (mutex, capacity, map[string]*list.Element, *list.List)
|
||||
// Remember: the list element must store the KEY too, or you can't
|
||||
// delete the map entry on eviction.
|
||||
mu sync.Mutex
|
||||
capacity int
|
||||
lruList *list.List
|
||||
kvMap map[string]*list.Element
|
||||
}
|
||||
|
||||
// NewLRU creates an LRU cache with the given capacity (capacity >= 1).
|
||||
// TODO: implement me.
|
||||
func NewLRU(capacity int) *LRU {
|
||||
return &LRU{}
|
||||
return &LRU{capacity: capacity, lruList: list.New(), kvMap: make(map[string]*list.Element)}
|
||||
}
|
||||
|
||||
// Get returns the value and whether it was present. Hit = mark MRU.
|
||||
// TODO: implement me.
|
||||
func (c *LRU) Get(key string) (any, bool) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
if el, ok := c.kvMap[key]; ok {
|
||||
c.lruList.MoveToFront(el)
|
||||
return el.Value.(*entry).val, true
|
||||
}
|
||||
|
||||
return nil, false
|
||||
}
|
||||
|
||||
// Put inserts/updates a value, evicting the LRU entry if over capacity.
|
||||
// TODO: implement me.
|
||||
func (c *LRU) Put(key string, val any) {
|
||||
// lock mutex
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
// if key exists, update value and move it to the front of the list and return/quit
|
||||
if el, ok := c.kvMap[key]; ok {
|
||||
c.lruList.MoveToFront(el)
|
||||
el.Value.(*entry).val = val
|
||||
return
|
||||
}
|
||||
|
||||
// if len > capacity, remove the last element from both the list and the map
|
||||
if c.lruList.Len() >= c.capacity {
|
||||
// remove last element from list
|
||||
lastEl := c.lruList.Back()
|
||||
c.lruList.Remove(lastEl)
|
||||
// remove it by key from map
|
||||
delete(c.kvMap, lastEl.Value.(*entry).key)
|
||||
}
|
||||
|
||||
// if key doesn't exist, insert'
|
||||
el := c.lruList.PushFront(&entry{key: key, val: val})
|
||||
c.kvMap[key] = el
|
||||
|
||||
}
|
||||
|
||||
// Len returns the current number of entries.
|
||||
// TODO: implement me.
|
||||
func (c *LRU) Len() int {
|
||||
return 0
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
return c.lruList.Len()
|
||||
}
|
||||
|
||||
func main() {
|
||||
c := NewLRU(2)
|
||||
c.Put("a", 1)
|
||||
c.Put("b", 2)
|
||||
c.Get("a") // a is now MRU
|
||||
c.Put("c", 3) // evicts b
|
||||
c.Get("a") // a is now MRU
|
||||
c.Put("c", 3) // evicts b
|
||||
_, ok := c.Get("b")
|
||||
fmt.Println("b present:", ok) // false
|
||||
}
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"fmt"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestBasicPutGet(t *testing.T) {
|
||||
@@ -82,3 +83,42 @@ func TestConcurrentAccess(t *testing.T) {
|
||||
t.Fatalf("capacity exceeded under concurrency: %d", c.Len())
|
||||
}
|
||||
}
|
||||
|
||||
// TestLenRacesWithPut calls Len concurrently WITH writers, not after them.
|
||||
// TestConcurrentAccess only reads Len after wg.Wait(), which establishes
|
||||
// happens-before with every writer — so an unsynchronized Len can't race
|
||||
// there. The race detector reports only races that actually occur.
|
||||
func TestLenRacesWithPut(t *testing.T) {
|
||||
c := NewLRU(64)
|
||||
done := make(chan struct{})
|
||||
var wg sync.WaitGroup
|
||||
|
||||
wg.Add(1)
|
||||
go func() { // writer
|
||||
defer wg.Done()
|
||||
for i := 0; ; i++ {
|
||||
select {
|
||||
case <-done:
|
||||
return
|
||||
default:
|
||||
c.Put(fmt.Sprintf("k%d", i%128), i)
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
wg.Add(1)
|
||||
go func() { // reader, concurrently
|
||||
defer wg.Done()
|
||||
for i := 0; i < 10000; i++ {
|
||||
if n := c.Len(); n > 64 {
|
||||
t.Errorf("Len observed %d > capacity 64 (mid-Put state leaked)", n)
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
go func() { wg.Wait() }()
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
close(done)
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
@@ -0,0 +1,283 @@
|
||||
# 98 — HTTP server connection experiments
|
||||
|
||||
How many connections can a Go HTTP server hold, and what gives out first? This
|
||||
folder finds the **order of magnitude** (not a portable constant) for four
|
||||
different failure modes, and — just as important — teaches you to tell a real
|
||||
server ceiling from your own load generator running out first.
|
||||
|
||||
One simple server is the subject under test; one load client drives it in four
|
||||
modes. Run them in two terminals so each side has its own fd budget.
|
||||
|
||||
```
|
||||
98-http-server-experiments/
|
||||
server/main.go subject under test: /static /compute /sleep /stats, ConnState + in_flight counters
|
||||
client/main.go load generator: -mode throughput | idle | slow | slowloris
|
||||
loadlib/ tested core: Burn (CPU), Percentile, Classify (which side ran out)
|
||||
README.md this file — decisions log + pasted evidence
|
||||
```
|
||||
|
||||
Pure stdlib, zero dependencies. Part of the `gosre-exercises` module, so:
|
||||
|
||||
```sh
|
||||
go run ./98-http-server-experiments/server # terminal 1
|
||||
go run ./98-http-server-experiments/client -mode idle -n 2000 # terminal 2
|
||||
go test -race ./98-http-server-experiments/loadlib/
|
||||
```
|
||||
|
||||
## Run it
|
||||
|
||||
**Terminal 1 — the server.** Give it its own fd budget so *you* decide when it
|
||||
runs out, not the OS default. bash sets both soft and hard limits:
|
||||
|
||||
```sh
|
||||
ulimit -Hn 4096; ulimit -Sn 4096
|
||||
go run ./98-http-server-experiments/server # naive: no timeouts
|
||||
go run ./98-http-server-experiments/server -read-header-timeout 3s -idle-timeout 30s # hardened
|
||||
```
|
||||
|
||||
(Since Go 1.19 the runtime raises the soft `RLIMIT_NOFILE` to the hard limit at
|
||||
startup, so only the **hard** limit really caps you. bash's `ulimit -n` sets
|
||||
both; zsh's sets only the soft one — use `-Hn`.)
|
||||
|
||||
The server prints one line per second:
|
||||
|
||||
```
|
||||
19:12:20 accepted_total=10001 active=10001 in_flight=1 goroutines=10006 heap=107MiB
|
||||
```
|
||||
|
||||
Three numbers that are easy to conflate:
|
||||
|
||||
| number | meaning | via |
|
||||
|---|---|---|
|
||||
| `accepted_total` | every TCP conn ever accepted (only grows) | `ConnState` StateNew |
|
||||
| `active` | conns open **right now** — idle keep-alive *or* busy | `ConnState` new − closed |
|
||||
| `in_flight` | handlers **actually running** right now | middleware around the mux |
|
||||
|
||||
An idle keep-alive counts in `active` but not `in_flight`; a running `/sleep`
|
||||
counts in both. That gap is the whole point of the idle-vs-slow contrast.
|
||||
|
||||
**Terminal 2 — the client.** Give it *its own* fd budget, separate from the server:
|
||||
|
||||
```sh
|
||||
ulimit -Hn 64; ulimit -Sn 64
|
||||
go run ./98-http-server-experiments/client -mode idle -n 200
|
||||
```
|
||||
|
||||
## The four dimensions
|
||||
|
||||
### 1. Idle connection ceiling — `-mode idle`
|
||||
|
||||
Open `-n` keep-alive conns and hold them idle. Uses **raw `net.Conn`**, not an
|
||||
`http.Client` — with a fast endpoint the client's connection pool reuses conns
|
||||
faster than you dial, so N concurrent GETs do *not* reliably produce N
|
||||
simultaneous conns. Dialing raw and keeping each fd open guarantees exactly N.
|
||||
|
||||
```sh
|
||||
go run ./98-http-server-experiments/client -mode idle -n 10000 -duration 30s
|
||||
```
|
||||
|
||||
```
|
||||
== mode=idle target=http://127.0.0.1:8098 ==
|
||||
server before : active=1 in_flight=1 goroutines=6 heap=0MiB
|
||||
opened : 10000/10000 idle conns held, 0 failed
|
||||
client goroutines : before=3 after=3 (delta +0; raw fds, no per-conn goroutine)
|
||||
server after open : active=10001 in_flight=1 goroutines=10006 heap=107MiB ← active≈n, in_flight≈0
|
||||
```
|
||||
|
||||
**Reading it:** 10 000 idle conns cost the server ~107 MiB (from a <1 MiB
|
||||
baseline) and one goroutine each → **order of magnitude ~10 KiB per idle
|
||||
connection**. A million idle keep-alives would be ~10 GiB and a million
|
||||
goroutines — the server holds them fine until it runs out of fds or RAM, not CPU.
|
||||
Note the client's goroutine delta is **0**: raw fds have no per-conn goroutine,
|
||||
unlike `http.Transport`'s pool (2 goroutines per pooled conn).
|
||||
|
||||
### 2. Active slow requests — `-mode slow`
|
||||
|
||||
Fire `-c` concurrent `/sleep` requests and hold them *all in flight* for the
|
||||
duration. Same conn count as idle, but now each pins a **running** server
|
||||
goroutine. Prints the server's `in_flight` each second.
|
||||
|
||||
```sh
|
||||
go run ./98-http-server-experiments/client -mode slow -c 500 -duration 10s
|
||||
```
|
||||
|
||||
```
|
||||
== mode=slow target=http://127.0.0.1:8098 ==
|
||||
firing 500 concurrent /sleep?d=10s ...
|
||||
server 01:13:13 active=501 in_flight=501 goroutines=1006 heap=25MiB
|
||||
requests : 500 ok, 0 failed (of 500 held in flight)
|
||||
errors : none
|
||||
```
|
||||
|
||||
**Reading it:** `in_flight=501` (the 500 sleepers + the stats poll) vs idle's
|
||||
`in_flight=1` — that is the difference between "conn open" and "handler running."
|
||||
Each in-flight request is one goroutine on a goroutine-per-request server; 500
|
||||
held costs ~1000 goroutines and a few MiB. Push `-c` up in orders of magnitude to
|
||||
find where the *server* (goroutine stacks, fds) — not the client — gives out.
|
||||
`/sleep` honors `r.Context()`, so a client that hangs up or a server timeout
|
||||
frees the goroutine immediately rather than sleeping to term.
|
||||
|
||||
### 3. Timeouts & slowloris — `-mode slowloris`
|
||||
|
||||
Open `-c` raw TCP conns, send a request line + one header on each, then dribble a
|
||||
bogus header every `-drip` and **never** send the terminating blank line. The
|
||||
handler never starts. The only thing that reclaims these conns is the server's
|
||||
`ReadHeaderTimeout`.
|
||||
|
||||
**Naive server** (`go run .../server`, no timeouts):
|
||||
|
||||
```
|
||||
== mode=slowloris target=http://127.0.0.1:8098 ==
|
||||
opening 200 slowloris conns ...
|
||||
server 01:13:26 active=201 in_flight=1 goroutines=206 heap=26MiB ← stuck, handler never runs
|
||||
reclaimed by server: 0
|
||||
(0 reclaimed: the server has no -read-header-timeout — these conns would stick until fds run out)
|
||||
```
|
||||
|
||||
**Hardened server** (`go run .../server -read-header-timeout 3s`):
|
||||
|
||||
```
|
||||
server 01:13:40 active=201 in_flight=1 goroutines=206 heap=2MiB
|
||||
server 01:13:42 active=1 in_flight=1 goroutines=6 heap=3MiB ← reclaimed ~3s after opening
|
||||
reclaimed by server: 200 (server closed the conn mid-headers → -read-header-timeout worked)
|
||||
```
|
||||
|
||||
**Reading it:** with no `ReadHeaderTimeout`, 200 clients doing *nothing* consume
|
||||
200 connections indefinitely — a handful of hosts can exhaust a server's fds
|
||||
without ever completing a request. Setting `ReadHeaderTimeout=3s` drops each conn
|
||||
~3 s after its last progress and `active` stays flat. The same logic applies to
|
||||
`IdleTimeout` for *completed* keep-alives that then go quiet. **Every production
|
||||
`http.Server` should set these; the stdlib defaults are all zero (off).**
|
||||
|
||||
### 4. Client-side limits — when the measurement is lying
|
||||
|
||||
The load generator hits its own walls, often *before* the server does. Push idle
|
||||
past the ephemeral-port ceiling and watch the **client** fail while the server is
|
||||
fine:
|
||||
|
||||
```sh
|
||||
go run ./98-http-server-experiments/client -mode idle -n 20000 -duration 5s
|
||||
```
|
||||
|
||||
```
|
||||
opened : 16312/20000 idle conns held, 3688 failed
|
||||
errors : client-port-limit=3688
|
||||
server after open : active=16313 in_flight=1 goroutines=16318 heap=199MiB
|
||||
```
|
||||
|
||||
**Reading it:** this host has 16 384 ephemeral ports (`49152–65535`), and every
|
||||
conn to the same server `ip:port` needs a distinct one. At 16 312 the client got
|
||||
`connect: can't assign requested address` — a **client-side** wall. The server
|
||||
was untroubled (it held 16 313 conns on ~199 MiB). If you didn't cross-check the
|
||||
server's `/stats`, you might wrongly conclude "the server maxes out at ~16 k."
|
||||
Every mode fetches `/stats` at start and end for exactly this reason.
|
||||
|
||||
## What to watch
|
||||
|
||||
- **`active` vs `in_flight`** — open connections vs running handlers (dimensions 1 & 2).
|
||||
- **`goroutines`** — ~1 per server conn, ~2 per pooled *client* conn, ~0 per raw client conn.
|
||||
- **`heap`** — server memory; ~10 KiB per idle conn on this host.
|
||||
- **`reused` vs `fresh`** (throughput) — is the pool actually reusing conns, or dialing every time?
|
||||
- **`p50/p99/max`** latency and **error histogram** classified by which side ran out.
|
||||
|
||||
The throughput control shows how fast it goes when nothing is broken, and how a
|
||||
CPU-bound endpoint plateaus near `GOMAXPROCS`:
|
||||
|
||||
```sh
|
||||
go run ./98-http-server-experiments/client -mode throughput -path /static -c 50 -duration 3s
|
||||
# requests : 445013 ok, 0 failed → 148338 req/s latency p50=259µs p99=1.4ms reused=444971 fresh=42
|
||||
|
||||
go run ./98-http-server-experiments/client -mode throughput -path "/compute?difficulty=200" -c 50 -duration 3s
|
||||
# requests : 370 ok, 0 failed → 123 req/s latency p50=393ms p99=1.24s
|
||||
```
|
||||
|
||||
150 k req/s on `/static` collapses to ~120 req/s on `/compute?difficulty=200`:
|
||||
that's CPU-bound, not connection-bound — 8 cores fully busy, more concurrency
|
||||
just deepens the queue.
|
||||
|
||||
## Reading it — symptom → cause
|
||||
|
||||
| symptom | cause |
|
||||
|---|---|
|
||||
| client `dial: socket: too many open files` | **client** fd/ulimit — raise the client's limit; the measurement was client-bound |
|
||||
| client `connect: can't assign requested address` | **client** ephemeral ports exhausted — reuse conns, or use multiple client IPs |
|
||||
| server log `accept: too many open files` | **server** fd ceiling — the real thing |
|
||||
| `timeout awaiting response headers` / `connection reset` | server can't accept/service; client is fine |
|
||||
| slowloris conns stick forever | server has no `ReadHeaderTimeout` |
|
||||
| idle keep-alives never drop | server has no `IdleTimeout` |
|
||||
| throughput plateaus near `GOMAXPROCS` on `/compute` | CPU-bound, not conn-bound |
|
||||
|
||||
The `Classify` helper in `loadlib` turns these strings into counted classes so
|
||||
the client's error histogram tells you which side gave out.
|
||||
|
||||
## Client-side pitfalls (know when the number is yours, not the server's)
|
||||
|
||||
- **Ephemeral port exhaustion.** One client → one server `ip:port` gets ~16 k–28 k
|
||||
ephemeral ports (`net.inet.ip.portrange.first/last` on macOS,
|
||||
`net.ipv4.ip_local_port_range` on Linux). Above that, `connect` fails. Fix for
|
||||
the *test*: reuse keep-alive conns (don't open a fresh 4-tuple per request), or
|
||||
drive from several client IPs/hosts.
|
||||
- **TIME_WAIT.** Every actively-closed conn sits in TIME_WAIT ~60 s holding its
|
||||
4-tuple; a close-per-request pattern piles them up
|
||||
(`netstat -an | grep -c TIME_WAIT`) and starves ephemeral ports. Keep-alive
|
||||
reuse avoids it. `SO_REUSEADDR` does not help the client here.
|
||||
- **`MaxIdleConnsPerHost` (default 2).** If it's below your concurrency, the
|
||||
client dials fresh every round and you measure TCP handshakes, not the server.
|
||||
The client auto-sizes it to `-c`; override with `-idle-per-host`.
|
||||
- **`Client.Timeout` / `ResponseHeaderTimeout`.** A timeout force-closes a stuck
|
||||
conn, which *hides* a hold or a leak. The client defaults `-timeout 0`; `slow`
|
||||
mode disables the header timeout entirely because its responses are slow **by
|
||||
design** (an early version misreported held requests as "server-cant-accept").
|
||||
- **Always cross-check `/stats`.** If the client says "failed" but the server
|
||||
never saw the conn, the client ran out first.
|
||||
|
||||
## Host caveats (numbers here are not portable)
|
||||
|
||||
Measured on: Apple M-series (Mac15,12), 8 cores, macOS (darwin/arm64), go 1.27.
|
||||
`kern.maxfilesperproc=92160`, ephemeral ports `49152–65535` (16 384).
|
||||
|
||||
- **macOS vs Linux differ.** fd caps (`kern.maxfilesperproc` / `ulimit`), TIME_WAIT
|
||||
and port-range sysctls, and even `accept`/`dial` error strings are not the same.
|
||||
- **Listen backlog** caps the accept queue independently of your handler:
|
||||
`kern.ipc.somaxconn` (macOS) / `net.core.somaxconn` (Linux).
|
||||
- **One process, shared fd limit.** If you run both roles in one process they
|
||||
share the fd budget and the server usually dies first — use two terminals.
|
||||
|
||||
## Decisions log (what we built, how, and why)
|
||||
|
||||
- **One server binary + one client binary; scenarios are `-mode`, not separate
|
||||
binaries.** The four dimensions stress the *same* server; splitting them would
|
||||
duplicate the transport/httptrace/reporting scaffolding four times. Two-process
|
||||
split so each side gets its own fd/ulimit (same rationale as `99-experiments/body-drain`'s `-role`).
|
||||
- **Three live counters, not one.** `active` (ConnState) and `in_flight`
|
||||
(middleware) are tracked separately because "conn open" ≠ "handler running" —
|
||||
that distinction *is* the idle-vs-slow experiment.
|
||||
- **`idle` mode uses raw `net.Conn`, not `http.Client`.** The pool reuses conns
|
||||
faster than we dial against a fast endpoint, so a pool-based idle mode
|
||||
under-counts non-deterministically (20 000 GETs held only ~8 700). Raw dials
|
||||
give exactly N — and, as a bonus, expose ephemeral-port exhaustion cleanly.
|
||||
- **`slowloris` uses raw `net.Conn`.** The stdlib client won't send a deliberately
|
||||
incomplete request, so we speak HTTP by hand: request line + headers, never the
|
||||
terminating blank line.
|
||||
- **`Burn` is deterministic and non-elidable.** A data-dependent integer hash of
|
||||
`difficulty × 200k` iterations; the handler writes the checksum so the compiler
|
||||
can't drop the loop. No `runtime.Gosched`, so it genuinely holds a P. Tested by
|
||||
iteration count, not wall-clock, to stay non-flaky.
|
||||
- **Zero dependencies.** No `x/net/netutil.LimitListener` for `-max-conns`; a real
|
||||
server caps concurrency with a semaphore-guarded handler or a wrapped listener,
|
||||
and we'd rather observe the OS ceiling than hide it.
|
||||
- **Server timeouts default to OFF.** So the slowloris vulnerability is
|
||||
demonstrable out of the box; turn `-read-header-timeout` / `-idle-timeout` on to
|
||||
watch the server reclaim.
|
||||
- **Network ceilings are not unit-tested.** They're environmental (ulimit, sysctl,
|
||||
host); only `Burn`, `Percentile`, and `Classify` are tested. The ceilings live
|
||||
here as pasted evidence.
|
||||
|
||||
## References
|
||||
|
||||
- `net/http.Server` timeout fields — https://pkg.go.dev/net/http#Server
|
||||
- Cloudflare, "The complete guide to Go net/http timeouts" — https://blog.cloudflare.com/the-complete-guide-to-golang-net-http-timeouts/
|
||||
- `net/http/httptrace` — https://pkg.go.dev/net/http/httptrace
|
||||
- `http.ConnState` — https://pkg.go.dev/net/http#ConnState
|
||||
- Slowloris (CWE-400 / low-and-slow) — https://en.wikipedia.org/wiki/Slowloris_(cyber_attack)
|
||||
- Ephemeral ports & TIME_WAIT — https://www.rfc-editor.org/rfc/rfc6335, `sysctl net.inet.ip.portrange` / `net.ipv4.ip_local_port_range`
|
||||
@@ -0,0 +1,487 @@
|
||||
// Load generator for the experiment server. One binary, four modes, each a
|
||||
// different way to ask "how many connections, and what gives out first?".
|
||||
//
|
||||
// Modes (-mode):
|
||||
//
|
||||
// throughput closed loop: -c workers hammer -path for -duration, reusing conns.
|
||||
// The control — how fast when nothing is broken. Point -path at
|
||||
// /compute?difficulty=N to watch it plateau near GOMAXPROCS.
|
||||
// idle open -n keep-alive conns to /static and HOLD them (don't close).
|
||||
// Measures the cost of an idle connection: fds, goroutines, memory.
|
||||
// slow fire -c concurrent /sleep requests and hold them all in flight.
|
||||
// Same conn count as idle, but each pins a running server goroutine.
|
||||
// slowloris -c raw TCP conns that dribble headers forever and never finish
|
||||
// the request. Defeated only by the server's -read-header-timeout.
|
||||
//
|
||||
// Every mode cross-checks the server's own /stats so you can tell a real server
|
||||
// ceiling from your own client running out first. Which error you get tells you
|
||||
// which side gave out:
|
||||
//
|
||||
// dial ...: socket: too many open files → THIS process (raise the client's ulimit)
|
||||
// timeout awaiting response headers → the server can't accept
|
||||
// connection reset by peer → the server closed on you (or a timeout fired)
|
||||
// connection refused → nothing listening
|
||||
//
|
||||
// Give the client its own fd budget, separate from the server's:
|
||||
//
|
||||
// ulimit -Hn 64; ulimit -Sn 64; go run ./98-http-server-experiments/client -mode idle -n 200
|
||||
//
|
||||
// (bash sets soft+hard; zsh only soft — use -Hn. Since Go 1.19 only the hard
|
||||
// limit really caps you.)
|
||||
package main
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"flag"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/httptrace"
|
||||
"os"
|
||||
"os/signal"
|
||||
"runtime"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"gosre-exercises/98-http-server-experiments/loadlib"
|
||||
)
|
||||
|
||||
// serverStats mirrors the server's /stats JSON for the cross-check.
|
||||
type serverStats struct {
|
||||
AcceptedTotal int64 `json:"accepted_total"`
|
||||
ActiveConns int64 `json:"active_conns"`
|
||||
InFlight int64 `json:"in_flight"`
|
||||
Goroutines int `json:"goroutines"`
|
||||
HeapBytes uint64 `json:"heap_bytes"`
|
||||
}
|
||||
|
||||
func (s serverStats) String() string {
|
||||
return fmt.Sprintf("active=%d in_flight=%d goroutines=%d heap=%dMiB",
|
||||
s.ActiveConns, s.InFlight, s.Goroutines, s.HeapBytes>>20)
|
||||
}
|
||||
|
||||
// fetchStats reads the server's /stats over a throwaway conn (own transport so it
|
||||
// never borrows from or perturbs the pool we're measuring).
|
||||
func fetchStats(base string) (serverStats, error) {
|
||||
var st serverStats
|
||||
c := &http.Client{Timeout: 2 * time.Second}
|
||||
resp, err := c.Get(base + "/stats")
|
||||
if err != nil {
|
||||
return st, err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
err = json.NewDecoder(resp.Body).Decode(&st)
|
||||
return st, err
|
||||
}
|
||||
|
||||
// opts bundle the CLI knobs the modes read.
|
||||
type opts struct {
|
||||
url, path string
|
||||
n, c int
|
||||
duration time.Duration
|
||||
idlePerHost int
|
||||
headerTimeout time.Duration
|
||||
clientTimeout time.Duration
|
||||
drip time.Duration
|
||||
}
|
||||
|
||||
// newTransport builds a fresh pool so runs don't share state. idlePerHost must be
|
||||
// >= the concurrency you want reused, or conns get dropped instead of pooled.
|
||||
func newTransport(o opts) *http.Transport {
|
||||
tr := http.DefaultTransport.(*http.Transport).Clone()
|
||||
tr.MaxIdleConns = o.n + o.c
|
||||
tr.MaxIdleConnsPerHost = o.idlePerHost
|
||||
tr.ResponseHeaderTimeout = o.headerTimeout
|
||||
return tr
|
||||
}
|
||||
|
||||
// errHistogram counts failures by which side they blame.
|
||||
type errHistogram struct {
|
||||
mu sync.Mutex
|
||||
m map[loadlib.ErrClass]int
|
||||
}
|
||||
|
||||
func newErrHistogram() *errHistogram { return &errHistogram{m: map[loadlib.ErrClass]int{}} }
|
||||
func (h *errHistogram) add(err error) {
|
||||
h.mu.Lock()
|
||||
h.m[loadlib.Classify(err)]++
|
||||
h.mu.Unlock()
|
||||
}
|
||||
func (h *errHistogram) String() string {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
if len(h.m) == 0 {
|
||||
return "none"
|
||||
}
|
||||
s := ""
|
||||
for cls, n := range h.m {
|
||||
s += fmt.Sprintf("%s=%d ", cls, n)
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// ---- mode: throughput ----
|
||||
|
||||
// runThroughput drives a closed loop: c workers each loop Do()+drain for the
|
||||
// duration. Conns are reused, so this measures service speed, not dial cost.
|
||||
func runThroughput(o opts) {
|
||||
tr := newTransport(o)
|
||||
defer tr.CloseIdleConnections()
|
||||
|
||||
var reused, fresh, ok, failed atomic.Int64
|
||||
trace := &httptrace.ClientTrace{GotConn: func(i httptrace.GotConnInfo) {
|
||||
if i.Reused {
|
||||
reused.Add(1)
|
||||
} else {
|
||||
fresh.Add(1)
|
||||
}
|
||||
}}
|
||||
ctx := httptrace.WithClientTrace(context.Background(), trace)
|
||||
deadline := time.Now().Add(o.duration)
|
||||
client := &http.Client{Transport: tr, Timeout: o.clientTimeout}
|
||||
hist := newErrHistogram()
|
||||
target := o.url + o.path
|
||||
|
||||
var wg sync.WaitGroup
|
||||
var lats []loadlib.Latencies
|
||||
lats = make([]loadlib.Latencies, o.c)
|
||||
for w := 0; w < o.c; w++ {
|
||||
wg.Add(1)
|
||||
go func(w int) {
|
||||
defer wg.Done()
|
||||
for time.Now().Before(deadline) {
|
||||
start := time.Now()
|
||||
req, _ := http.NewRequestWithContext(ctx, http.MethodGet, target, nil)
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
failed.Add(1)
|
||||
hist.add(err)
|
||||
continue
|
||||
}
|
||||
io.Copy(io.Discard, resp.Body)
|
||||
resp.Body.Close()
|
||||
lats[w].Add(time.Since(start))
|
||||
ok.Add(1)
|
||||
}
|
||||
}(w)
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
var all loadlib.Latencies
|
||||
for i := range lats {
|
||||
all.Merge(&lats[i])
|
||||
}
|
||||
p50, p99, max := all.Snapshot()
|
||||
rps := float64(ok.Load()) / o.duration.Seconds()
|
||||
fmt.Printf(" target : %s (concurrency %d, %v)\n", target, o.c, o.duration)
|
||||
fmt.Printf(" requests : %d ok, %d failed → %.0f req/s\n", ok.Load(), failed.Load(), rps)
|
||||
fmt.Printf(" latency : p50=%v p99=%v max=%v\n", p50.Round(time.Microsecond), p99.Round(time.Microsecond), max.Round(time.Microsecond))
|
||||
fmt.Printf(" conns : reused=%d fresh=%d\n", reused.Load(), fresh.Load())
|
||||
fmt.Printf(" errors : %s\n", hist)
|
||||
}
|
||||
|
||||
// ---- mode: idle ----
|
||||
|
||||
// runIdle opens n keep-alive conns and HOLDS them idle, then reports the server's
|
||||
// view. It uses raw net.Conn rather than http.Client on purpose: with a fast
|
||||
// endpoint the client's connection pool reuses conns faster than we dial, so n
|
||||
// concurrent GETs do NOT reliably produce n simultaneous conns. Dialing raw and
|
||||
// keeping each fd open guarantees exactly n idle keep-alives — the number we want
|
||||
// to be honest about. Each does one real GET /static (so the server completes a
|
||||
// request and parks the conn as idle: active=n, in_flight≈0), then we sit on it.
|
||||
func runIdle(o opts) {
|
||||
before, _ := fetchStats(o.url)
|
||||
fmt.Printf(" server before : %s\n", before)
|
||||
|
||||
host := hostPort(o.url)
|
||||
grBefore := runtime.NumGoroutine()
|
||||
|
||||
var opened, failed atomic.Int64
|
||||
hist := newErrHistogram()
|
||||
conns := make([]net.Conn, 0, o.n)
|
||||
var mu sync.Mutex
|
||||
|
||||
// Bounded dial concurrency: fast enough to open thousands quickly, without a
|
||||
// thundering herd of goroutines. Each conn, once its response is drained,
|
||||
// just sits open — no per-conn goroutine, unlike the transport's 2-per-conn.
|
||||
sem := make(chan struct{}, 256)
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < o.n; i++ {
|
||||
sem <- struct{}{}
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer func() { <-sem }()
|
||||
conn, err := net.DialTimeout("tcp", host, 5*time.Second)
|
||||
if err != nil {
|
||||
failed.Add(1)
|
||||
hist.add(err)
|
||||
return
|
||||
}
|
||||
// One real keep-alive request, response drained, conn left open.
|
||||
if _, err := fmt.Fprintf(conn, "GET /static HTTP/1.1\r\nHost: %s\r\nConnection: keep-alive\r\n\r\n", host); err != nil {
|
||||
failed.Add(1)
|
||||
hist.add(err)
|
||||
conn.Close()
|
||||
return
|
||||
}
|
||||
br := bufio.NewReader(conn)
|
||||
resp, err := http.ReadResponse(br, nil)
|
||||
if err != nil {
|
||||
failed.Add(1)
|
||||
hist.add(err)
|
||||
conn.Close()
|
||||
return
|
||||
}
|
||||
io.Copy(io.Discard, resp.Body)
|
||||
resp.Body.Close()
|
||||
mu.Lock()
|
||||
conns = append(conns, conn)
|
||||
mu.Unlock()
|
||||
opened.Add(1)
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
time.Sleep(100 * time.Millisecond) // let the server finish parking them as idle
|
||||
|
||||
peak, _ := fetchStats(o.url)
|
||||
fmt.Printf(" opened : %d/%d idle conns held, %d failed\n", opened.Load(), o.n, failed.Load())
|
||||
fmt.Printf(" errors : %s\n", hist)
|
||||
fmt.Printf(" client goroutines : before=%d after=%d (delta %+d; raw fds, no per-conn goroutine)\n",
|
||||
grBefore, runtime.NumGoroutine(), runtime.NumGoroutine()-grBefore)
|
||||
fmt.Printf(" server after open : %s ← active≈n, in_flight≈0: these conns are OPEN but no handler runs\n", peak)
|
||||
fmt.Printf(" holding %d idle conns for %v — watch the server's active= stay put. Ctrl-C to release early.\n", opened.Load(), o.duration)
|
||||
|
||||
holdFor(o.url, o.duration)
|
||||
|
||||
for _, c := range conns {
|
||||
c.Close()
|
||||
}
|
||||
time.Sleep(200 * time.Millisecond)
|
||||
after, _ := fetchStats(o.url)
|
||||
fmt.Printf(" server after close : %s (active should fall back toward its baseline)\n", after)
|
||||
}
|
||||
|
||||
// ---- mode: slow ----
|
||||
|
||||
// runSlow fires c concurrent /sleep requests that each stay in flight for the
|
||||
// whole duration, so c server goroutines are pinned at once. Prints the server's
|
||||
// in_flight/goroutines each second while they're held.
|
||||
func runSlow(o opts) {
|
||||
before, _ := fetchStats(o.url)
|
||||
fmt.Printf(" server before : %s\n", before)
|
||||
|
||||
// By design every response is slow (it arrives only after the sleep), so we
|
||||
// must NOT apply a response-header timeout here — it would abort legitimate
|
||||
// slow responses and lie that the server "can't accept". Instead we bound
|
||||
// each request with a context deadline a bit past the sleep, so a truly
|
||||
// stuck request still unblocks wg.Wait() and gets reported.
|
||||
tr := newTransport(o)
|
||||
tr.ResponseHeaderTimeout = 0
|
||||
defer tr.CloseIdleConnections()
|
||||
client := &http.Client{Transport: tr}
|
||||
target := fmt.Sprintf("%s/sleep?d=%s", o.url, o.duration)
|
||||
reqTimeout := o.duration + 10*time.Second
|
||||
|
||||
var ok, failed atomic.Int64
|
||||
hist := newErrHistogram()
|
||||
|
||||
stopPoll := make(chan struct{})
|
||||
go pollStats(o.url, stopPoll)
|
||||
|
||||
fmt.Printf(" firing %d concurrent /sleep?d=%v ...\n", o.c, o.duration)
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < o.c; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
ctx, cancel := context.WithTimeout(context.Background(), reqTimeout)
|
||||
defer cancel()
|
||||
req, _ := http.NewRequestWithContext(ctx, http.MethodGet, target, nil)
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
failed.Add(1)
|
||||
hist.add(err)
|
||||
return
|
||||
}
|
||||
io.Copy(io.Discard, resp.Body)
|
||||
resp.Body.Close()
|
||||
ok.Add(1)
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
close(stopPoll)
|
||||
|
||||
fmt.Printf(" requests : %d ok, %d failed (of %d held in flight)\n", ok.Load(), failed.Load(), o.c)
|
||||
fmt.Printf(" errors : %s\n", hist)
|
||||
}
|
||||
|
||||
// ---- mode: slowloris ----
|
||||
|
||||
// runSlowloris opens c raw TCP conns, sends a request line + one header on each,
|
||||
// then dribbles a bogus header every -drip and NEVER sends the terminating blank
|
||||
// line. The handler never starts. A server with -read-header-timeout closes the
|
||||
// conn after that timeout; we detect it as a read returning EOF/reset and count
|
||||
// how many the server reclaimed.
|
||||
func runSlowloris(o opts) {
|
||||
host := hostPort(o.url)
|
||||
before, _ := fetchStats(o.url)
|
||||
fmt.Printf(" server before : %s\n", before)
|
||||
fmt.Printf(" opening %d slowloris conns to %s (drip every %v) ...\n", o.c, host, o.drip)
|
||||
|
||||
var held, reclaimed, dialFailed atomic.Int64
|
||||
hist := newErrHistogram()
|
||||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
|
||||
defer stop()
|
||||
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < o.c; i++ {
|
||||
conn, err := net.DialTimeout("tcp", host, 5*time.Second)
|
||||
if err != nil {
|
||||
dialFailed.Add(1)
|
||||
hist.add(err)
|
||||
continue
|
||||
}
|
||||
held.Add(1)
|
||||
wg.Add(1)
|
||||
go func(conn net.Conn) {
|
||||
defer wg.Done()
|
||||
defer conn.Close()
|
||||
// Incomplete request: a valid start line and Host, but no blank line.
|
||||
fmt.Fprintf(conn, "GET /static HTTP/1.1\r\nHost: %s\r\n", host)
|
||||
dripTick := time.NewTicker(o.drip)
|
||||
defer dripTick.Stop()
|
||||
// A reader goroutine turns a server-side close into a signal.
|
||||
closed := make(chan struct{})
|
||||
go func() {
|
||||
buf := make([]byte, 1)
|
||||
conn.Read(buf) // blocks until the server closes → EOF/reset
|
||||
close(closed)
|
||||
}()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-closed:
|
||||
reclaimed.Add(1)
|
||||
return
|
||||
case <-dripTick.C:
|
||||
// One more bogus header — still no terminating blank line.
|
||||
if _, err := fmt.Fprint(conn, "X-Keep: alive\r\n"); err != nil {
|
||||
reclaimed.Add(1)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}(conn)
|
||||
}
|
||||
|
||||
// Report the server's view periodically while conns are held.
|
||||
stopPoll := make(chan struct{})
|
||||
go pollStats(o.url, stopPoll)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
case <-time.After(o.duration):
|
||||
}
|
||||
stop() // stop the drip loops
|
||||
close(stopPoll) // stop polling
|
||||
wg.Wait()
|
||||
|
||||
after, _ := fetchStats(o.url)
|
||||
fmt.Printf(" held : %d conns opened, %d dial-failed\n", held.Load(), dialFailed.Load())
|
||||
fmt.Printf(" reclaimed by server: %d (server closed the conn mid-headers → -read-header-timeout worked)\n", reclaimed.Load())
|
||||
fmt.Printf(" errors : %s\n", hist)
|
||||
fmt.Printf(" server after : %s\n", after)
|
||||
if reclaimed.Load() == 0 {
|
||||
fmt.Println(" (0 reclaimed: the server has no -read-header-timeout — these conns would stick until fds run out)")
|
||||
}
|
||||
}
|
||||
|
||||
// ---- helpers ----
|
||||
|
||||
// pollStats prints the server's /stats once a second until stop is closed.
|
||||
func pollStats(base string, stop <-chan struct{}) {
|
||||
tick := time.NewTicker(time.Second)
|
||||
defer tick.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-stop:
|
||||
return
|
||||
case <-tick.C:
|
||||
if st, err := fetchStats(base); err == nil {
|
||||
fmt.Printf(" server %s %s\n", time.Now().Format("15:04:05"), st)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// holdFor blocks for d or until Ctrl-C, printing the server's view each second so
|
||||
// you can watch the held conns sit there.
|
||||
func holdFor(base string, d time.Duration) {
|
||||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
|
||||
defer stop()
|
||||
stopPoll := make(chan struct{})
|
||||
go pollStats(base, stopPoll)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
case <-time.After(d):
|
||||
}
|
||||
close(stopPoll)
|
||||
}
|
||||
|
||||
// hostPort turns http://host:port[/...] into host:port for net.Dial.
|
||||
func hostPort(rawurl string) string {
|
||||
s := strings.TrimPrefix(rawurl, "http://")
|
||||
if i := strings.IndexByte(s, '/'); i >= 0 {
|
||||
s = s[:i]
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
func main() {
|
||||
mode := flag.String("mode", "throughput", "throughput | idle | slow | slowloris")
|
||||
url := flag.String("url", "http://127.0.0.1:8098", "server base URL")
|
||||
path := flag.String("path", "/static", "throughput: path to hit, e.g. /static or /compute?difficulty=50")
|
||||
n := flag.Int("n", 1000, "idle: number of keep-alive conns to open and hold")
|
||||
c := flag.Int("c", 50, "throughput/slow/slowloris: concurrency (workers / held conns)")
|
||||
duration := flag.Duration("duration", 10*time.Second, "how long to run / hold")
|
||||
idlePerHost := flag.Int("idle-per-host", 0, "Transport.MaxIdleConnsPerHost (0 = auto: n for idle, c otherwise)")
|
||||
headerTimeout := flag.Duration("header-timeout", 3*time.Second, "Transport.ResponseHeaderTimeout (0 = none)")
|
||||
clientTimeout := flag.Duration("timeout", 0, "http.Client.Timeout (0 = none; a timeout force-closes stuck conns and hides holds)")
|
||||
drip := flag.Duration("drip", 10*time.Second, "slowloris: interval between dribbled header bytes")
|
||||
flag.Parse()
|
||||
|
||||
o := opts{
|
||||
url: *url, path: *path, n: *n, c: *c, duration: *duration,
|
||||
idlePerHost: *idlePerHost, headerTimeout: *headerTimeout,
|
||||
clientTimeout: *clientTimeout, drip: *drip,
|
||||
}
|
||||
// Auto-size the idle pool so conns are actually reused/held, not dropped.
|
||||
if o.idlePerHost == 0 {
|
||||
if *mode == "idle" {
|
||||
o.idlePerHost = o.n
|
||||
} else {
|
||||
o.idlePerHost = o.c
|
||||
}
|
||||
}
|
||||
|
||||
fmt.Printf("== mode=%s target=%s ==\n", *mode, o.url)
|
||||
switch *mode {
|
||||
case "throughput":
|
||||
runThroughput(o)
|
||||
case "idle":
|
||||
runIdle(o)
|
||||
case "slow":
|
||||
runSlow(o)
|
||||
case "slowloris":
|
||||
runSlowloris(o)
|
||||
default:
|
||||
fmt.Fprintf(os.Stderr, "unknown -mode %q (throughput|idle|slow|slowloris)\n", *mode)
|
||||
os.Exit(2)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,164 @@
|
||||
// Package loadlib holds the small, pure pieces the server and load client share
|
||||
// and that are worth a unit test: a deterministic CPU burn, a percentile helper,
|
||||
// and an error classifier that says which side of a connection ran out.
|
||||
//
|
||||
// Everything network-shaped lives in the two binaries; those ceilings are
|
||||
// environmental (ulimit/sysctl/host) and are documented as pasted evidence in
|
||||
// the folder README rather than asserted in a test.
|
||||
package loadlib
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"net"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// burnIters is the inner-loop count per unit of difficulty. Kept as a named
|
||||
// constant so the test can reason about total iterations without a wall clock.
|
||||
const burnIters = 200_000
|
||||
|
||||
// Iterations returns how many inner-loop steps Burn(difficulty) will run.
|
||||
// It is the seam the test asserts on: cost is linear in difficulty, no timing.
|
||||
func Iterations(difficulty int) int {
|
||||
if difficulty < 1 {
|
||||
difficulty = 1
|
||||
}
|
||||
return difficulty * burnIters
|
||||
}
|
||||
|
||||
// Burn occupies a CPU for an amount of work proportional to difficulty and
|
||||
// returns a checksum. The caller must use the return value (write it to the
|
||||
// response) so the compiler cannot elide the loop.
|
||||
//
|
||||
// The mix is a cheap integer hash with a data dependency across iterations, so
|
||||
// each step depends on the last and the optimizer can't collapse the loop or
|
||||
// vectorize it away. There is deliberately no runtime.Gosched inside: Burn is
|
||||
// meant to genuinely hold a P, which is the whole point of the /compute probe
|
||||
// (how many CPU-bound requests before GOMAXPROCS saturates).
|
||||
func Burn(difficulty int) uint64 {
|
||||
var x uint64 = 1469598103934665603 // FNV-1a offset basis, just a seed
|
||||
n := Iterations(difficulty)
|
||||
for i := 0; i < n; i++ {
|
||||
x ^= uint64(i)
|
||||
x *= 1099511628211 // FNV-1a prime
|
||||
x = x<<13 | x>>51 // rotate so high bits feed back in
|
||||
}
|
||||
return x
|
||||
}
|
||||
|
||||
// Percentile returns the p-th percentile (p in [0,1]) of an already-sorted
|
||||
// slice. Empty → 0, p<=0 → first, p>=1 → last. Nearest-rank, no interpolation —
|
||||
// good enough for "orders of magnitude" latency reporting.
|
||||
func Percentile(sorted []time.Duration, p float64) time.Duration {
|
||||
if len(sorted) == 0 {
|
||||
return 0
|
||||
}
|
||||
if p < 0 {
|
||||
p = 0
|
||||
}
|
||||
if p > 1 {
|
||||
p = 1
|
||||
}
|
||||
i := int(float64(len(sorted)-1) * p)
|
||||
return sorted[i]
|
||||
}
|
||||
|
||||
// Latencies accumulates request durations and reports percentiles, keeping the
|
||||
// client's main loop thin. Not safe for concurrent use — each worker should own
|
||||
// one and they get merged, or a single collector goroutine owns it.
|
||||
type Latencies struct {
|
||||
d []time.Duration
|
||||
}
|
||||
|
||||
func (l *Latencies) Add(d time.Duration) { l.d = append(l.d, d) }
|
||||
func (l *Latencies) Len() int { return len(l.d) }
|
||||
|
||||
// Merge folds another accumulator's samples in (for per-worker collection).
|
||||
func (l *Latencies) Merge(o *Latencies) { l.d = append(l.d, o.d...) }
|
||||
|
||||
// Snapshot sorts and returns p50, p99, and max. Sorts in place; call once at end.
|
||||
func (l *Latencies) Snapshot() (p50, p99, max time.Duration) {
|
||||
sort.Slice(l.d, func(i, j int) bool { return l.d[i] < l.d[j] })
|
||||
return Percentile(l.d, 0.50), Percentile(l.d, 0.99), Percentile(l.d, 1)
|
||||
}
|
||||
|
||||
// ErrClass names which side of the connection gave out. It turns the body-drain
|
||||
// "which error string means what" table into something the client can count.
|
||||
type ErrClass int
|
||||
|
||||
const (
|
||||
ErrNone ErrClass = iota // no error
|
||||
ErrClientFdLimit // this process ran out of fds (dial: too many open files)
|
||||
ErrClientPortLimit // this process ran out of ephemeral ports (can't assign requested address)
|
||||
ErrServerCantAccept // server can't accept (out of fds / backlog full) — we time out
|
||||
ErrTimeout // request/response timed out (ambiguous, but usually the server)
|
||||
ErrConnReset // peer reset/closed the connection
|
||||
ErrRefused // nothing listening / server gone
|
||||
ErrOther
|
||||
)
|
||||
|
||||
func (c ErrClass) String() string {
|
||||
switch c {
|
||||
case ErrNone:
|
||||
return "none"
|
||||
case ErrClientFdLimit:
|
||||
return "client-fd-limit"
|
||||
case ErrClientPortLimit:
|
||||
return "client-port-limit"
|
||||
case ErrServerCantAccept:
|
||||
return "server-cant-accept"
|
||||
case ErrTimeout:
|
||||
return "timeout"
|
||||
case ErrConnReset:
|
||||
return "conn-reset"
|
||||
case ErrRefused:
|
||||
return "conn-refused"
|
||||
default:
|
||||
return "other"
|
||||
}
|
||||
}
|
||||
|
||||
// Classify maps a net/http client error to the side that most likely caused it.
|
||||
// It leans on error strings because the kernel/runtime messages are the only
|
||||
// distinguishing signal for several of these; the checks are ordered most- to
|
||||
// least-specific so, e.g., "too many open files" wins over the generic timeout.
|
||||
func Classify(err error) ErrClass {
|
||||
if err == nil {
|
||||
return ErrNone
|
||||
}
|
||||
msg := strings.ToLower(err.Error())
|
||||
|
||||
switch {
|
||||
case strings.Contains(msg, "too many open files"):
|
||||
// A dial failing this way is THIS process out of fds. (A server out of
|
||||
// fds fails on accept and never produces this string on the client.)
|
||||
return ErrClientFdLimit
|
||||
case strings.Contains(msg, "assign requested address"):
|
||||
// EADDRNOTAVAIL: no ephemeral port left for a fresh 4-tuple to this
|
||||
// server. A client-side wall, often hit before the fd limit.
|
||||
return ErrClientPortLimit
|
||||
case strings.Contains(msg, "connection refused"):
|
||||
return ErrRefused
|
||||
case strings.Contains(msg, "connection reset") || strings.Contains(msg, "broken pipe") ||
|
||||
strings.Contains(msg, "eof"):
|
||||
return ErrConnReset
|
||||
}
|
||||
|
||||
// "awaiting response headers" / a dial timeout usually means the server
|
||||
// accepted us into its backlog but can't service (or can't accept at all),
|
||||
// so we sit waiting. Treat a timeout as a server-can't-keep-up signal, but
|
||||
// keep a generic ErrTimeout for the plain deadline case.
|
||||
var ne net.Error
|
||||
if errors.As(err, &ne) && ne.Timeout() {
|
||||
if strings.Contains(msg, "awaiting response headers") {
|
||||
return ErrServerCantAccept
|
||||
}
|
||||
return ErrTimeout
|
||||
}
|
||||
if strings.Contains(msg, "timeout") || strings.Contains(msg, "deadline exceeded") {
|
||||
return ErrTimeout
|
||||
}
|
||||
return ErrOther
|
||||
}
|
||||
@@ -0,0 +1,147 @@
|
||||
package loadlib
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestBurnDeterministic(t *testing.T) {
|
||||
// Same difficulty must give the same checksum: the /compute endpoint returns
|
||||
// it, and a flaky checksum would mean a non-deterministic loop.
|
||||
for _, d := range []int{1, 3, 10} {
|
||||
if a, b := Burn(d), Burn(d); a != b {
|
||||
t.Errorf("Burn(%d) not deterministic: %d != %d", d, a, b)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestBurnDifficultyChangesResult(t *testing.T) {
|
||||
// Different difficulties should not collide (they run different iteration
|
||||
// counts); a collision here would suggest the loop body ignores i.
|
||||
if Burn(1) == Burn(2) {
|
||||
t.Error("Burn(1) == Burn(2): difficulty has no effect on the checksum")
|
||||
}
|
||||
}
|
||||
|
||||
func TestIterationsLinearAndFloored(t *testing.T) {
|
||||
if got := Iterations(2); got != 2*burnIters {
|
||||
t.Errorf("Iterations(2) = %d, want %d", got, 2*burnIters)
|
||||
}
|
||||
// difficulty < 1 is clamped to 1 so /compute?difficulty=0 still does work.
|
||||
for _, d := range []int{0, -5} {
|
||||
if got := Iterations(d); got != burnIters {
|
||||
t.Errorf("Iterations(%d) = %d, want %d (floor of 1)", d, got, burnIters)
|
||||
}
|
||||
}
|
||||
// Cost is monotonic in difficulty.
|
||||
if Iterations(3) <= Iterations(2) {
|
||||
t.Error("Iterations not increasing with difficulty")
|
||||
}
|
||||
}
|
||||
|
||||
func TestPercentileEdges(t *testing.T) {
|
||||
if got := Percentile(nil, 0.5); got != 0 {
|
||||
t.Errorf("Percentile(empty) = %v, want 0", got)
|
||||
}
|
||||
s := []time.Duration{10, 20, 30, 40, 50}
|
||||
cases := []struct {
|
||||
p float64
|
||||
want time.Duration
|
||||
}{
|
||||
{-1, 10}, // clamped to first
|
||||
{0, 10}, // first
|
||||
{0.5, 30}, // middle
|
||||
{1, 50}, // last
|
||||
{2, 50}, // clamped to last
|
||||
}
|
||||
for _, c := range cases {
|
||||
if got := Percentile(s, c.p); got != c.want {
|
||||
t.Errorf("Percentile(%v, %v) = %v, want %v", s, c.p, got, c.want)
|
||||
}
|
||||
}
|
||||
if got := Percentile([]time.Duration{99}, 0.9); got != 99 {
|
||||
t.Errorf("Percentile(single) = %v, want 99", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestLatenciesSnapshot(t *testing.T) {
|
||||
var l Latencies
|
||||
// Add out of order; Snapshot must sort before reporting.
|
||||
for _, d := range []time.Duration{50, 10, 30, 40, 20} {
|
||||
l.Add(d * time.Millisecond)
|
||||
}
|
||||
if l.Len() != 5 {
|
||||
t.Fatalf("Len = %d, want 5", l.Len())
|
||||
}
|
||||
p50, p99, max := l.Snapshot()
|
||||
if p50 != 30*time.Millisecond {
|
||||
t.Errorf("p50 = %v, want 30ms", p50)
|
||||
}
|
||||
if max != 50*time.Millisecond {
|
||||
t.Errorf("max = %v, want 50ms", max)
|
||||
}
|
||||
// Nearest-rank: index int(4*0.99)=3 → 40ms for this 5-sample set.
|
||||
if p99 != 40*time.Millisecond {
|
||||
t.Errorf("p99 = %v, want 40ms", p99)
|
||||
}
|
||||
}
|
||||
|
||||
func TestLatenciesMerge(t *testing.T) {
|
||||
var a, b Latencies
|
||||
a.Add(10 * time.Millisecond)
|
||||
b.Add(20 * time.Millisecond)
|
||||
a.Merge(&b)
|
||||
if a.Len() != 2 {
|
||||
t.Fatalf("after merge Len = %d, want 2", a.Len())
|
||||
}
|
||||
}
|
||||
|
||||
func TestClassify(t *testing.T) {
|
||||
timeoutErr := &net.OpError{
|
||||
Op: "dial",
|
||||
Err: timeoutError{},
|
||||
}
|
||||
headerTimeout := fmt.Errorf("net/http: timeout awaiting response headers")
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
err error
|
||||
want ErrClass
|
||||
}{
|
||||
{"nil", nil, ErrNone},
|
||||
{"client fds", errors.New("dial tcp 127.0.0.1:8098: socket: too many open files"), ErrClientFdLimit},
|
||||
{"client ports", errors.New("dial tcp 127.0.0.1:8098: connect: can't assign requested address"), ErrClientPortLimit},
|
||||
{"refused", errors.New("dial tcp 127.0.0.1:8098: connect: connection refused"), ErrRefused},
|
||||
{"reset", errors.New("read tcp: connection reset by peer"), ErrConnReset},
|
||||
{"eof", errors.New("unexpected EOF"), ErrConnReset},
|
||||
{"plain net timeout", timeoutErr, ErrTimeout},
|
||||
{"server cant accept", headerTimeoutNetErr{}, ErrServerCantAccept},
|
||||
{"header timeout string", headerTimeout, ErrTimeout}, // not a net.Error, but has "timeout"
|
||||
{"deadline", errors.New("context deadline exceeded"), ErrTimeout},
|
||||
{"other", errors.New("something weird"), ErrOther},
|
||||
}
|
||||
for _, c := range cases {
|
||||
if got := Classify(c.err); got != c.want {
|
||||
t.Errorf("%s: Classify(%v) = %v, want %v", c.name, c.err, got, c.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// timeoutError is a minimal net.Error whose Timeout() reports true, used to build
|
||||
// a synthetic *net.OpError the way the runtime would on a real dial timeout.
|
||||
type timeoutError struct{}
|
||||
|
||||
func (timeoutError) Error() string { return "i/o timeout" }
|
||||
func (timeoutError) Timeout() bool { return true }
|
||||
func (timeoutError) Temporary() bool { return true }
|
||||
|
||||
// headerTimeoutNetErr is a net.Error whose message is the exact one net/http
|
||||
// produces when ResponseHeaderTimeout fires — the "server can't accept" signal.
|
||||
type headerTimeoutNetErr struct{}
|
||||
|
||||
func (headerTimeoutNetErr) Error() string { return "net/http: timeout awaiting response headers" }
|
||||
func (headerTimeoutNetErr) Timeout() bool { return true }
|
||||
func (headerTimeoutNetErr) Temporary() bool { return true }
|
||||
@@ -0,0 +1,219 @@
|
||||
// Experiment subject — a deliberately simple HTTP server whose only job is to
|
||||
// be measured. How many connections can it hold, and what breaks first?
|
||||
//
|
||||
// Endpoints:
|
||||
//
|
||||
// GET /static fixed tiny body, no per-request work (the baseline)
|
||||
// GET /compute?difficulty=N burns CPU ∝ N, returns the checksum (CPU-bound probe)
|
||||
// GET /sleep?d=500ms sleeps d, honoring client disconnect (in-flight holder)
|
||||
// GET /healthz liveness
|
||||
// GET /stats JSON: accepted_total, active_conns, in_flight, goroutines, heap_bytes
|
||||
//
|
||||
// Observability. We separate three numbers that are easy to conflate:
|
||||
//
|
||||
// accepted_total every TCP conn ever accepted (only grows)
|
||||
// active_conns conns open right now (idle keep-alive OR busy) — via ConnState
|
||||
// in_flight handlers actually running right now — via middleware
|
||||
//
|
||||
// An idle keep-alive conn counts in active_conns but not in_flight; a running
|
||||
// /sleep counts in both. That gap is the whole point of the idle-vs-slow contrast.
|
||||
//
|
||||
// Timeout flags are the naive-vs-hardened knobs. They default to OFF so the
|
||||
// slowloris experiment works out of the box; turn them on to watch the server
|
||||
// reclaim connections:
|
||||
//
|
||||
// -read-header-timeout 3s drop a conn that dawdles sending headers (kills slowloris)
|
||||
// -idle-timeout 30s drop a keep-alive conn that goes quiet
|
||||
// -read-timeout / -write-timeout whole-request deadlines
|
||||
//
|
||||
// Run (terminal 1), then drive it with ./client (terminal 2):
|
||||
//
|
||||
// go run ./98-http-server-experiments/server
|
||||
// go run ./98-http-server-experiments/server -read-header-timeout 3s -idle-timeout 30s # hardened
|
||||
//
|
||||
// Give the server its own fd budget so YOU decide when it runs out (not the OS
|
||||
// default of a few hundred / a few thousand). bash sets both soft and hard:
|
||||
//
|
||||
// ulimit -Hn 4096; ulimit -Sn 4096; go run ./98-http-server-experiments/server
|
||||
//
|
||||
// (Since Go 1.19 the runtime raises the soft RLIMIT_NOFILE to the hard limit at
|
||||
// startup, so only the hard limit really caps you. zsh's `ulimit -n` sets only
|
||||
// the soft one — use `-Hn`.)
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"flag"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/signal"
|
||||
"runtime"
|
||||
"strconv"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"gosre-exercises/98-http-server-experiments/loadlib"
|
||||
)
|
||||
|
||||
// server is the subject under test: an http.Server plus the three live counters.
|
||||
type server struct {
|
||||
srv *http.Server
|
||||
ln net.Listener
|
||||
acceptedTot atomic.Int64 // every conn ever accepted
|
||||
activeConns atomic.Int64 // conns open right now
|
||||
inFlight atomic.Int64 // handlers running right now
|
||||
}
|
||||
|
||||
func (s *server) handleStatic(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "text/plain")
|
||||
fmt.Fprint(w, "ok\n")
|
||||
}
|
||||
|
||||
// handleCompute burns CPU proportional to ?difficulty= and writes the checksum
|
||||
// back. Writing it matters: it forces the burn to actually happen (the compiler
|
||||
// can't prove the result is unused) and lets the client confirm the work ran.
|
||||
func (s *server) handleCompute(w http.ResponseWriter, r *http.Request) {
|
||||
difficulty := 1
|
||||
if v := r.URL.Query().Get("difficulty"); v != "" {
|
||||
if n, err := strconv.Atoi(v); err == nil {
|
||||
difficulty = n
|
||||
}
|
||||
}
|
||||
sum := loadlib.Burn(difficulty)
|
||||
w.Header().Set("Content-Type", "text/plain")
|
||||
fmt.Fprintf(w, "difficulty=%d checksum=%d\n", difficulty, sum)
|
||||
}
|
||||
|
||||
// handleSleep keeps the request in flight for ?d= but returns early if the
|
||||
// client disconnects or a server timeout fires — otherwise a client that hangs
|
||||
// up would leave the goroutine sleeping to full term, hiding the reclaim.
|
||||
func (s *server) handleSleep(w http.ResponseWriter, r *http.Request) {
|
||||
d := 500 * time.Millisecond
|
||||
if v := r.URL.Query().Get("d"); v != "" {
|
||||
if parsed, err := time.ParseDuration(v); err == nil {
|
||||
d = parsed
|
||||
}
|
||||
}
|
||||
select {
|
||||
case <-time.After(d):
|
||||
w.Header().Set("Content-Type", "text/plain")
|
||||
fmt.Fprintf(w, "slept %v\n", d)
|
||||
case <-r.Context().Done():
|
||||
// Client went away or a timeout fired; nothing to write.
|
||||
}
|
||||
}
|
||||
|
||||
type statsSnapshot struct {
|
||||
AcceptedTotal int64 `json:"accepted_total"`
|
||||
ActiveConns int64 `json:"active_conns"`
|
||||
InFlight int64 `json:"in_flight"`
|
||||
Goroutines int `json:"goroutines"`
|
||||
HeapBytes uint64 `json:"heap_bytes"`
|
||||
}
|
||||
|
||||
func (s *server) snapshot() statsSnapshot {
|
||||
var m runtime.MemStats
|
||||
runtime.ReadMemStats(&m)
|
||||
return statsSnapshot{
|
||||
AcceptedTotal: s.acceptedTot.Load(),
|
||||
ActiveConns: s.activeConns.Load(),
|
||||
InFlight: s.inFlight.Load(),
|
||||
Goroutines: runtime.NumGoroutine(),
|
||||
HeapBytes: m.HeapInuse,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *server) handleStats(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(s.snapshot())
|
||||
}
|
||||
|
||||
// countInFlight wraps a handler so in_flight tracks handlers actually running,
|
||||
// as distinct from conns merely open (which ConnState already counts).
|
||||
func (s *server) countInFlight(next http.Handler) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
s.inFlight.Add(1)
|
||||
defer s.inFlight.Add(-1)
|
||||
next.ServeHTTP(w, r)
|
||||
})
|
||||
}
|
||||
|
||||
func main() {
|
||||
addr := flag.String("addr", "127.0.0.1:8098", "address to listen on")
|
||||
readHeaderTimeout := flag.Duration("read-header-timeout", 0, "Server.ReadHeaderTimeout (0 = off; set 3s to defeat slowloris)")
|
||||
readTimeout := flag.Duration("read-timeout", 0, "Server.ReadTimeout (0 = off; whole-request read deadline)")
|
||||
writeTimeout := flag.Duration("write-timeout", 0, "Server.WriteTimeout (0 = off)")
|
||||
idleTimeout := flag.Duration("idle-timeout", 0, "Server.IdleTimeout (0 = off; set 30s to reclaim quiet keep-alives)")
|
||||
maxHeaderBytes := flag.Int("max-header-bytes", http.DefaultMaxHeaderBytes, "Server.MaxHeaderBytes (1<<20 default)")
|
||||
flag.Parse()
|
||||
|
||||
// NOTE: no -max-conns. Capping concurrent connections in the stdlib means
|
||||
// wrapping the Listener (x/net/netutil.LimitListener) or a semaphore in the
|
||||
// handler; we keep this zero-dependency and instead observe the ceiling the
|
||||
// OS imposes (fds/backlog). See the README for the stdlib-only pattern.
|
||||
|
||||
ln, err := net.Listen("tcp", *addr)
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, "listen:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
s := &server{ln: ln}
|
||||
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("/static", s.handleStatic)
|
||||
mux.HandleFunc("/compute", s.handleCompute)
|
||||
mux.HandleFunc("/sleep", s.handleSleep)
|
||||
mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { fmt.Fprintln(w, "ok") })
|
||||
mux.HandleFunc("/stats", s.handleStats)
|
||||
|
||||
s.srv = &http.Server{
|
||||
Handler: s.countInFlight(mux),
|
||||
ReadHeaderTimeout: *readHeaderTimeout,
|
||||
ReadTimeout: *readTimeout,
|
||||
WriteTimeout: *writeTimeout,
|
||||
IdleTimeout: *idleTimeout,
|
||||
MaxHeaderBytes: *maxHeaderBytes,
|
||||
ConnState: func(c net.Conn, st http.ConnState) {
|
||||
switch st {
|
||||
case http.StateNew:
|
||||
s.acceptedTot.Add(1)
|
||||
s.activeConns.Add(1)
|
||||
case http.StateClosed, http.StateHijacked:
|
||||
s.activeConns.Add(-1)
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
go func() {
|
||||
if err := s.srv.Serve(ln); err != nil && err != http.ErrServerClosed {
|
||||
fmt.Fprintln(os.Stderr, "serve:", err)
|
||||
}
|
||||
}()
|
||||
|
||||
fmt.Printf("serving on http://%s (timeouts: readHeader=%v read=%v write=%v idle=%v)\n",
|
||||
ln.Addr(), *readHeaderTimeout, *readTimeout, *writeTimeout, *idleTimeout)
|
||||
fmt.Println("point the client at it: go run ./98-http-server-experiments/client -mode idle -n 2000")
|
||||
|
||||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
|
||||
defer stop()
|
||||
tick := time.NewTicker(time.Second)
|
||||
defer tick.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
fmt.Println("\nshutting down")
|
||||
shutCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||||
defer cancel()
|
||||
s.srv.Shutdown(shutCtx)
|
||||
return
|
||||
case <-tick.C:
|
||||
snap := s.snapshot()
|
||||
fmt.Printf("%s accepted_total=%d active=%d in_flight=%d goroutines=%d heap=%dMiB\n",
|
||||
time.Now().Format("15:04:05"), snap.AcceptedTotal, snap.ActiveConns,
|
||||
snap.InFlight, snap.Goroutines, snap.HeapBytes>>20)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -6,9 +6,73 @@ properly drain+close response bodies so connections are reused").
|
||||
```
|
||||
go run ./99-experiments/body-drain -mode all
|
||||
go run ./99-experiments/body-drain -mode leak -n 200
|
||||
ulimit -n 64 && go run ./99-experiments/body-drain -mode leak -n 200 # watch it die
|
||||
```
|
||||
|
||||
## 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
|
||||
request costs *two* fds (client socket + accepted server socket). The server
|
||||
usually loses first — `accept` fails with `too many open files`, its accept
|
||||
loop retries with backoff forever, and the client silently hangs waiting for
|
||||
headers on a connection nobody will ever accept. Realistic, but confusing. So
|
||||
split them:
|
||||
|
||||
```
|
||||
# terminal 1
|
||||
go run ./99-experiments/body-drain -role server
|
||||
|
||||
# terminal 2 — client is the one out of fds
|
||||
ulimit -Hn 64; ulimit -Sn 64
|
||||
go run ./99-experiments/body-drain -role client -mode leak -n 200
|
||||
# request 60 failed: dial tcp 127.0.0.1:8099: socket: too many open files
|
||||
```
|
||||
|
||||
Now swap: limit the *server* terminal instead. It logs
|
||||
`http: Accept error: accept tcp ...: too many open files; retrying in 5ms`,
|
||||
while the client (unlimited) either hangs (Linux — pass `-header-timeout 3s`
|
||||
to turn that into `timeout awaiting response headers`) or gets
|
||||
`read: connection reset by peer` (macOS).
|
||||
|
||||
Which error string you see tells you which side is exhausted:
|
||||
|
||||
| error | who ran out |
|
||||
|---|---|
|
||||
| `dial tcp ...: socket: too many open files` | the client (this process) |
|
||||
| `http: Accept error: accept ...: too many open files` | the server |
|
||||
| `timeout awaiting response headers` / `connection reset by peer` | the client is fine; the *server* can't accept |
|
||||
|
||||
Note on `ulimit`: since Go 1.19 the runtime raises the soft `RLIMIT_NOFILE`
|
||||
to the hard limit at startup, so lowering only the soft limit does nothing.
|
||||
bash's plain `ulimit -n 64` sets both; zsh's sets only the soft one, hence
|
||||
the explicit `-Hn`.
|
||||
|
||||
## What it does
|
||||
|
||||
A local `httptest.Server` returns a 1 MiB body and counts accepted TCP
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
// Experiment — what happens when you don't drain/close an HTTP response body
|
||||
//
|
||||
// A local server hands out 1 MiB responses and counts every TCP connection it
|
||||
// A server hands out 1 MiB responses and counts every TCP connection it
|
||||
// accepts. The client performs N sequential GETs against it in one of three
|
||||
// modes and we observe:
|
||||
//
|
||||
@@ -15,26 +15,50 @@
|
||||
// close — body.Close() only → connection thrown away, new dial every time
|
||||
// leak — neither → connection AND 2 goroutines AND 1 fd leaked per request
|
||||
//
|
||||
// Run: go run ./99-experiments/body-drain -mode drain
|
||||
// go run ./99-experiments/body-drain -mode close
|
||||
// go run ./99-experiments/body-drain -mode leak
|
||||
// go run ./99-experiments/body-drain -mode all (all three, fresh transport each)
|
||||
// Roles (so you can watch each side fail on its own, with its own fd limit):
|
||||
//
|
||||
// To see the leak actually kill the process, lower the fd limit first:
|
||||
// ulimit -n 64 && go run ./99-experiments/body-drain -mode leak -n 200
|
||||
// both — server + client in one process (default)
|
||||
// server — only serve on -addr, print connection stats every second, Ctrl-C to stop
|
||||
// client — only run the client against -url
|
||||
//
|
||||
// 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):
|
||||
//
|
||||
// go run ./99-experiments/body-drain -role server
|
||||
// ulimit -Hn 64; ulimit -Sn 64; go run ./99-experiments/body-drain -role client -mode leak -n 200
|
||||
//
|
||||
// (Lower the HARD limit too: since Go 1.19 the runtime raises the soft
|
||||
// RLIMIT_NOFILE to the hard limit at startup. bash's `ulimit -n 64` sets both;
|
||||
// zsh's sets only the soft one.)
|
||||
//
|
||||
// With the client alone at 64 fds you'll see the CLIENT die:
|
||||
//
|
||||
// dial tcp 127.0.0.1:8099: socket: too many open files
|
||||
//
|
||||
// With the limit on the SERVER terminal instead, the SERVER logs
|
||||
// "accept: too many open files" and the client either hangs waiting for
|
||||
// headers (Linux — use -header-timeout 3s) or gets "connection reset by
|
||||
// peer" (macOS).
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"flag"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/http/httptrace"
|
||||
"os"
|
||||
"os/signal"
|
||||
"runtime"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
)
|
||||
@@ -49,70 +73,156 @@ const (
|
||||
modeLeak mode = "leak"
|
||||
)
|
||||
|
||||
// countingServer wraps httptest.Server and counts accepted connections.
|
||||
// ---- server ----
|
||||
|
||||
// countingServer is an http.Server that counts accepted connections.
|
||||
type countingServer struct {
|
||||
*httptest.Server
|
||||
srv *http.Server
|
||||
ln net.Listener
|
||||
newConns atomic.Int64
|
||||
activeConns atomic.Int64
|
||||
}
|
||||
|
||||
func newCountingServer() *countingServer {
|
||||
payload := make([]byte, bodySize)
|
||||
s := &countingServer{}
|
||||
s.Server = httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Length", fmt.Sprint(bodySize))
|
||||
w.WriteHeader(http.StatusOK)
|
||||
w.Write(payload)
|
||||
}))
|
||||
// ConnState is the server-side hook that lets us see the connection lifecycle.
|
||||
s.Config.ConnState = func(c net.Conn, st http.ConnState) {
|
||||
switch st {
|
||||
case http.StateNew:
|
||||
s.newConns.Add(1)
|
||||
s.activeConns.Add(1)
|
||||
case http.StateClosed, http.StateHijacked:
|
||||
s.activeConns.Add(-1)
|
||||
}
|
||||
func (s *countingServer) URL() string { return "http://" + s.ln.Addr().String() }
|
||||
|
||||
// newCountingServer listens on addr ("127.0.0.1:0" = random port) and starts serving.
|
||||
func newCountingServer(addr string) (*countingServer, error) {
|
||||
ln, err := net.Listen("tcp", addr)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
s.Start()
|
||||
return s
|
||||
payload := make([]byte, bodySize)
|
||||
s := &countingServer{ln: ln}
|
||||
s.srv = &http.Server{
|
||||
Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Length", fmt.Sprint(bodySize))
|
||||
w.WriteHeader(http.StatusOK)
|
||||
w.Write(payload)
|
||||
}),
|
||||
// ConnState is the server-side hook that lets us see the connection lifecycle.
|
||||
ConnState: func(c net.Conn, st http.ConnState) {
|
||||
switch st {
|
||||
case http.StateNew:
|
||||
s.newConns.Add(1)
|
||||
s.activeConns.Add(1)
|
||||
case http.StateClosed, http.StateHijacked:
|
||||
s.activeConns.Add(-1)
|
||||
}
|
||||
},
|
||||
}
|
||||
go s.srv.Serve(ln)
|
||||
return s, nil
|
||||
}
|
||||
|
||||
// run performs n GETs in the given mode and reports what it observed.
|
||||
func run(m mode, n int, url string) {
|
||||
// close force-closes everything. In leak mode the handlers are STILL BLOCKED
|
||||
// in Write(): the kernel socket buffers filled up because nobody on the
|
||||
// client side is reading the 1 MiB body. A graceful Shutdown would wait for
|
||||
// them forever — that's the server-side symptom of a client leak: goroutines
|
||||
// and memory pinned per stuck connection. So we use Close(), not Shutdown().
|
||||
func (s *countingServer) close() { s.srv.Close() }
|
||||
|
||||
func (s *countingServer) report() {
|
||||
fmt.Printf(" server new conns : %d (ideal: 1)\n", s.newConns.Load())
|
||||
fmt.Printf(" server active conns: %d (ideal after cleanup: 0)\n", s.activeConns.Load())
|
||||
}
|
||||
|
||||
// serveForever runs the server role: serve until Ctrl-C, printing stats each second.
|
||||
func serveForever(addr string) {
|
||||
s, err := newCountingServer(addr)
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, "listen:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
fmt.Printf("serving on %s (point the client at it with -url %s)\n", s.URL(), s.URL())
|
||||
|
||||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
|
||||
defer stop()
|
||||
tick := time.NewTicker(time.Second)
|
||||
defer tick.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
fmt.Println("\nshutting down")
|
||||
s.close()
|
||||
return
|
||||
case <-tick.C:
|
||||
// accepted_total only ever grows (every conn ever accepted);
|
||||
// active is what's open right now.
|
||||
fmt.Printf("%s accepted_total=%d active=%d goroutines=%d\n",
|
||||
time.Now().Format("15:04:05"), s.newConns.Load(), s.activeConns.Load(), runtime.NumGoroutine())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---- client ----
|
||||
|
||||
// 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 = 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)
|
||||
|
||||
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 {
|
||||
fmt.Printf(" request %d failed: %v\n", i, err)
|
||||
// This is where `leak` eventually dies with "socket: too many open files".
|
||||
break
|
||||
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)
|
||||
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.Add(1)
|
||||
|
||||
switch m {
|
||||
case modeDrain:
|
||||
@@ -125,13 +235,33 @@ func run(m mode, n int, url string) {
|
||||
}
|
||||
}
|
||||
|
||||
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 in %v\n", 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)
|
||||
|
||||
@@ -143,34 +273,60 @@ func run(m mode, n int, url string) {
|
||||
}
|
||||
|
||||
func main() {
|
||||
modeFlag := flag.String("mode", "all", "drain | close | leak | all")
|
||||
n := flag.Int("n", 20, "number of requests per mode")
|
||||
role := flag.String("role", "both", "both | server | client")
|
||||
addr := flag.String("addr", "127.0.0.1:8099", "server: address to listen on")
|
||||
url := flag.String("url", "http://127.0.0.1:8099", "client: server URL to hit")
|
||||
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" {
|
||||
modes = []mode{mode(*modeFlag)}
|
||||
}
|
||||
|
||||
for _, m := range modes {
|
||||
srv := newCountingServer()
|
||||
fmt.Printf("== mode=%s ==\n", m)
|
||||
run(m, *n, srv.URL)
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
fmt.Printf(" server new conns : %d (ideal: 1)\n", srv.newConns.Load())
|
||||
fmt.Printf(" server active conns: %d (ideal after cleanup: 0)\n", srv.activeConns.Load())
|
||||
fmt.Println()
|
||||
// In leak mode the server's handlers are STILL BLOCKED in Write(): the
|
||||
// kernel socket buffers filled up because nobody on the client side is
|
||||
// reading the 1 MiB body. srv.Close() waits for active conns and would
|
||||
// hang forever (it logs "blocked in Close after 5 seconds" and lists
|
||||
// them). That's the server-side symptom of a client leak: goroutines
|
||||
// and memory pinned per stuck connection. Force-close them first.
|
||||
srv.CloseClientConnections()
|
||||
srv.Close()
|
||||
}
|
||||
switch *role {
|
||||
case "server":
|
||||
serveForever(*addr)
|
||||
|
||||
if len(modes) == 1 && modes[0] == modeLeak {
|
||||
fmt.Fprintln(os.Stderr, "tip: rerun with `ulimit -n 64` and -n 200 to watch it hit 'too many open files'")
|
||||
case "client":
|
||||
for _, m := range modes {
|
||||
fmt.Printf("== mode=%s (client only, target %s) ==\n", m, *url)
|
||||
runClient(m, *url, opts)
|
||||
fmt.Println()
|
||||
}
|
||||
if len(modes) == 1 && modes[0] == modeLeak {
|
||||
fmt.Fprintln(os.Stderr, "tip: rerun with `ulimit -n 64` and -n 200 to watch the client hit 'too many open files'")
|
||||
}
|
||||
|
||||
case "both":
|
||||
for _, m := range modes {
|
||||
srv, err := newCountingServer("127.0.0.1:0")
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, "listen:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
fmt.Printf("== mode=%s ==\n", m)
|
||||
runClient(m, srv.URL(), opts)
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
srv.report()
|
||||
fmt.Println()
|
||||
srv.close()
|
||||
}
|
||||
if len(modes) == 1 && modes[0] == modeLeak {
|
||||
fmt.Fprintln(os.Stderr, "tip: in one process both sides share the fd limit and the SERVER usually hits it first (accept4 error) while the client hangs. Use -role server / -role client in two terminals to see each side fail alone.")
|
||||
}
|
||||
|
||||
default:
|
||||
fmt.Fprintf(os.Stderr, "unknown -role %q\n", *role)
|
||||
os.Exit(2)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,118 @@
|
||||
# Experiment — a global rate limiter as a service
|
||||
|
||||
Exercise 03 built an in-process token bucket. That limits **one process**.
|
||||
When N replicas of a service must share one quota ("tenant-a gets 10 req/s,
|
||||
total"), the bucket has to live somewhere they can all reach. This is the
|
||||
smallest honest version of that: an HTTP service holding per-key buckets, and
|
||||
a demo client that plays one replica.
|
||||
|
||||
```
|
||||
99-experiments/global-ratelimit/
|
||||
limiter/ token bucket (ex03) + sharded keyed store (ex09) + idle eviction go test -race
|
||||
server/ HTTP service: /allow, /stats, /healthz
|
||||
client/ demo replica: asks /allow before each unit of work
|
||||
```
|
||||
|
||||
## Run it
|
||||
|
||||
```
|
||||
# terminal 1 — the limiter
|
||||
go run ./99-experiments/global-ratelimit/server -rate 10 -burst 10
|
||||
|
||||
# quick check
|
||||
for i in $(seq 12); do curl -s -o /dev/null -w '%{http_code} ' 'localhost:8100/allow?key=t'; done
|
||||
# 200 200 200 200 200 200 200 200 200 200 429 429
|
||||
curl -si 'localhost:8100/allow?key=t' | grep -iE 'HTTP/|retry|ratelimit'
|
||||
|
||||
# terminals 2–4 — three replicas, each WANTS 20/s, all on key tenant-a
|
||||
go run ./99-experiments/global-ratelimit/client -id inst-1 -rps 20
|
||||
go run ./99-experiments/global-ratelimit/client -id inst-2 -rps 20
|
||||
go run ./99-experiments/global-ratelimit/client -id inst-3 -rps 20
|
||||
|
||||
curl -s localhost:8100/stats
|
||||
```
|
||||
|
||||
Observed (3 × 20 rps attempted, limit 10/s, 8 s):
|
||||
|
||||
```
|
||||
inst-1 summary: attempts=159 allowed=31 denied=127 → work done=32
|
||||
inst-2 summary: attempts=159 allowed=28 denied=130 → work done=29
|
||||
inst-3 summary: attempts=160 allowed=29 denied=130 → work done=30
|
||||
───
|
||||
88 ≈ 10/s × 8 s + burst 10
|
||||
inst-1 /allow latency: p50=0.6ms p99=11ms
|
||||
```
|
||||
|
||||
The **sum** stays at the limit, not each instance — that's the whole point of
|
||||
centralising it. Each replica gets roughly a third, with no coordination
|
||||
between replicas and no per-replica configuration.
|
||||
|
||||
## API
|
||||
|
||||
```
|
||||
POST|GET /allow?key=K[&n=1]
|
||||
200 {"allowed":true,"remaining":9,"limit":10,"rate":10}
|
||||
429 {"allowed":false,"remaining":0.3,"limit":10,"rate":10,"retry_after_ms":70}
|
||||
headers: X-RateLimit-Limit, X-RateLimit-Remaining, Retry-After (429; whole seconds, rounded up)
|
||||
400 missing key / bad n
|
||||
GET /stats {"keys":2,"allowed_total":99,"denied_total":391}
|
||||
GET /healthz ok
|
||||
```
|
||||
|
||||
Server flags: `-addr`, `-rate`/`-burst` (defaults for unknown keys),
|
||||
`-limits "tenant-a=100:200,tenant-b=5:5"` (per-key overrides), `-sweep 1m`.
|
||||
|
||||
Client flags: `-id`, `-limiter`, `-key`, `-rps`, `-duration`, `-timeout 200ms`,
|
||||
`-fail open|closed`.
|
||||
|
||||
## What to observe
|
||||
|
||||
**The hop.** Every decision is an HTTP round trip: p50 ≈ 0.6 ms on loopback,
|
||||
1–2 ms across an AZ. The in-process limiter is ~100 ns. That's the price of
|
||||
"global", and why you only pay it at the edge (per-tenant quota), not in inner
|
||||
loops.
|
||||
|
||||
**Fail-open vs fail-closed.** Start a client with each mode, then Ctrl-C the
|
||||
server mid-run:
|
||||
|
||||
```
|
||||
open allowed=0 denied=0 errors=10 → work done keeps going (49 of 59)
|
||||
closed allowed=0 denied=0 errors=10 → work stops (18 of 59)
|
||||
```
|
||||
|
||||
The limiter is now a dependency on the request path. Fail-open protects your
|
||||
users and risks overloading what the limiter guards; fail-closed protects the
|
||||
backend and turns a limiter outage into *your* outage. Pick deliberately, per
|
||||
call site, and alert on the error rate either way. `-timeout` matters for the
|
||||
same reason: a *slow* limiter must not make you slow.
|
||||
|
||||
**Idle eviction.** `-sweep` removes buckets untouched for that long. An idle
|
||||
bucket is full anyway (once `idle ≥ burst/rate`), so recreating it later is
|
||||
indistinguishable from keeping it — `last` doubles as the idle timestamp,
|
||||
nothing extra to track. Without this, one bucket per API key ever seen lives
|
||||
forever.
|
||||
|
||||
**Burst semantics across replicas.** The first second admits ~20 (burst 10 +
|
||||
10/s). Replicas that start first eat the burst. If that's unfair for your use
|
||||
case, lower `burst` — global buckets make the burst global too.
|
||||
|
||||
## What this is not
|
||||
|
||||
- **Not HA.** One process, in memory. Restart = all buckets full again
|
||||
(a free burst for everyone) and, for the duration, fail-open or fail-closed
|
||||
as configured. Real deployments put the bucket in Redis and run the
|
||||
refill+consume as one Lua script so it's atomic — same arithmetic as
|
||||
`limiter.Bucket.AllowN`, different storage.
|
||||
- **Not accurate under Redis Cluster / multiple limiter nodes** unless keys are
|
||||
routed consistently to one node. Keys → node is what makes a "global" bucket
|
||||
global.
|
||||
- **No local tier.** Production usually also keeps a coarse in-process limiter
|
||||
(ex03) as a pre-filter, so a flooding client is cut off before it costs a
|
||||
network hop per request, and caches "denied until T" locally.
|
||||
|
||||
## References
|
||||
|
||||
- Stripe, "Scaling your API with rate limiters" — the token bucket + Redis
|
||||
pattern and why to shed load with it
|
||||
- Envoy `ratelimit` service / `local_ratelimit` filter — the two-tier design
|
||||
- `golang.org/x/time/rate` — the in-process reference implementation
|
||||
@@ -0,0 +1,144 @@
|
||||
// Demo downstream service replica that consults the global rate limiter
|
||||
// before each unit of work.
|
||||
//
|
||||
// Run several in parallel with different -id against one server and watch
|
||||
// the SUM of their allowed rates stay at the key's limit:
|
||||
//
|
||||
// go run ./99-experiments/global-ratelimit/client -id inst-1 -rps 20
|
||||
// go run ./99-experiments/global-ratelimit/client -id inst-2 -rps 20
|
||||
// go run ./99-experiments/global-ratelimit/client -id inst-3 -rps 20
|
||||
//
|
||||
// Then Ctrl-C the server mid-run and compare -fail open vs -fail closed.
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"flag"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
"os/signal"
|
||||
"sort"
|
||||
"time"
|
||||
)
|
||||
|
||||
type allowResponse struct {
|
||||
Allowed bool `json:"allowed"`
|
||||
RetryAfterMS int64 `json:"retry_after_ms"`
|
||||
}
|
||||
|
||||
type outcome int
|
||||
|
||||
const (
|
||||
allowed outcome = iota
|
||||
denied
|
||||
errored // limiter unreachable / timed out / bad response
|
||||
)
|
||||
|
||||
// ask performs one /allow call. It is the ONLY place that talks to the limiter.
|
||||
func ask(ctx context.Context, client *http.Client, endpoint string) (outcome, time.Duration) {
|
||||
start := time.Now()
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, nil)
|
||||
if err != nil {
|
||||
return errored, time.Since(start)
|
||||
}
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
return errored, time.Since(start)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
// Decode fully (which also drains the body → connection reuse).
|
||||
var ar allowResponse
|
||||
if err := json.NewDecoder(resp.Body).Decode(&ar); err != nil {
|
||||
io.Copy(io.Discard, resp.Body)
|
||||
return errored, time.Since(start)
|
||||
}
|
||||
switch resp.StatusCode {
|
||||
case http.StatusOK:
|
||||
return allowed, time.Since(start)
|
||||
case http.StatusTooManyRequests:
|
||||
return denied, time.Since(start)
|
||||
default:
|
||||
return errored, time.Since(start)
|
||||
}
|
||||
}
|
||||
|
||||
func percentile(sorted []time.Duration, p float64) time.Duration {
|
||||
if len(sorted) == 0 {
|
||||
return 0
|
||||
}
|
||||
i := int(float64(len(sorted)-1) * p)
|
||||
return sorted[i]
|
||||
}
|
||||
|
||||
func main() {
|
||||
id := flag.String("id", "inst-1", "this replica's name (for output only)")
|
||||
limiterURL := flag.String("limiter", "http://127.0.0.1:8100", "limiter service base URL")
|
||||
key := flag.String("key", "tenant-a", "rate-limit key to consume from")
|
||||
rps := flag.Float64("rps", 20, "how many units of work this replica ATTEMPTS per second")
|
||||
duration := flag.Duration("duration", 30*time.Second, "how long to run")
|
||||
failMode := flag.String("fail", "open", "what to do when the limiter is unreachable: open (proceed) | closed (deny)")
|
||||
timeout := flag.Duration("timeout", 200*time.Millisecond, "per-call timeout to the limiter")
|
||||
flag.Parse()
|
||||
|
||||
if *failMode != "open" && *failMode != "closed" {
|
||||
fmt.Fprintln(os.Stderr, "-fail must be open or closed")
|
||||
os.Exit(2)
|
||||
}
|
||||
|
||||
endpoint := *limiterURL + "/allow?" + url.Values{"key": {*key}}.Encode()
|
||||
|
||||
// One shared client, with a timeout: a limiter that is slow must not make
|
||||
// US slow beyond `timeout`. Keep-alive means one TCP conn for the whole run.
|
||||
client := &http.Client{Timeout: *timeout}
|
||||
|
||||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
|
||||
defer stop()
|
||||
ctx, cancel := context.WithTimeout(ctx, *duration)
|
||||
defer cancel()
|
||||
|
||||
tick := time.NewTicker(time.Duration(float64(time.Second) / *rps))
|
||||
defer tick.Stop()
|
||||
report := time.NewTicker(time.Second)
|
||||
defer report.Stop()
|
||||
|
||||
var sec, total [3]int // per-second and cumulative counts by outcome
|
||||
var latencies []time.Duration
|
||||
workDone := 0 // units of work actually performed (allowed + fail-open errors)
|
||||
|
||||
fmt.Printf("%s: attempting %.0f/s on key %q via %s, fail-%s, timeout %v\n",
|
||||
*id, *rps, *key, *limiterURL, *failMode, *timeout)
|
||||
|
||||
loop:
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
break loop
|
||||
case <-report.C:
|
||||
fmt.Printf("%s allowed=%-3d denied=%-3d errors=%-3d (last 1s)\n", *id, sec[allowed], sec[denied], sec[errored])
|
||||
sec = [3]int{}
|
||||
case <-tick.C:
|
||||
o, lat := ask(ctx, client, endpoint)
|
||||
if ctx.Err() != nil {
|
||||
break loop // aborted by our own shutdown, not a limiter failure
|
||||
}
|
||||
sec[o]++
|
||||
total[o]++
|
||||
latencies = append(latencies, lat)
|
||||
// Do the actual "work" only if allowed — or on error, if fail-open.
|
||||
if o == allowed || (o == errored && *failMode == "open") {
|
||||
workDone++
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
sort.Slice(latencies, func(i, j int) bool { return latencies[i] < latencies[j] })
|
||||
n := total[allowed] + total[denied] + total[errored]
|
||||
fmt.Printf("\n%s summary: attempts=%d allowed=%d denied=%d errors=%d → work done=%d (fail-%s)\n",
|
||||
*id, n, total[allowed], total[denied], total[errored], workDone, *failMode)
|
||||
fmt.Printf("%s /allow latency: p50=%v p99=%v max=%v\n",
|
||||
*id, percentile(latencies, 0.50), percentile(latencies, 0.99), percentile(latencies, 1))
|
||||
}
|
||||
@@ -0,0 +1,185 @@
|
||||
// Package limiter is the token bucket from exercise 03 turned into a keyed
|
||||
// store suitable for a shared rate-limiter service: many buckets addressed by
|
||||
// a string key (tenant, client id, API key), sharded like exercise 09, with
|
||||
// eviction of idle buckets so the map does not grow forever.
|
||||
package limiter
|
||||
|
||||
import (
|
||||
"hash/fnv"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
)
|
||||
|
||||
// ---- Bucket: a single token bucket (exercise 03) ----
|
||||
|
||||
// Bucket is a lazily refilled token bucket. Safe for concurrent use.
|
||||
type Bucket struct {
|
||||
mu sync.Mutex
|
||||
tokens float64
|
||||
max float64
|
||||
rate float64 // tokens per second
|
||||
last time.Time
|
||||
}
|
||||
|
||||
// NewBucket returns a full bucket allowing `rate` tokens/s with bursts up to `burst`.
|
||||
func NewBucket(rate float64, burst int) *Bucket {
|
||||
return &Bucket{tokens: float64(burst), max: float64(burst), rate: rate, last: time.Now()}
|
||||
}
|
||||
|
||||
// Decision is the outcome of an AllowN call.
|
||||
type Decision struct {
|
||||
Allowed bool
|
||||
Remaining float64 // tokens left after this decision
|
||||
Limit int // burst (bucket capacity)
|
||||
Rate float64 // refill rate, tokens/s
|
||||
RetryAfter time.Duration // when denied: how long until n tokens would be available
|
||||
}
|
||||
|
||||
// AllowN tries to take n tokens. Non-blocking.
|
||||
func (b *Bucket) AllowN(n int) Decision {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
|
||||
// 1. lazy refill, capped at max; `last` advances on every call
|
||||
now := time.Now()
|
||||
b.tokens += now.Sub(b.last).Seconds() * b.rate
|
||||
if b.tokens > b.max {
|
||||
b.tokens = b.max
|
||||
}
|
||||
b.last = now
|
||||
|
||||
d := Decision{Limit: int(b.max), Rate: b.rate}
|
||||
// 2. consume or compute how long until we could
|
||||
if b.tokens >= float64(n) {
|
||||
b.tokens -= float64(n)
|
||||
d.Allowed = true
|
||||
} else if b.rate > 0 {
|
||||
d.RetryAfter = time.Duration((float64(n) - b.tokens) / b.rate * float64(time.Second))
|
||||
}
|
||||
d.Remaining = b.tokens
|
||||
return d
|
||||
}
|
||||
|
||||
// idleSince reports the last time the bucket was touched and whether it is
|
||||
// full (i.e. nothing would be lost by forgetting it).
|
||||
func (b *Bucket) idleSince() (time.Time, bool) {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
// A bucket untouched for `max/rate` seconds is full again by definition,
|
||||
// so idle time alone decides; we don't need to refill here.
|
||||
return b.last, true
|
||||
}
|
||||
|
||||
// ---- Store: many buckets by key (exercise 09's sharding) ----
|
||||
|
||||
const shards = 16
|
||||
|
||||
type shard struct {
|
||||
mu sync.Mutex
|
||||
m map[string]*Bucket
|
||||
}
|
||||
|
||||
// Limit is a (rate, burst) pair.
|
||||
type Limit struct {
|
||||
Rate float64
|
||||
Burst int
|
||||
}
|
||||
|
||||
// Store holds one Bucket per key, created lazily with the default Limit or a
|
||||
// per-key override.
|
||||
type Store struct {
|
||||
def Limit
|
||||
overrides map[string]Limit // read-only after construction
|
||||
shards [shards]shard
|
||||
|
||||
allowed atomic.Int64
|
||||
denied atomic.Int64
|
||||
}
|
||||
|
||||
// NewStore creates a store; unknown keys get `def`, keys in `overrides` get theirs.
|
||||
func NewStore(def Limit, overrides map[string]Limit) *Store {
|
||||
s := &Store{def: def, overrides: overrides}
|
||||
for i := range s.shards {
|
||||
s.shards[i].m = make(map[string]*Bucket)
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
func (s *Store) shardFor(key string) *shard {
|
||||
h := fnv.New32a()
|
||||
h.Write([]byte(key))
|
||||
return &s.shards[h.Sum32()%shards]
|
||||
}
|
||||
|
||||
func (s *Store) limitFor(key string) Limit {
|
||||
if l, ok := s.overrides[key]; ok {
|
||||
return l
|
||||
}
|
||||
return s.def
|
||||
}
|
||||
|
||||
// bucket returns the bucket for key, creating it on first sight.
|
||||
func (s *Store) bucket(key string) *Bucket {
|
||||
sh := s.shardFor(key)
|
||||
sh.mu.Lock()
|
||||
defer sh.mu.Unlock()
|
||||
b, ok := sh.m[key]
|
||||
if !ok {
|
||||
l := s.limitFor(key)
|
||||
b = NewBucket(l.Rate, l.Burst)
|
||||
sh.m[key] = b
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
||||
// Allow tries to take n tokens from key's bucket.
|
||||
func (s *Store) Allow(key string, n int) Decision {
|
||||
d := s.bucket(key).AllowN(n)
|
||||
if d.Allowed {
|
||||
s.allowed.Add(1)
|
||||
} else {
|
||||
s.denied.Add(1)
|
||||
}
|
||||
return d
|
||||
}
|
||||
|
||||
// Sweep removes buckets not touched for at least `idle`. Such a bucket has
|
||||
// refilled to full anyway (as long as idle >= burst/rate), so dropping it and
|
||||
// recreating it later is unobservable to clients. Returns how many were removed.
|
||||
func (s *Store) Sweep(idle time.Duration) int {
|
||||
cutoff := time.Now().Add(-idle)
|
||||
removed := 0
|
||||
for i := range s.shards {
|
||||
sh := &s.shards[i]
|
||||
sh.mu.Lock()
|
||||
for k, b := range sh.m {
|
||||
if last, _ := b.idleSince(); last.Before(cutoff) {
|
||||
delete(sh.m, k)
|
||||
removed++
|
||||
}
|
||||
}
|
||||
sh.mu.Unlock()
|
||||
}
|
||||
return removed
|
||||
}
|
||||
|
||||
// Stats is a point-in-time snapshot for observability.
|
||||
type Stats struct {
|
||||
Keys int `json:"keys"`
|
||||
Allowed int64 `json:"allowed_total"`
|
||||
Denied int64 `json:"denied_total"`
|
||||
}
|
||||
|
||||
// Stats returns counters. Keys is counted shard by shard, not atomically
|
||||
// across shards (fine for metrics; see exercise 09 Snapshot discussion).
|
||||
func (s *Store) Stats() Stats {
|
||||
st := Stats{Allowed: s.allowed.Load(), Denied: s.denied.Load()}
|
||||
for i := range s.shards {
|
||||
sh := &s.shards[i]
|
||||
sh.mu.Lock()
|
||||
st.Keys += len(sh.m)
|
||||
sh.mu.Unlock()
|
||||
}
|
||||
return st
|
||||
}
|
||||
@@ -0,0 +1,128 @@
|
||||
package limiter
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestBurstThenDeny(t *testing.T) {
|
||||
s := NewStore(Limit{Rate: 1, Burst: 3}, nil)
|
||||
for i := 0; i < 3; i++ {
|
||||
if !s.Allow("k", 1).Allowed {
|
||||
t.Fatalf("request %d within burst should be allowed", i)
|
||||
}
|
||||
}
|
||||
d := s.Allow("k", 1)
|
||||
if d.Allowed {
|
||||
t.Fatal("4th immediate request should be denied")
|
||||
}
|
||||
if d.RetryAfter <= 0 || d.RetryAfter > time.Second {
|
||||
t.Fatalf("retry-after should be ~1s at rate 1, got %v", d.RetryAfter)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRefill(t *testing.T) {
|
||||
s := NewStore(Limit{Rate: 20, Burst: 1}, nil)
|
||||
if !s.Allow("k", 1).Allowed {
|
||||
t.Fatal("initial token missing")
|
||||
}
|
||||
if s.Allow("k", 1).Allowed {
|
||||
t.Fatal("should be empty immediately after")
|
||||
}
|
||||
time.Sleep(120 * time.Millisecond)
|
||||
if !s.Allow("k", 1).Allowed {
|
||||
t.Fatal("should have refilled")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBurstIsCapped(t *testing.T) {
|
||||
s := NewStore(Limit{Rate: 1000, Burst: 2}, nil)
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
allowed := 0
|
||||
for i := 0; i < 10; i++ {
|
||||
if s.Allow("k", 1).Allowed {
|
||||
allowed++
|
||||
}
|
||||
}
|
||||
if allowed > 2 {
|
||||
t.Fatalf("idle time built credit beyond burst: %d", allowed)
|
||||
}
|
||||
}
|
||||
|
||||
func TestKeysAreIsolatedAndOverridden(t *testing.T) {
|
||||
s := NewStore(Limit{Rate: 0.0001, Burst: 1}, map[string]Limit{"vip": {Rate: 0.0001, Burst: 5}})
|
||||
if !s.Allow("a", 1).Allowed || s.Allow("a", 1).Allowed {
|
||||
t.Fatal("key a should allow exactly one")
|
||||
}
|
||||
if !s.Allow("b", 1).Allowed {
|
||||
t.Fatal("key b must have its own bucket")
|
||||
}
|
||||
for i := 0; i < 5; i++ {
|
||||
if !s.Allow("vip", 1).Allowed {
|
||||
t.Fatalf("vip request %d should be allowed (burst 5)", i)
|
||||
}
|
||||
}
|
||||
if s.Allow("vip", 1).Allowed {
|
||||
t.Fatal("vip 6th should be denied")
|
||||
}
|
||||
if got := s.Stats().Keys; got != 3 {
|
||||
t.Fatalf("expected 3 keys, got %d", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAllowN(t *testing.T) {
|
||||
s := NewStore(Limit{Rate: 0.0001, Burst: 10}, nil)
|
||||
if !s.Allow("k", 7).Allowed {
|
||||
t.Fatal("7 of 10 should be allowed")
|
||||
}
|
||||
if s.Allow("k", 4).Allowed {
|
||||
t.Fatal("4 more (3 left) should be denied")
|
||||
}
|
||||
if !s.Allow("k", 3).Allowed {
|
||||
t.Fatal("exactly 3 left should be allowed")
|
||||
}
|
||||
}
|
||||
|
||||
func TestConcurrentNoOverAdmission(t *testing.T) {
|
||||
s := NewStore(Limit{Rate: 0.0001, Burst: 100}, nil)
|
||||
var allowed atomic.Int64
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < 50; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for j := 0; j < 10; j++ {
|
||||
if s.Allow("shared", 1).Allowed {
|
||||
allowed.Add(1)
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
if got := allowed.Load(); got != 100 {
|
||||
t.Fatalf("expected exactly 100 admissions, got %d", got)
|
||||
}
|
||||
st := s.Stats()
|
||||
if st.Allowed != 100 || st.Denied != 400 {
|
||||
t.Fatalf("stats mismatch: %+v", st)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSweepEvictsIdleOnly(t *testing.T) {
|
||||
s := NewStore(Limit{Rate: 1000, Burst: 1}, nil)
|
||||
s.Allow("old", 1)
|
||||
time.Sleep(30 * time.Millisecond)
|
||||
s.Allow("fresh", 1)
|
||||
if n := s.Sweep(20 * time.Millisecond); n != 1 {
|
||||
t.Fatalf("expected 1 evicted, got %d", n)
|
||||
}
|
||||
if got := s.Stats().Keys; got != 1 {
|
||||
t.Fatalf("expected 1 key left, got %d", got)
|
||||
}
|
||||
// evicted key comes back full, which is what an idle bucket would be anyway
|
||||
if !s.Allow("old", 1).Allowed {
|
||||
t.Fatal("recreated bucket should start full")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,182 @@
|
||||
// Global rate limiter service.
|
||||
//
|
||||
// One process holds per-key token buckets; any number of downstream service
|
||||
// replicas ask it "may I?" before doing work, so the quota is enforced
|
||||
// globally rather than per replica.
|
||||
//
|
||||
// go run ./99-experiments/global-ratelimit/server -rate 10 -burst 10
|
||||
// curl -i 'localhost:8100/allow?key=tenant-a'
|
||||
// curl -s 'localhost:8100/stats'
|
||||
//
|
||||
// API
|
||||
//
|
||||
// POST|GET /allow?key=K[&n=1]
|
||||
// 200 {"allowed":true, "remaining":9, "limit":10, "rate":10}
|
||||
// 429 {"allowed":false, "remaining":0.3, "limit":10, "rate":10, "retry_after_ms":70}
|
||||
// headers: X-RateLimit-Limit, X-RateLimit-Remaining, Retry-After (429 only)
|
||||
// 400 when key is missing or n is not a positive integer
|
||||
// GET /stats → {"keys":N,"allowed_total":A,"denied_total":D}
|
||||
// GET /healthz → 200 ok
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"flag"
|
||||
"fmt"
|
||||
"log"
|
||||
"math"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/signal"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"gosre-exercises/99-experiments/global-ratelimit/limiter"
|
||||
)
|
||||
|
||||
type allowResponse struct {
|
||||
Allowed bool `json:"allowed"`
|
||||
Remaining float64 `json:"remaining"`
|
||||
Limit int `json:"limit"`
|
||||
Rate float64 `json:"rate"`
|
||||
RetryAfterMS int64 `json:"retry_after_ms,omitempty"`
|
||||
}
|
||||
|
||||
func allowHandler(store *limiter.Store) http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
key := r.URL.Query().Get("key")
|
||||
if key == "" {
|
||||
http.Error(w, `missing "key" query parameter`, http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
n := 1
|
||||
if raw := r.URL.Query().Get("n"); raw != "" {
|
||||
v, err := strconv.Atoi(raw)
|
||||
if err != nil || v <= 0 {
|
||||
http.Error(w, `"n" must be a positive integer`, http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
n = v
|
||||
}
|
||||
|
||||
d := store.Allow(key, n)
|
||||
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.Header().Set("X-RateLimit-Limit", strconv.Itoa(d.Limit))
|
||||
w.Header().Set("X-RateLimit-Remaining", strconv.Itoa(int(math.Floor(d.Remaining))))
|
||||
resp := allowResponse{Allowed: d.Allowed, Remaining: d.Remaining, Limit: d.Limit, Rate: d.Rate}
|
||||
status := http.StatusOK
|
||||
if !d.Allowed {
|
||||
status = http.StatusTooManyRequests
|
||||
resp.RetryAfterMS = d.RetryAfter.Milliseconds()
|
||||
// Retry-After is whole seconds; round UP so clients don't retry too early.
|
||||
w.Header().Set("Retry-After", strconv.Itoa(int(math.Ceil(d.RetryAfter.Seconds()))))
|
||||
}
|
||||
w.WriteHeader(status)
|
||||
json.NewEncoder(w).Encode(resp)
|
||||
}
|
||||
}
|
||||
|
||||
func statsHandler(store *limiter.Store) http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(store.Stats())
|
||||
}
|
||||
}
|
||||
|
||||
// parseLimits parses "key=rate:burst,key2=rate:burst".
|
||||
func parseLimits(s string) (map[string]limiter.Limit, error) {
|
||||
out := map[string]limiter.Limit{}
|
||||
if strings.TrimSpace(s) == "" {
|
||||
return out, nil
|
||||
}
|
||||
for _, item := range strings.Split(s, ",") {
|
||||
key, spec, ok := strings.Cut(strings.TrimSpace(item), "=")
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("bad limit %q: want key=rate:burst", item)
|
||||
}
|
||||
rateS, burstS, ok := strings.Cut(spec, ":")
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("bad limit %q: want key=rate:burst", item)
|
||||
}
|
||||
rate, err1 := strconv.ParseFloat(rateS, 64)
|
||||
burst, err2 := strconv.Atoi(burstS)
|
||||
if err1 != nil || err2 != nil || rate <= 0 || burst <= 0 {
|
||||
return nil, fmt.Errorf("bad limit %q: rate and burst must be positive", item)
|
||||
}
|
||||
out[key] = limiter.Limit{Rate: rate, Burst: burst}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func main() {
|
||||
addr := flag.String("addr", ":8100", "listen address")
|
||||
rate := flag.Float64("rate", 10, "default tokens/second for keys without an override")
|
||||
burst := flag.Int("burst", 20, "default burst for keys without an override")
|
||||
limits := flag.String("limits", "", `per-key overrides, e.g. "tenant-a=100:200,tenant-b=5:5"`)
|
||||
sweep := flag.Duration("sweep", time.Minute, "evict buckets idle for at least this long (0 = never)")
|
||||
flag.Parse()
|
||||
|
||||
overrides, err := parseLimits(*limits)
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
os.Exit(2)
|
||||
}
|
||||
store := limiter.NewStore(limiter.Limit{Rate: *rate, Burst: *burst}, overrides)
|
||||
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("/allow", allowHandler(store))
|
||||
mux.HandleFunc("/stats", statsHandler(store))
|
||||
mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { fmt.Fprintln(w, "ok") })
|
||||
|
||||
srv := &http.Server{
|
||||
Addr: *addr,
|
||||
Handler: mux,
|
||||
// Never run a server with zero timeouts (see 99-experiments/body-drain).
|
||||
ReadHeaderTimeout: 5 * time.Second,
|
||||
WriteTimeout: 5 * time.Second,
|
||||
IdleTimeout: 60 * time.Second,
|
||||
}
|
||||
|
||||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
|
||||
defer stop()
|
||||
|
||||
// Idle-key eviction: the "map grows forever" fix. Also a periodic stats line.
|
||||
go func() {
|
||||
if *sweep <= 0 {
|
||||
return
|
||||
}
|
||||
t := time.NewTicker(*sweep)
|
||||
defer t.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-t.C:
|
||||
removed := store.Sweep(*sweep)
|
||||
st := store.Stats()
|
||||
log.Printf("sweep: evicted=%d keys=%d allowed=%d denied=%d", removed, st.Keys, st.Allowed, st.Denied)
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
go func() {
|
||||
log.Printf("global rate limiter listening on %s (default %v/s burst %d, %d overrides)", *addr, *rate, *burst, len(overrides))
|
||||
if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
|
||||
log.Fatal(err)
|
||||
}
|
||||
}()
|
||||
|
||||
<-ctx.Done()
|
||||
log.Println("shutting down")
|
||||
shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
if err := srv.Shutdown(shutdownCtx); err != nil {
|
||||
log.Printf("shutdown: %v", err)
|
||||
}
|
||||
st := store.Stats()
|
||||
log.Printf("final: keys=%d allowed=%d denied=%d", st.Keys, st.Allowed, st.Denied)
|
||||
}
|
||||
Reference in New Issue
Block a user