groupcache: populate hotCache based on reported peer QPS

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

lru/lru.go

@@ -82,6 +82,17 @@ func (c *Cache) Get(key Key) (value interface{}, ok bool) {
 	return
 }
 
+// Peek looks up a key's value without updating its recency.
+func (c *Cache) Peek(key Key) (value interface{}, ok bool) {
+	if c.cache == nil {
+		return
+	}
+	if ele, hit := c.cache[key]; hit {
+		return ele.Value.(*entry).value, true
+	}
+	return
+}
+
 // Remove removes the provided key from the cache.
 func (c *Cache) Remove(key Key) {
 	if c.cache == nil {

lru/lru_test.go

@@ -73,6 +73,26 @@ func TestRemove(t *testing.T) {
 	}
 }
 
+func TestPeek(t *testing.T) {
+	lru := New(2)
+	lru.Add("a", 1)
+	lru.Add("b", 2)
+
+	// Peeking "a" must not refresh its recency.
+	if val, ok := lru.Peek("a"); !ok || val != 1 {
+		t.Fatalf("Peek(a) = %v, %v; want 1, true", val, ok)
+	}
+	if _, ok := lru.Peek("missing"); ok {
+		t.Fatal("Peek(missing) reported a hit")
+	}
+
+	// Adding "c" should evict "a" as the least recently used.
+	lru.Add("c", 3)
+	if _, ok := lru.Get("a"); ok {
+		t.Fatal("Peek refreshed recency; oldest entry survived eviction")
+	}
+}
+
 func TestEvict(t *testing.T) {
 	evictedKeys := make([]Key, 0)
 	onEvictedFun := func(key Key, value interface{}) {

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))
+	}
+}