LFU Cache Fixes

dev.env

@@ -1,13 +1,13 @@
 # No shebang line as this script is sourced from an external shell.
 
 # Copyright 2019 The Vitess Authors.
-# 
+#
 # 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.
@@ -34,6 +34,6 @@ export PATH
 
 # According to https://github.com/etcd-io/etcd/blob/a621d807f061e1dd635033a8d6bc261461429e27/Documentation/op-guide/supported-platform.md,
 # currently, etcd is unstable on arm64, so ETCD_UNSUPPORTED_ARCH should be set.
-if [ "$(uname -m)" == aarch64 ]; then
+if [ "$(uname -m)" = aarch64 ]; then
     export ETCD_UNSUPPORTED_ARCH=arm64
 fi

go/cache/cache.go

@@ -76,5 +76,5 @@ type Config struct {
 var DefaultConfig = &Config{
 	MaxEntries:     5000,
 	MaxMemoryUsage: 32 * 1024 * 1024,
-	LFU:            false,
+	LFU:            true,
 }

go/cache/ristretto/bloom/bbloom.go

@@ -21,9 +21,6 @@
 package bloom
 
 import (
-	"bytes"
-	"encoding/json"
-	"log"
 	"math"
 	"unsafe"
 )
@@ -43,26 +40,16 @@ func getSize(ui64 uint64) (size uint64, exponent uint64) {
 	return size, exponent
 }
 
-func calcSizeByWrongPositives(numEntries, wrongs float64) (uint64, uint64) {
-	size := -1 * numEntries * math.Log(wrongs) / math.Pow(float64(0.69314718056), 2)
-	locs := math.Ceil(float64(0.69314718056) * size / numEntries)
-	return uint64(size), uint64(locs)
+// NewBloomFilterWithErrorRate returns a new bloomfilter with optimal size for the given
+// error rate
+func NewBloomFilterWithErrorRate(numEntries uint64, wrongs float64) *Bloom {
+	size := -1 * float64(numEntries) * math.Log(wrongs) / math.Pow(0.69314718056, 2)
+	locs := math.Ceil(0.69314718056 * size / float64(numEntries))
+	return NewBloomFilter(uint64(size), uint64(locs))
 }
 
 // NewBloomFilter returns a new bloomfilter.
-func NewBloomFilter(params ...float64) (bloomfilter *Bloom) {
-	var entries, locs uint64
-	if len(params) == 2 {
-		if params[1] < 1 {
-			entries, locs = calcSizeByWrongPositives(params[0], params[1])
-		} else {
-			entries, locs = uint64(params[0]), uint64(params[1])
-		}
-	} else {
-		log.Fatal("usage: New(float64(number_of_entries), float64(number_of_hashlocations))" +
-			" i.e. New(float64(1000), float64(3)) or New(float64(number_of_entries)," +
-			" float64(number_of_hashlocations)) i.e. New(float64(1000), float64(0.03))")
-	}
+func NewBloomFilter(entries, locs uint64) (bloomfilter *Bloom) {
 	size, exponent := getSize(entries)
 	bloomfilter = &Bloom{
 		sizeExp: exponent,
@@ -162,49 +149,3 @@ func (bl *Bloom) IsSet(idx uint64) bool {
 	r := ((*(*uint8)(ptr)) >> (idx % 8)) & 1
 	return r == 1
 }
-
-// bloomJSONImExport
-// Im/Export structure used by JSONMarshal / JSONUnmarshal
-type bloomJSONImExport struct {
-	FilterSet []byte
-	SetLocs   uint64
-}
-
-// NewWithBoolset takes a []byte slice and number of locs per entry,
-// returns the bloomfilter with a bitset populated according to the input []byte.
-func newWithBoolset(bs *[]byte, locs uint64) *Bloom {
-	bloomfilter := NewBloomFilter(float64(len(*bs)<<3), float64(locs))
-	for i, b := range *bs {
-		*(*uint8)(unsafe.Pointer(uintptr(unsafe.Pointer(&bloomfilter.bitset[0])) + uintptr(i))) = b
-	}
-	return bloomfilter
-}
-
-// JSONUnmarshal takes JSON-Object (type bloomJSONImExport) as []bytes
-// returns bloom32 / bloom64 object.
-func JSONUnmarshal(dbData []byte) (*Bloom, error) {
-	bloomImEx := bloomJSONImExport{}
-	if err := json.Unmarshal(dbData, &bloomImEx); err != nil {
-		return nil, err
-	}
-	buf := bytes.NewBuffer(bloomImEx.FilterSet)
-	bs := buf.Bytes()
-	bf := newWithBoolset(&bs, bloomImEx.SetLocs)
-	return bf, nil
-}
-
-// JSONMarshal returns JSON-object (type bloomJSONImExport) as []byte.
-func (bl Bloom) JSONMarshal() []byte {
-	bloomImEx := bloomJSONImExport{}
-	bloomImEx.SetLocs = bl.setLocs
-	bloomImEx.FilterSet = make([]byte, len(bl.bitset)<<3)
-	for i := range bloomImEx.FilterSet {
-		bloomImEx.FilterSet[i] = *(*byte)(unsafe.Pointer(uintptr(unsafe.Pointer(&bl.bitset[0])) +
-			uintptr(i)))
-	}
-	data, err := json.Marshal(bloomImEx)
-	if err != nil {
-		log.Fatal("json.Marshal failed: ", err)
-	}
-	return data
-}

go/cache/ristretto/bloom/bbloom_test.go

@@ -6,14 +6,12 @@ import (
 	"os"
 	"testing"
 
-	"github.com/stretchr/testify/require"
-
 	"vitess.io/vitess/go/hack"
 )
 
 var (
 	wordlist1 [][]byte
-	n         = 1 << 16
+	n         = uint64(1 << 16)
 	bf        *Bloom
 )
 
@@ -31,7 +29,7 @@ func TestMain(m *testing.M) {
 }
 
 func TestM_NumberOfWrongs(t *testing.T) {
-	bf = NewBloomFilter(float64(n*10), float64(7))
+	bf = NewBloomFilter(n*10, 7)
 
 	cnt := 0
 	for i := range wordlist1 {
@@ -44,43 +42,14 @@ func TestM_NumberOfWrongs(t *testing.T) {
 
 }
 
-func TestM_JSON(t *testing.T) {
-	const shallBe = int(1 << 16)
-
-	bf = NewBloomFilter(float64(n*10), float64(7))
-
-	cnt := 0
-	for i := range wordlist1 {
-		hash := hack.RuntimeMemhash(wordlist1[i], 0)
-		if !bf.AddIfNotHas(hash) {
-			cnt++
-		}
-	}
-
-	jsonm := bf.JSONMarshal()
-
-	// create new bloomfilter from bloomfilter's JSON representation
-	bf2, err := JSONUnmarshal(jsonm)
-	require.NoError(t, err)
-
-	cnt2 := 0
-	for i := range wordlist1 {
-		hash := hack.RuntimeMemhash(wordlist1[i], 0)
-		if !bf2.AddIfNotHas(hash) {
-			cnt2++
-		}
-	}
-	require.Equal(t, shallBe, cnt2)
-}
-
 func BenchmarkM_New(b *testing.B) {
 	for r := 0; r < b.N; r++ {
-		_ = NewBloomFilter(float64(n*10), float64(7))
+		_ = NewBloomFilter(n*10, 7)
 	}
 }
 
 func BenchmarkM_Clear(b *testing.B) {
-	bf = NewBloomFilter(float64(n*10), float64(7))
+	bf = NewBloomFilter(n*10, 7)
 	for i := range wordlist1 {
 		hash := hack.RuntimeMemhash(wordlist1[i], 0)
 		bf.Add(hash)
@@ -92,7 +61,7 @@ func BenchmarkM_Clear(b *testing.B) {
 }
 
 func BenchmarkM_Add(b *testing.B) {
-	bf = NewBloomFilter(float64(n*10), float64(7))
+	bf = NewBloomFilter(n*10, 7)
 	b.ResetTimer()
 	for r := 0; r < b.N; r++ {
 		for i := range wordlist1 {

go/cache/ristretto/cache.go

@@ -45,7 +45,8 @@ func defaultStringHash(key string) (uint64, uint64) {
 
 type itemCallback func(*Item)
 
-const itemSize = int64(unsafe.Sizeof(storeItem{}))
+// CacheItemSize is the overhead in bytes for every stored cache item
+const CacheItemSize = int64(unsafe.Sizeof(storeItem{}))
 
 // Cache is a thread-safe implementation of a hashmap with a TinyLFU admission
 // policy and a Sampled LFU eviction policy. You can use the same Cache instance
@@ -446,7 +447,7 @@ func (c *Cache) processItems() {
 			}
 			if !c.ignoreInternalCost {
 				// Add the cost of internally storing the object.
-				i.Cost += itemSize
+				i.Cost += CacheItemSize
 			}
 
 			switch i.flag {

go/cache/ristretto/policy.go

@@ -375,7 +375,7 @@ type tinyLFU struct {
 func newTinyLFU(numCounters int64) *tinyLFU {
 	return &tinyLFU{
 		freq:    newCmSketch(numCounters),
-		door:    bloom.NewBloomFilter(float64(numCounters), 0.01),
+		door:    bloom.NewBloomFilterWithErrorRate(uint64(numCounters), 0.01),
 		resetAt: numCounters,
 	}
 }

go/mysql/fakesqldb/server.go

@@ -50,8 +50,8 @@ const appendEntry = -1
 type DB struct {
 	// Fields set at construction time.
 
-	// t is our testing.T instance
-	t *testing.T
+	// t is our testing.TB instance
+	t testing.TB
 
 	// listener is our mysql.Listener.
 	listener *mysql.Listener
@@ -151,7 +151,7 @@ type ExpectedExecuteFetch struct {
 }
 
 // New creates a server, and starts listening.
-func New(t *testing.T) *DB {
+func New(t testing.TB) *DB {
 	// Pick a path for our socket.
 	socketDir, err := ioutil.TempDir("", "fakesqldb")
 	if err != nil {

go/vt/vttablet/endtoend/config_test.go

@@ -27,6 +27,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"vitess.io/vitess/go/cache"
+	"vitess.io/vitess/go/cache/ristretto"
 	"vitess.io/vitess/go/sqltypes"
 	querypb "vitess.io/vitess/go/vt/proto/query"
 	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
@@ -178,9 +179,8 @@ func TestConsolidatorReplicasOnly(t *testing.T) {
 
 func TestQueryPlanCache(t *testing.T) {
 	if cache.DefaultConfig.LFU {
-		const cacheItemSize = 40
-		const cachedPlanSize = 2275 + cacheItemSize
-		const cachePlanSize2 = 2254 + cacheItemSize
+		const cachedPlanSize = 2352 + int(ristretto.CacheItemSize)
+		const cachePlanSize2 = 2326 + int(ristretto.CacheItemSize)
 		testQueryPlanCache(t, cachedPlanSize, cachePlanSize2)
 	} else {
 		testQueryPlanCache(t, 1, 1)
@@ -203,22 +203,23 @@ func testQueryPlanCache(t *testing.T, cachedPlanSize, cachePlanSize2 int) {
 	client := framework.NewClient()
 	_, _ = client.Execute("select * from vitess_test where intval=:ival1", bindVars)
 	_, _ = client.Execute("select * from vitess_test where intval=:ival2", bindVars)
-	time.Sleep(100 * time.Millisecond)
+	assert.Equal(t, 1, framework.Server.QueryPlanCacheLen())
 
 	vend := framework.DebugVars()
-	verifyIntValue(t, vend, "QueryCacheLength", 1)
+	// verifyIntValue(t, vend, "QueryCacheLength", 1)
 	verifyIntValue(t, vend, "QueryCacheSize", cachedPlanSize)
 	verifyIntValue(t, vend, "QueryCacheCapacity", cachedPlanSize)
 
 	framework.Server.SetQueryPlanCacheCap(64 * 1024)
 	_, _ = client.Execute("select * from vitess_test where intval=:ival1", bindVars)
-	time.Sleep(100 * time.Millisecond)
+	require.Equal(t, 2, framework.Server.QueryPlanCacheLen())
 
 	vend = framework.DebugVars()
 	verifyIntValue(t, vend, "QueryCacheLength", 2)
 	verifyIntValue(t, vend, "QueryCacheSize", cachedPlanSize*2)
+
 	_, _ = client.Execute("select * from vitess_test where intval=1", bindVars)
-	time.Sleep(100 * time.Millisecond)
+	require.Equal(t, 3, framework.Server.QueryPlanCacheLen())
 
 	vend = framework.DebugVars()
 	verifyIntValue(t, vend, "QueryCacheLength", 3)

go/vt/vttablet/endtoend/queries_test.go

@@ -18,6 +18,7 @@ package endtoend
 
 import (
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 
@@ -1783,6 +1784,10 @@ func TestQueries(t *testing.T) {
 			},
 		},
 	}
+
+	// Wait for the vtgate caches to flush
+	time.Sleep(1 * time.Second)
+
 	for _, tcase := range testCases {
 		if err := tcase.Test("", client); err != nil {
 			t.Error(err)

go/vt/vttablet/tabletserver/query_engine.go

@@ -413,6 +413,12 @@ func (qe *QueryEngine) QueryPlanCacheCap() int {
 	return int(qe.plans.MaxCapacity())
 }
 
+// QueryPlanCacheLen returns the length (size in entries) of the query cache
+func (qe *QueryEngine) QueryPlanCacheLen() int {
+	qe.plans.Wait()
+	return qe.plans.Len()
+}
+
 // AddStats adds the given stats for the planName.tableName
 func (qe *QueryEngine) AddStats(planName, tableName string, queryCount int64, duration, mysqlTime time.Duration, rowCount, errorCount int64) {
 	// table names can contain "." characters, replace them!