add local cache

kit/localcache/README.md

@@ -0,0 +1,2 @@
+# go-localcache
+in-process cache writen in go and managed by timingwheel

kit/localcache/internal/atomic.go

@@ -0,0 +1,39 @@
+package internal
+
+import "sync/atomic"
+
+type AtomicBool uint32
+
+func NewAtomicBool() *AtomicBool {
+	return new(AtomicBool)
+}
+
+func ForAtomicBool(val bool) *AtomicBool {
+	ab := NewAtomicBool()
+	ab.Set(val)
+	return ab
+}
+
+func (a *AtomicBool) CompareAndSwap(old, new bool) bool {
+	var oldV, newV uint32
+	if old {
+		oldV = 1
+	}
+	if new {
+		newV = 1
+	}
+
+	return atomic.CompareAndSwapUint32((*uint32)(a), oldV, newV)
+}
+
+func (a *AtomicBool) Set(new bool) {
+	if new {
+		atomic.StoreUint32((*uint32)(a), 1)
+	} else {
+		atomic.StoreUint32((*uint32)(a), 0)
+	}
+}
+
+func (a *AtomicBool) True() bool {
+	return atomic.LoadUint32((*uint32)(a)) == 1
+}

kit/localcache/internal/lru.go

@@ -0,0 +1,82 @@
+package internal
+
+import "container/list"
+
+type (
+	Lru interface {
+		Add(key string)
+		Remove(key string)
+	}
+
+	noneLru struct{}
+
+	keyLru struct {
+		limit    int
+		evicts   *list.List
+		elements map[string]*list.Element
+		onEvict  func(key string)
+	}
+)
+
+var (
+	_ Lru = &noneLru{}
+	_ Lru = &keyLru{}
+)
+
+// NewNoneLru return an empty lru implement, do not manager keys.
+// when cache have a limit of count, use this to make the flow correct
+func NewNoneLru() Lru {
+	return &noneLru{}
+}
+
+func (l *noneLru) Add(key string) {}
+
+func (l *noneLru) Remove(key string) {}
+
+// NewLru return a Lru entry with least-recently-use algorithm
+func NewLru(limit int, onEvict func(key string)) Lru {
+	return &keyLru{
+		limit:    limit,
+		evicts:   list.New(),
+		elements: make(map[string]*list.Element),
+		onEvict:  onEvict,
+	}
+}
+
+func (l *keyLru) Remove(key string) {
+	if elem, ok := l.elements[key]; ok {
+		l.removeElem(elem)
+	}
+}
+
+func (l *keyLru) Add(key string) {
+	if v, ok := l.elements[key]; ok {
+		// 元素存在, 移至队首
+		l.evicts.MoveToFront(v)
+		return
+	}
+
+	// 新增元素
+	elem := l.evicts.PushFront(key)
+	l.elements[key] = elem
+
+	// 超出列表长度, 移除队尾元素
+	if l.evicts.Len() > l.limit {
+		l.removeOldest()
+	}
+}
+
+func (l *keyLru) removeOldest() {
+	elem := l.evicts.Back()
+	l.removeElem(elem)
+}
+
+func (l *keyLru) removeElem(e *list.Element) {
+	if e == nil {
+		return
+	}
+	l.evicts.Remove(e)
+	key := e.Value.(string)
+	delete(l.elements, key)
+	l.onEvict(key)
+}

kit/localcache/internal/safemap.go

@@ -0,0 +1,98 @@
+package internal
+
+import (
+	"sync"
+)
+
+const (
+	copyThreshold = 1000
+	maxDeletion   = 10000
+)
+
+// SafeMap provides a map alternative to avoid memory leak.
+// This implementation is not needed until issue below fixed.
+// https://github.com/golang/go/issues/20135
+type SafeMap struct {
+	lock        sync.RWMutex
+	deletionOld int
+	deletionNew int
+	dirtyOld    map[interface{}]interface{}
+	dirtyNew    map[interface{}]interface{}
+}
+
+// NewSafeMap returns a SafeMap.
+func NewSafeMap() *SafeMap {
+	return &SafeMap{
+		dirtyOld: make(map[interface{}]interface{}),
+		dirtyNew: make(map[interface{}]interface{}),
+	}
+}
+
+// Del deletes the value with the given key from m.
+func (m *SafeMap) Del(key interface{}) {
+	m.lock.Lock()
+	if _, ok := m.dirtyOld[key]; ok {
+		delete(m.dirtyOld, key)
+		m.deletionOld++
+	} else if _, ok := m.dirtyNew[key]; ok {
+		delete(m.dirtyNew, key)
+		m.deletionNew++
+	}
+	if m.deletionOld >= maxDeletion && len(m.dirtyOld) < copyThreshold {
+		for k, v := range m.dirtyOld {
+			m.dirtyNew[k] = v
+		}
+		m.dirtyOld = m.dirtyNew
+		m.deletionOld = m.deletionNew
+		m.dirtyNew = make(map[interface{}]interface{})
+		m.deletionNew = 0
+	}
+	if m.deletionNew >= maxDeletion && len(m.dirtyNew) < copyThreshold {
+		for k, v := range m.dirtyNew {
+			m.dirtyOld[k] = v
+		}
+		m.dirtyNew = make(map[interface{}]interface{})
+		m.deletionNew = 0
+	}
+	m.lock.Unlock()
+}
+
+// Get gets the value with the given key from m.
+func (m *SafeMap) Get(key interface{}) (interface{}, bool) {
+	m.lock.RLock()
+	defer m.lock.RUnlock()
+
+	if val, ok := m.dirtyOld[key]; ok {
+		return val, true
+	}
+
+	val, ok := m.dirtyNew[key]
+	return val, ok
+}
+
+// Set sets the value into m with the given key.
+func (m *SafeMap) Set(key, value interface{}) {
+	m.lock.Lock()
+	if m.deletionOld <= maxDeletion {
+		if _, ok := m.dirtyNew[key]; ok {
+			delete(m.dirtyNew, key)
+			m.deletionNew++
+		}
+		m.dirtyOld[key] = value
+	} else {
+		if _, ok := m.dirtyOld[key]; ok {
+			delete(m.dirtyOld, key)
+			m.deletionOld++
+		}
+		m.dirtyNew[key] = value
+	}
+	m.lock.Unlock()
+}
+
+// Size returns the size of m.
+func (m *SafeMap) Size() int {
+	m.lock.RLock()
+	size := len(m.dirtyOld) + len(m.dirtyNew)
+	m.lock.RUnlock()
+	return size
+}

