Rewrite sharde cache to generic one

This commit is contained in:
Michael Dresner 2022-03-18 22:19:19 +03:00
parent 9192b384ab
commit 0d5b08999a
No known key found for this signature in database
GPG Key ID: 039C3C305BAC5C23
2 changed files with 42 additions and 43 deletions

View File

@ -8,6 +8,8 @@ import (
"os" "os"
"runtime" "runtime"
"time" "time"
"golang.org/x/exp/constraints"
) )
// This is an experimental and unexported (for now) attempt at making a cache // This is an experimental and unexported (for now) attempt at making a cache
@ -19,15 +21,15 @@ import (
// //
// See cache_test.go for a few benchmarks. // See cache_test.go for a few benchmarks.
type unexportedShardedCache struct { type unexportedShardedCache[V constraints.Ordered] struct {
*shardedCache *shardedCache[V]
} }
type shardedCache struct { type shardedCache[V constraints.Ordered] struct {
seed uint32 seed uint32
m uint32 m uint32
cs []*cache cs []*orderedCache[string, V]
janitor *shardedJanitor janitor *shardedJanitor[V]
} }
// djb2 with better shuffling. 5x faster than FNV with the hash.Hash overhead. // djb2 with better shuffling. 5x faster than FNV with the hash.Hash overhead.
@ -62,43 +64,36 @@ func djb33(seed uint32, k string) uint32 {
return d ^ (d >> 16) return d ^ (d >> 16)
} }
func (sc *shardedCache) bucket(k string) *cache { func (sc *shardedCache[V]) bucket(k string) *orderedCache[string, V] {
return sc.cs[djb33(sc.seed, k)%sc.m] return sc.cs[djb33(sc.seed, k)%sc.m]
} }
func (sc *shardedCache) Set(k string, x interface{}, d time.Duration) { func (sc *shardedCache[V]) Set(k string, x V, d time.Duration) {
sc.bucket(k).Set(k, x, d) sc.bucket(k).Set(k, x, d)
} }
func (sc *shardedCache) Add(k string, x interface{}, d time.Duration) error { func (sc *shardedCache[V]) Add(k string, x V, d time.Duration) error {
return sc.bucket(k).Add(k, x, d) return sc.bucket(k).Add(k, x, d)
} }
func (sc *shardedCache) Replace(k string, x interface{}, d time.Duration) error { func (sc *shardedCache[V]) Replace(k string, x V, d time.Duration) error {
return sc.bucket(k).Replace(k, x, d) return sc.bucket(k).Replace(k, x, d)
} }
func (sc *shardedCache) Get(k string) (interface{}, bool) { func (sc *shardedCache[V]) Get(k string) (*V, bool) {
return sc.bucket(k).Get(k) return sc.bucket(k).Get(k)
} }
func (sc *shardedCache) Increment(k string, n int64) error { func (sc *shardedCache[V]) Increment(k string, n V) error {
return sc.bucket(k).Increment(k, n) _, err := sc.bucket(k).Increment(k, n)
return err
} }
func (sc *shardedCache) IncrementFloat(k string, n float64) error { func (sc *shardedCache[V]) Delete(k string) {
return sc.bucket(k).IncrementFloat(k, n)
}
func (sc *shardedCache) Decrement(k string, n int64) error {
return sc.bucket(k).Decrement(k, n)
}
func (sc *shardedCache) Delete(k string) {
sc.bucket(k).Delete(k) sc.bucket(k).Delete(k)
} }
func (sc *shardedCache) DeleteExpired() { func (sc *shardedCache[V]) DeleteExpired() {
for _, v := range sc.cs { for _, v := range sc.cs {
v.DeleteExpired() v.DeleteExpired()
} }
@ -109,26 +104,26 @@ func (sc *shardedCache) DeleteExpired() {
// fields of the items should be checked. Note that explicit synchronization // fields of the items should be checked. Note that explicit synchronization
// is needed to use a cache and its corresponding Items() return values at // is needed to use a cache and its corresponding Items() return values at
// the same time, as the maps are shared. // the same time, as the maps are shared.
func (sc *shardedCache) Items() []map[string]Item { func (sc *shardedCache[V]) Items() []map[string]Item[V] {
res := make([]map[string]Item, len(sc.cs)) res := make([]map[string]Item[V], len(sc.cs))
for i, v := range sc.cs { for i, v := range sc.cs {
res[i] = v.Items() res[i] = v.Items()
} }
return res return res
} }
func (sc *shardedCache) Flush() { func (sc *shardedCache[V]) Flush() {
for _, v := range sc.cs { for _, v := range sc.cs {
v.Flush() v.Flush()
} }
} }
type shardedJanitor struct { type shardedJanitor[V constraints.Ordered] struct {
Interval time.Duration Interval time.Duration
stop chan bool stop chan bool
} }
func (j *shardedJanitor) Run(sc *shardedCache) { func (j *shardedJanitor[V]) Run(sc *shardedCache[V]) {
j.stop = make(chan bool) j.stop = make(chan bool)
tick := time.Tick(j.Interval) tick := time.Tick(j.Interval)
for { for {
@ -141,19 +136,19 @@ func (j *shardedJanitor) Run(sc *shardedCache) {
} }
} }
func stopShardedJanitor(sc *unexportedShardedCache) { func stopShardedJanitor[V constraints.Ordered](sc *unexportedShardedCache[V]) {
sc.janitor.stop <- true sc.janitor.stop <- true
} }
func runShardedJanitor(sc *shardedCache, ci time.Duration) { func runShardedJanitor[V constraints.Ordered](sc *shardedCache[V], ci time.Duration) {
j := &shardedJanitor{ j := &shardedJanitor[V]{
Interval: ci, Interval: ci,
} }
sc.janitor = j sc.janitor = j
go j.Run(sc) go j.Run(sc)
} }
func newShardedCache(n int, de time.Duration) *shardedCache { func newShardedCache[V constraints.Ordered](n int, de time.Duration) *shardedCache[V] {
max := big.NewInt(0).SetUint64(uint64(math.MaxUint32)) max := big.NewInt(0).SetUint64(uint64(math.MaxUint32))
rnd, err := rand.Int(rand.Reader, max) rnd, err := rand.Int(rand.Reader, max)
var seed uint32 var seed uint32
@ -163,30 +158,34 @@ func newShardedCache(n int, de time.Duration) *shardedCache {
} else { } else {
seed = uint32(rnd.Uint64()) seed = uint32(rnd.Uint64())
} }
sc := &shardedCache{ sc := &shardedCache[V]{
seed: seed, seed: seed,
m: uint32(n), m: uint32(n),
cs: make([]*cache, n), cs: make([]*orderedCache[string, V], n),
} }
for i := 0; i < n; i++ { for i := 0; i < n; i++ {
c := &cache{ c := &orderedCache[string, V]{
defaultExpiration: de, Cache: &Cache[string, V]{
items: map[string]Item{}, cache: &cache[string, V]{
defaultExpiration: de,
items: map[string]Item[V]{},
},
},
} }
sc.cs[i] = c sc.cs[i] = c
} }
return sc return sc
} }
func unexportedNewSharded(defaultExpiration, cleanupInterval time.Duration, shards int) *unexportedShardedCache { func unexportedNewSharded[V constraints.Ordered](defaultExpiration, cleanupInterval time.Duration, shards int) *unexportedShardedCache[V] {
if defaultExpiration == 0 { if defaultExpiration == 0 {
defaultExpiration = -1 defaultExpiration = -1
} }
sc := newShardedCache(shards, defaultExpiration) sc := newShardedCache[V](shards, defaultExpiration)
SC := &unexportedShardedCache{sc} SC := &unexportedShardedCache[V]{sc}
if cleanupInterval > 0 { if cleanupInterval > 0 {
runShardedJanitor(sc, cleanupInterval) runShardedJanitor(sc, cleanupInterval)
runtime.SetFinalizer(SC, stopShardedJanitor) runtime.SetFinalizer(SC, stopShardedJanitor[V])
} }
return SC return SC
} }

View File

@ -27,7 +27,7 @@ var shardedKeys = []string{
} }
func TestShardedCache(t *testing.T) { func TestShardedCache(t *testing.T) {
tc := unexportedNewSharded(DefaultExpiration, 0, 13) tc := unexportedNewSharded[string](DefaultExpiration, 0, 13)
for _, v := range shardedKeys { for _, v := range shardedKeys {
tc.Set(v, "value", DefaultExpiration) tc.Set(v, "value", DefaultExpiration)
} }
@ -43,7 +43,7 @@ func BenchmarkShardedCacheGetNotExpiring(b *testing.B) {
func benchmarkShardedCacheGet(b *testing.B, exp time.Duration) { func benchmarkShardedCacheGet(b *testing.B, exp time.Duration) {
b.StopTimer() b.StopTimer()
tc := unexportedNewSharded(exp, 0, 10) tc := unexportedNewSharded[string](exp, 0, 10)
tc.Set("foobarba", "zquux", DefaultExpiration) tc.Set("foobarba", "zquux", DefaultExpiration)
b.StartTimer() b.StartTimer()
for i := 0; i < b.N; i++ { for i := 0; i < b.N; i++ {
@ -62,7 +62,7 @@ func BenchmarkShardedCacheGetManyConcurrentNotExpiring(b *testing.B) {
func benchmarkShardedCacheGetManyConcurrent(b *testing.B, exp time.Duration) { func benchmarkShardedCacheGetManyConcurrent(b *testing.B, exp time.Duration) {
b.StopTimer() b.StopTimer()
n := 10000 n := 10000
tsc := unexportedNewSharded(exp, 0, 20) tsc := unexportedNewSharded[string](exp, 0, 20)
keys := make([]string, n) keys := make([]string, n)
for i := 0; i < n; i++ { for i := 0; i < n; i++ {
k := "foo" + strconv.Itoa(i) k := "foo" + strconv.Itoa(i)