diff --git a/groupcache.go b/groupcache.go index bc123f1d..c31b370f 100644 --- a/groupcache.go +++ b/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() { diff --git a/groupcache_test.go b/groupcache_test.go index 1bfe278c..f873e94d 100644 --- a/groupcache_test.go +++ b/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. diff --git a/http.go b/http.go index e0d391a5..6dae3c39 100644 --- a/http.go +++ b/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 diff --git a/lru/lru.go b/lru/lru.go index eac1c766..9b9c6b89 100644 --- a/lru/lru.go +++ b/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 { diff --git a/lru/lru_test.go b/lru/lru_test.go index a14f439e..a3e7663e 100644 --- a/lru/lru_test.go +++ b/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{}) { diff --git a/stats.go b/stats.go new file mode 100644 index 00000000..cefada20 --- /dev/null +++ b/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)) + } +} diff --git a/stats_test.go b/stats_test.go new file mode 100644 index 00000000..31b7337b --- /dev/null +++ b/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) + } +}