groupcache: populate hotCache based on reported peer QPS

Previously, getFromPeer mirrored a remotely fetched value into the
hotCache on a fixed 10% of fetches, chosen at random. This was a
placeholder (noted in a TODO) and is a poor signal: it pollutes the
hotCache with one-off keys while taking ~10 round trips on average to
mirror a genuinely hot key, defeating the purpose of the hotCache.

Use the value's owner as the authority on hotness instead. The owner
tracks a per-key request rate using an exponentially weighted moving
average and reports it in the GetResponse.MinuteQps field (which the
wire format already carried but nobody populated). A non-owning peer
mirrors a key into its hotCache only once the owner reports a rate at
or above hotQPS.

The per-key rate state lives inside the mainCache entry, so it is
bounded by cache residency and cleaned up by ordinary eviction, with
no separate accounting to evict.

This replaces the test-only *rand.Rand hook (added in #175) with an
injectable clock, which TestPeers and the new stats tests use to drive
QPS deterministically.

Author: Mert Ovun

groupcache.go

@@ -27,16 +27,21 @@ package groupcache
 import (
 	"context"
 	"errors"
-	"math/rand"
 	"strconv"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	pb "github.com/golang/groupcache/groupcachepb"
 	"github.com/golang/groupcache/lru"
 	"github.com/golang/groupcache/singleflight"
 )
 
+// hotQPS is the request rate, in requests per qpsWindow as seen by a key's
+// owner, above which non-owning peers begin mirroring the key into their
+// hotCache to shed load from the owner.
+const hotQPS = 10
+
 // A Getter loads data for a key.
 type Getter interface {
 	// Get returns the value identified by key, populating dest.
@@ -103,6 +108,7 @@ func newGroup(name string, cacheBytes int64, getter Getter, peers PeerPicker) *G
 		cacheBytes: cacheBytes,
 		loadGroup:  &singleflight.Group{},
 	}
+	g.mainCache.trackQPS = true
 	if fn := newGroupHook; fn != nil {
 		fn(g)
 	}
@@ -172,9 +178,16 @@ type Group struct {
 	// Stats are statistics on the group.
 	Stats Stats
 
-	// rand is only non-nil when testing,
-	// to get predictable results in TestPeers.
-	rand *rand.Rand
+	// now returns the current time. It is overridden in tests to make
+	// the QPS-driven hotCache population deterministic.
+	now func() time.Time
+}
+
+func (g *Group) timeNow() time.Time {
+	if g.now != nil {
+		return g.now()
+	}
+	return time.Now()
 }
 
 // flightGroup is defined as an interface which flightgroup.Group
@@ -215,6 +228,7 @@ func (g *Group) Get(ctx context.Context, key string, dest Sink) error {
 	if dest == nil {
 		return errors.New("groupcache: nil dest Sink")
 	}
+	g.mainCache.recordRequest(key, g.timeNow())
 	value, cacheHit := g.lookupCache(key)
 
 	if cacheHit {
@@ -316,16 +330,7 @@ func (g *Group) getFromPeer(ctx context.Context, peer ProtoGetter, key string) (
 		return ByteView{}, err
 	}
 	value := ByteView{b: res.Value}
-	// TODO(bradfitz): use res.MinuteQps or something smart to
-	// conditionally populate hotCache.  For now just do it some
-	// percentage of the time.
-	var pop bool
-	if g.rand != nil {
-		pop = g.rand.Intn(10) == 0
-	} else {
-		pop = rand.Intn(10) == 0
-	}
-	if pop {
+	if res.GetMinuteQps() >= hotQPS {
 		g.populateCache(key, value, &g.hotCache)
 	}
 	return value, nil
@@ -398,13 +403,24 @@ func (g *Group) CacheStats(which CacheType) CacheStats {
 // makes values always be ByteView, and counts the size of all keys and
 // values.
 type cache struct {
-	mu         sync.RWMutex
-	nbytes     int64 // of all keys and values
-	lru        *lru.Cache
+	mu     sync.RWMutex
+	nbytes int64 // of all keys and values
+	lru    *lru.Cache
+	// trackQPS records a per-key request rate for owner-side hotness
+	// reporting. It is enabled only on a group's mainCache.
+	trackQPS   bool
 	nhit, nget int64
 	nevict     int64 // number of evictions
 }
 
+// cacheValue is the value stored in the underlying LRU. Its optional stats
+// live and die with the entry, so per-key rate tracking is bounded by cache
+// residency and cleaned up by ordinary eviction.
+type cacheValue struct {
+	view  ByteView
+	stats *keyStats
+}
+
 func (c *cache) stats() CacheStats {
 	c.mu.RLock()
 	defer c.mu.RUnlock()
@@ -423,13 +439,17 @@ func (c *cache) add(key string, value ByteView) {
 	if c.lru == nil {
 		c.lru = &lru.Cache{
 			OnEvicted: func(key lru.Key, value interface{}) {
-				val := value.(ByteView)
-				c.nbytes -= int64(len(key.(string))) + int64(val.Len())
+				cv := value.(*cacheValue)
+				c.nbytes -= int64(len(key.(string))) + int64(cv.view.Len())
 				c.nevict++
 			},
 		}
 	}
-	c.lru.Add(key, value)
+	cv := &cacheValue{view: value}
+	if c.trackQPS {
+		cv.stats = &keyStats{}
+	}
+	c.lru.Add(key, cv)
 	c.nbytes += int64(len(key)) + int64(value.Len())
 }
 
@@ -445,7 +465,39 @@ func (c *cache) get(key string) (value ByteView, ok bool) {
 		return
 	}
 	c.nhit++
-	return vi.(ByteView), true
+	return vi.(*cacheValue).view, true
+}
+
+// recordRequest registers one request for key against its tracked rate, if the
+// key is present and tracked.
+func (c *cache) recordRequest(key string, now time.Time) {
+	c.stats0(key, now, true)
+}
+
+// peekQPS returns key's estimated requests-per-qpsWindow without registering a
+// new request, or 0 if the key is absent or untracked.
+func (c *cache) peekQPS(key string, now time.Time) float64 {
+	return c.stats0(key, now, false)
+}
+
+func (c *cache) stats0(key string, now time.Time, record bool) float64 {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	if c.lru == nil {
+		return 0
+	}
+	vi, ok := c.lru.Peek(key)
+	if !ok {
+		return 0
+	}
+	stats := vi.(*cacheValue).stats
+	if stats == nil {
+		return 0
+	}
+	if record {
+		return stats.touch(now)
+	}
+	return stats.peek(now)
 }
 
 func (c *cache) removeOldest() {

groupcache_test.go

@@ -23,7 +23,6 @@ import (
 	"errors"
 	"fmt"
 	"hash/crc32"
-	"math/rand"
 	"reflect"
 	"sync"
 	"testing"
@@ -229,6 +228,7 @@ func TestCacheEviction(t *testing.T) {
 type fakePeer struct {
 	hits int
 	fail bool
+	qps  float64 // reported owner-side request rate
 }
 
 func (p *fakePeer) Get(_ context.Context, in *pb.GetRequest, out *pb.GetResponse) error {
@@ -237,6 +237,7 @@ func (p *fakePeer) Get(_ context.Context, in *pb.GetRequest, out *pb.GetResponse
 		return errors.New("simulated error from peer")
 	}
 	out.Value = []byte("got:" + in.GetKey())
+	out.MinuteQps = &p.qps
 	return nil
 }
 
@@ -264,7 +265,6 @@ func TestPeers(t *testing.T) {
 		return dest.SetString("got:" + key)
 	}
 	testGroup := newGroup("TestPeers-group", cacheSize, GetterFunc(getter), peerList)
-	testGroup.rand = rand.New(rand.NewSource(123))
 	run := func(name string, n int, wantSummary string) {
 		// Reset counters
 		localHits = 0
@@ -303,9 +303,21 @@ func TestPeers(t *testing.T) {
 	resetCacheSize(1 << 20)
 	run("base", 200, "localHits = 49, peers = 51 49 51")
 
-	// Verify cache was hit.  All localHits are gone, and some of
-	// the peer hits (the ones randomly selected to be maybe hot)
-	run("cached_base", 200, "localHits = 0, peers = 49 47 48")
+	// Peers report cold keys, so nothing is mirrored into hotCache.
+	// Locally-owned keys are served from mainCache (no local hits),
+	// but peer-owned keys still require a peer fetch every time.
+	run("cached_base", 200, "localHits = 0, peers = 51 49 51")
+
+	// Peers now report hot keys, so peer-owned keys are mirrored into
+	// hotCache on this pass...
+	for _, p := range []*fakePeer{peer0, peer1, peer2} {
+		p.qps = hotQPS
+	}
+	run("warm_hot", 200, "localHits = 0, peers = 51 49 51")
+
+	// ...and the subsequent pass serves them all from hotCache.
+	run("cached_hot", 200, "localHits = 0, peers = 0 0 0")
+
 	resetCacheSize(0)
 
 	// With one of the peers being down.

http.go

@@ -174,7 +174,8 @@ func (p *HTTPPool) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 
 	// Write the value to the response body as a proto message.
-	body, err := proto.Marshal(&pb.GetResponse{Value: value})
+	qps := group.mainCache.peekQPS(key, group.timeNow())
+	body, err := proto.Marshal(&pb.GetResponse{Value: value, MinuteQps: &qps})
 	if err != nil {
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return

stats.go

@@ -0,0 +1,62 @@
+/*
+Copyright 2012 Google Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+     http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package groupcache
+
+import (
+	"math"
+	"sync"
+	"time"
+)
+
+// qpsWindow is the half-life over which request rate is measured.
+const qpsWindow = time.Minute
+
+// keyStats estimates the per-minute request rate for a single key using an
+// exponentially weighted moving average. Each observation decays toward zero
+// with a half-life of qpsWindow, so a key that stops being requested fades out
+// of "hotness" on its own.
+type keyStats struct {
+	mu    sync.Mutex
+	rate  float64
+	stamp time.Time
+}
+
+// touch records one request at time now and returns the resulting estimated
+// rate in requests per qpsWindow.
+func (k *keyStats) touch(now time.Time) float64 {
+	k.mu.Lock()
+	defer k.mu.Unlock()
+	k.decay(now)
+	k.rate++
+	k.stamp = now
+	return k.rate
+}
+
+// peek returns the estimated rate at time now without recording a request.
+func (k *keyStats) peek(now time.Time) float64 {
+	k.mu.Lock()
+	defer k.mu.Unlock()
+	k.decay(now)
+	k.stamp = now
+	return k.rate
+}
+
+func (k *keyStats) decay(now time.Time) {
+	if !k.stamp.IsZero() {
+		k.rate *= math.Exp2(-float64(now.Sub(k.stamp)) / float64(qpsWindow))
+	}
+}

stats_test.go

@@ -0,0 +1,75 @@
+/*
+Copyright 2012 Google Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+     http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package groupcache
+
+import (
+	"testing"
+	"time"
+)
+
+func TestKeyStatsRampUp(t *testing.T) {
+	var k keyStats
+	now := time.Unix(0, 0)
+	if got := k.touch(now); got != 1 {
+		t.Fatalf("first touch = %v; want 1", got)
+	}
+	// Many requests within the same instant accumulate without decay.
+	for i := 0; i < 9; i++ {
+		k.touch(now)
+	}
+	if got := k.peek(now); got != 10 {
+		t.Errorf("rate after 10 instantaneous touches = %v; want 10", got)
+	}
+}
+
+func TestKeyStatsDecay(t *testing.T) {
+	var k keyStats
+	now := time.Unix(0, 0)
+	k.touch(now)
+	// After one half-life with no requests, the rate should halve.
+	if got := k.peek(now.Add(qpsWindow)); got != 0.5 {
+		t.Errorf("rate after one window = %v; want 0.5", got)
+	}
+	// And after another, halve again.
+	if got := k.peek(now.Add(2 * qpsWindow)); got != 0.25 {
+		t.Errorf("rate after two windows = %v; want 0.25", got)
+	}
+}
+
+func TestCacheQPSUntracked(t *testing.T) {
+	// hotCache does not track QPS, so peekQPS is always 0.
+	var c cache
+	c.add("k", ByteView{s: "v"})
+	if got := c.peekQPS("k", time.Unix(0, 0)); got != 0 {
+		t.Errorf("untracked peekQPS = %v; want 0", got)
+	}
+	if got := c.peekQPS("absent", time.Unix(0, 0)); got != 0 {
+		t.Errorf("absent peekQPS = %v; want 0", got)
+	}
+}
+
+func TestCacheQPSTracked(t *testing.T) {
+	c := cache{trackQPS: true}
+	c.add("k", ByteView{s: "v"})
+	now := time.Unix(0, 0)
+	for i := 0; i < 5; i++ {
+		c.recordRequest("k", now)
+	}
+	if got := c.peekQPS("k", now); got != 5 {
+		t.Errorf("tracked peekQPS = %v; want 5", got)
+	}
+}