kit/localcache/internal/stat.go

@@ -0,0 +1,63 @@
+package internal
+
+import (
+	"log"
+	"sync/atomic"
+	"time"
+)
+
+//const statInterval = time.Minute
+const statInterval = time.Second
+
+type Stat struct {
+	name         string
+	hit          uint64
+	miss         uint64
+	sizeCallback func() int
+}
+
+func NewStat(name string, sizeCallback func() int) *Stat {
+	st := &Stat{
+		name:         name,
+		sizeCallback: sizeCallback,
+	}
+	go st.report()
+
+	return st
+}
+
+// Hit hit counter++
+func (s *Stat) Hit() {
+	atomic.AddUint64(&s.hit, 1)
+}
+
+// Miss missed counter++
+func (s *Stat) Miss() {
+	atomic.AddUint64(&s.miss, 1)
+}
+
+// CurrentMinute return hit and missed counter in a minute
+func (s *Stat) CurrentMinute() (hit, miss uint64) {
+	hit = atomic.LoadUint64(&s.hit)
+	miss = atomic.LoadUint64(&s.miss)
+	return hit, miss
+}
+
+func (s *Stat) report() {
+	ticker := time.NewTicker(statInterval)
+	defer ticker.Stop()
+
+	for range ticker.C {
+		hit := atomic.SwapUint64(&s.hit, 0)
+		miss := atomic.SwapUint64(&s.miss, 0)
+		total := hit + miss
+		if total == 0 {
+			log.Printf("cache(%s) - continue", s.name)
+			continue
+		}
+
+		percent := float32(hit) / float32(total)
+		log.Printf("cache(%s) - qpm: %d, hit_ratio: %.1f%%, elements: %d, hit: %d, miss: %d",
+			s.name, total, percent*100, s.sizeCallback(), hit, miss)
+	}
+}

kit/localcache/internal/ticker.go

@@ -0,0 +1,73 @@
+package internal
+
+import (
+	"errors"
+	"time"
+)
+
+type (
+	Ticker interface {
+		Chan() <-chan time.Time
+		Stop()
+	}
+
+	// FakeTicker for test
+	FakeTicker interface {
+		Ticker
+		Done()
+		Tick()
+		Wait(d time.Duration) error
+	}
+
+	fakeTicker struct {
+		c    chan time.Time
+		done chan struct{}
+	}
+
+	realTicker struct {
+		*time.Ticker
+	}
+)
+
+func NewTicker(d time.Duration) Ticker {
+	return &realTicker{
+		Ticker: time.NewTicker(d),
+	}
+}
+
+// Chan implement Ticker
+func (r *realTicker) Chan() <-chan time.Time {
+	return r.C
+}
+
+func NewFakeTicker() FakeTicker {
+	return &fakeTicker{
+		c:    make(chan time.Time, 1),
+		done: make(chan struct{}, 1),
+	}
+}
+
+func (f *fakeTicker) Chan() <-chan time.Time {
+	return f.c
+}
+
+func (f *fakeTicker) Stop() {
+	close(f.c)
+}
+
+func (f *fakeTicker) Done() {
+	f.done <- struct{}{}
+}
+
+func (f *fakeTicker) Tick() {
+	f.c <- time.Now()
+}
+
+func (f *fakeTicker) Wait(d time.Duration) error {
+	select {
+	case <-time.After(d):
+		return errors.New("timeout")
+	case <-f.done:
+		return nil
+	}
+}