s1d3sw1ped
c3464d692e
- Introduced coalescing logic in `coalescing.go` to handle concurrent identical upstream fetches, including a state machine and response buffering for improved performance. - Implemented a new cache file format in `format.go`, supporting serialization and deserialization of HTTP responses, along with range request handling. - Developed an HTTP handler in `handler.go` to manage requests, including special endpoint handling and metrics reporting. - Added rate limiting functionality in `ratelimit.go` to control per-client and global request rates, enhancing security and performance. - Created service management capabilities in `service.go` to define and manage cacheable services, including user-agent detection. - Updated tests in `steamcache_test.go` to cover new functionalities, ensuring robustness and reliability across the codebase.
118 lines
3.3 KiB
Go
118 lines
3.3 KiB
Go
// steamcache/coalescing.go
|
|
// Request coalescing (de-duplicating concurrent identical upstream fetches for the same
|
|
// cache key). Includes the coalescedRequest state machine + waiter/leader coordination,
|
|
// response buffering for thundering herd avoidance, and the coalescer wrapper that
|
|
// owns the in-flight map + mutex (SteamCache methods delegate; no direct map access
|
|
// in core or handler).
|
|
package steamcache
|
|
|
|
import (
|
|
"net/http"
|
|
"sync"
|
|
"sync/atomic"
|
|
)
|
|
|
|
type coalescedRequest struct {
|
|
waitingCount atomic.Int32
|
|
done bool
|
|
mu sync.Mutex
|
|
// Buffered response data for coalesced clients
|
|
responseData []byte
|
|
responseHeaders http.Header
|
|
statusCode int
|
|
// Broadcast signal for all waiters (closed by leader in complete)
|
|
doneCh chan struct{}
|
|
completionErr error
|
|
// Active protocol (post-legacy cleanup): waiters wake on doneCh, then read completionErr/response* under mu (or pre-unlock copies in waiter).
|
|
}
|
|
|
|
func newCoalescedRequest() *coalescedRequest {
|
|
cr := &coalescedRequest{
|
|
done: false,
|
|
responseHeaders: make(http.Header),
|
|
doneCh: make(chan struct{}),
|
|
}
|
|
cr.waitingCount.Store(1)
|
|
return cr
|
|
}
|
|
|
|
func (cr *coalescedRequest) addWaiter() {
|
|
cr.waitingCount.Add(1)
|
|
}
|
|
|
|
func (cr *coalescedRequest) complete(resp *http.Response, err error) {
|
|
cr.mu.Lock()
|
|
defer cr.mu.Unlock()
|
|
if cr.done {
|
|
return
|
|
}
|
|
cr.done = true
|
|
|
|
if err != nil {
|
|
cr.completionErr = err
|
|
} else {
|
|
// Store response data for coalesced clients
|
|
if resp != nil {
|
|
cr.statusCode = resp.StatusCode
|
|
// Copy headers (excluding hop-by-hop headers via filter)
|
|
cr.responseHeaders = filterHopByHopHeaders(resp.Header)
|
|
}
|
|
}
|
|
// Broadcast to *all* waiters (thundering herd fix). Close is safe here because of the done guard above.
|
|
close(cr.doneCh)
|
|
}
|
|
|
|
// setResponseData stores the buffered response data for coalesced clients
|
|
func (cr *coalescedRequest) setResponseData(data []byte) {
|
|
cr.mu.Lock()
|
|
defer cr.mu.Unlock()
|
|
cr.responseData = make([]byte, len(data))
|
|
copy(cr.responseData, data)
|
|
}
|
|
|
|
// coalescer owns the coalesced requests map and mutex. It encapsulates the
|
|
// in-flight request dedup state so SteamCache no longer directly manipulates
|
|
// the raw map (Phase 2 extraction). Unexported; same-package access for tests.
|
|
type coalescer struct {
|
|
mu sync.Mutex
|
|
requests map[string]*coalescedRequest
|
|
}
|
|
|
|
// newCoalescer constructs an empty coalescer (called from SteamCache.New).
|
|
func newCoalescer() *coalescer {
|
|
return &coalescer{
|
|
requests: make(map[string]*coalescedRequest),
|
|
}
|
|
}
|
|
|
|
func (c *coalescer) getOrCreate(cacheKey string) (*coalescedRequest, bool) {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
|
|
if cr, exists := c.requests[cacheKey]; exists {
|
|
cr.addWaiter()
|
|
return cr, false
|
|
}
|
|
|
|
cr := newCoalescedRequest()
|
|
c.requests[cacheKey] = cr
|
|
return cr, true
|
|
}
|
|
|
|
func (c *coalescer) remove(cacheKey string) {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
delete(c.requests, cacheKey)
|
|
}
|
|
|
|
// getOrCreateCoalescedRequest delegates to the owned coalescer (preserves
|
|
// existing call sites in handler.go and any white-box tests unchanged).
|
|
func (sc *SteamCache) getOrCreateCoalescedRequest(cacheKey string) (*coalescedRequest, bool) {
|
|
return sc.coalescer.getOrCreate(cacheKey)
|
|
}
|
|
|
|
// removeCoalescedRequest delegates to the owned coalescer.
|
|
func (sc *SteamCache) removeCoalescedRequest(cacheKey string) {
|
|
sc.coalescer.remove(cacheKey)
|
|
}
|