cache: Coalesce in-flight identical upstream fetches
In-flight coalescing already dedups identical misses, but acceptance still lacked a gated test (without a hold, waiters can become sequential HITs after the first fill) and the Quick check metrics table omitted cache_coalesced. Add steamcache/coalesce_test.go: N concurrent identical GETs share one upstream fill (leader MISS, waiters HIT-COALESCED) and a 5xx sibling. Document cache_coalesced next to the other hit/miss fields. Fixes #35
This commit is contained in:
@@ -0,0 +1,199 @@
|
||||
package steamcache
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func coalescerWaiterCount(sc *SteamCache, cacheKey string) int32 {
|
||||
sc.coalescer.mu.Lock()
|
||||
defer sc.coalescer.mu.Unlock()
|
||||
cr := sc.coalescer.requests[cacheKey]
|
||||
if cr == nil {
|
||||
return 0
|
||||
}
|
||||
return cr.waitingCount.Load()
|
||||
}
|
||||
|
||||
func steamCoalesceRequest(path string) *http.Request {
|
||||
req := httptest.NewRequest(http.MethodGet, path, nil)
|
||||
req.Header.Set("User-Agent", "Valve/Steam HTTP Client 1.0")
|
||||
return req
|
||||
}
|
||||
|
||||
func waitForCoalescerJoin(t *testing.T, sc *SteamCache, cacheKey string, n int, release func(), wg *sync.WaitGroup, upstreamCalls *atomic.Int64) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(2 * time.Second)
|
||||
var waiters int32
|
||||
for {
|
||||
waiters = coalescerWaiterCount(sc, cacheKey)
|
||||
if waiters >= int32(n) {
|
||||
return
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
release()
|
||||
wg.Wait()
|
||||
t.Fatalf("coalescer waiters=%d want %d (upstreamCalls=%d)", waiters, n, upstreamCalls.Load())
|
||||
}
|
||||
time.Sleep(1 * time.Millisecond)
|
||||
}
|
||||
}
|
||||
|
||||
// TestCoalesceIdenticalMissesOneUpstreamGET holds the leader's upstream GET
|
||||
// open until every concurrent client has joined the in-flight coalescer.
|
||||
// Without that hold, later requests can become sequential HITs after the first
|
||||
// miss fills, which would not prove coalescing.
|
||||
func TestCoalesceIdenticalMissesOneUpstreamGET(t *testing.T) {
|
||||
const nClients = 8
|
||||
body := []byte("coalesced depot chunk body")
|
||||
var upstreamCalls atomic.Int64
|
||||
release := make(chan struct{})
|
||||
var releaseOnce sync.Once
|
||||
releaseUpstream := func() { releaseOnce.Do(func() { close(release) }) }
|
||||
t.Cleanup(releaseUpstream)
|
||||
|
||||
f := func(w http.ResponseWriter, r *http.Request) {
|
||||
upstreamCalls.Add(1)
|
||||
select {
|
||||
case <-release:
|
||||
case <-r.Context().Done():
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/octet-stream")
|
||||
_, _ = w.Write(body)
|
||||
}
|
||||
sc, _ := newTestCacheWithFakeUpstream(t, f, "1MB", "0")
|
||||
sc.ResetMetrics()
|
||||
|
||||
const depotPath = "/depot/1684171/chunk/coalesce-inflight"
|
||||
cacheKey, err := generateServiceCacheKey(depotPath, "steam")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
type clientResult struct {
|
||||
status int
|
||||
hdr string
|
||||
body []byte
|
||||
}
|
||||
results := make([]clientResult, nClients)
|
||||
|
||||
var wg sync.WaitGroup
|
||||
start := make(chan struct{})
|
||||
wg.Add(nClients)
|
||||
for i := 0; i < nClients; i++ {
|
||||
go func(i int) {
|
||||
defer wg.Done()
|
||||
<-start
|
||||
rec := httptest.NewRecorder()
|
||||
sc.ServeHTTP(rec, steamCoalesceRequest(depotPath))
|
||||
results[i] = clientResult{
|
||||
status: rec.Code,
|
||||
hdr: rec.Header().Get("X-LanCache-Status"),
|
||||
body: rec.Body.Bytes(),
|
||||
}
|
||||
}(i)
|
||||
}
|
||||
close(start)
|
||||
|
||||
waitForCoalescerJoin(t, sc, cacheKey, nClients, releaseUpstream, &wg, &upstreamCalls)
|
||||
releaseUpstream()
|
||||
wg.Wait()
|
||||
|
||||
if got := upstreamCalls.Load(); got != 1 {
|
||||
t.Fatalf("expected exactly 1 upstream GET, got %d", got)
|
||||
}
|
||||
|
||||
var miss, coalesced int
|
||||
for i, r := range results {
|
||||
if r.status != http.StatusOK {
|
||||
t.Errorf("client %d: expected 200, got %d", i, r.status)
|
||||
}
|
||||
if !bytes.Equal(r.body, body) {
|
||||
t.Errorf("client %d: body mismatch: got %q", i, r.body)
|
||||
}
|
||||
switch r.hdr {
|
||||
case "MISS":
|
||||
miss++
|
||||
case "HIT-COALESCED":
|
||||
coalesced++
|
||||
default:
|
||||
t.Errorf("client %d: unexpected X-LanCache-Status %q", i, r.hdr)
|
||||
}
|
||||
}
|
||||
if miss != 1 {
|
||||
t.Errorf("expected 1 MISS leader, got %d", miss)
|
||||
}
|
||||
if coalesced != nClients-1 {
|
||||
t.Errorf("expected %d HIT-COALESCED waiters, got %d", nClients-1, coalesced)
|
||||
}
|
||||
if got := sc.GetMetrics().CacheCoalesced; got < int64(nClients-1) {
|
||||
t.Errorf("CacheCoalesced=%d, want >= %d", got, nClients-1)
|
||||
}
|
||||
}
|
||||
|
||||
// TestCoalesceIdenticalMissesSharedUpstreamError is the 5xx sibling: waiters
|
||||
// share the leader's failure instead of each hitting origin. Upstream 500 is
|
||||
// retried, so the call count is the leader's retry budget (not N).
|
||||
func TestCoalesceIdenticalMissesSharedUpstreamError(t *testing.T) {
|
||||
const nClients = 8
|
||||
var upstreamCalls atomic.Int64
|
||||
release := make(chan struct{})
|
||||
var releaseOnce sync.Once
|
||||
releaseUpstream := func() { releaseOnce.Do(func() { close(release) }) }
|
||||
t.Cleanup(releaseUpstream)
|
||||
|
||||
f := func(w http.ResponseWriter, r *http.Request) {
|
||||
upstreamCalls.Add(1)
|
||||
select {
|
||||
case <-release:
|
||||
case <-r.Context().Done():
|
||||
return
|
||||
}
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
}
|
||||
sc, _ := newTestCacheWithFakeUpstream(t, f, "1MB", "0")
|
||||
sc.ResetMetrics()
|
||||
|
||||
const depotPath = "/depot/1684171/chunk/coalesce-inflight-err"
|
||||
cacheKey, err := generateServiceCacheKey(depotPath, "steam")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
codes := make([]int, nClients)
|
||||
var wg sync.WaitGroup
|
||||
start := make(chan struct{})
|
||||
wg.Add(nClients)
|
||||
for i := 0; i < nClients; i++ {
|
||||
go func(i int) {
|
||||
defer wg.Done()
|
||||
<-start
|
||||
rec := httptest.NewRecorder()
|
||||
sc.ServeHTTP(rec, steamCoalesceRequest(depotPath))
|
||||
codes[i] = rec.Code
|
||||
}(i)
|
||||
}
|
||||
close(start)
|
||||
|
||||
waitForCoalescerJoin(t, sc, cacheKey, nClients, releaseUpstream, &wg, &upstreamCalls)
|
||||
releaseUpstream()
|
||||
wg.Wait()
|
||||
|
||||
if got := upstreamCalls.Load(); got < 1 || got >= int64(nClients) {
|
||||
t.Fatalf("expected coalesced origin GETs (leader + retries, < %d waiters), got %d", nClients, got)
|
||||
}
|
||||
for i, code := range codes {
|
||||
if code != http.StatusInternalServerError {
|
||||
t.Errorf("client %d: expected 500, got %d", i, code)
|
||||
}
|
||||
}
|
||||
if got := sc.GetMetrics().Errors; got < int64(nClients) {
|
||||
t.Errorf("Errors=%d, want >= %d (once per client)", got, nClients)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user