节点所属集群删除后，不再接收API请求

internal/utils/ttlcache/cache.go

@@ -0,0 +1,161 @@
+package ttlcache
+
+import (
+	"github.com/TeaOSLab/EdgeAPI/internal/utils"
+	"time"
+)
+
+var SharedCache = NewCache()
+
+// Cache TTL缓存
+// 最大的缓存时间为30 * 86400
+// Piece数据结构：
+//      Piece1            |  Piece2 | Piece3 | ...
+//  [ Item1, Item2, ... ] |   ...
+// KeyMap列表数据结构
+// { timestamp1 => [key1, key2, ...] }, ...
+type Cache struct {
+	isDestroyed bool
+	pieces      []*Piece
+	countPieces uint64
+	maxItems    int
+
+	gcPieceIndex int
+}
+
+func NewCache(opt ...OptionInterface) *Cache {
+	var countPieces = 256
+	var maxItems = 1_000_000
+
+	var totalMemory = utils.SystemMemoryGB()
+	if totalMemory < 2 {
+		// 我们限制内存过小的服务能够使用的数量
+		maxItems = 500_000
+	} else {
+		var delta = totalMemory / 8
+		if delta > 0 {
+			maxItems *= delta
+		}
+	}
+
+	for _, option := range opt {
+		if option == nil {
+			continue
+		}
+		switch o := option.(type) {
+		case *PiecesOption:
+			if o.Count > 0 {
+				countPieces = o.Count
+			}
+		case *MaxItemsOption:
+			if o.Count > 0 {
+				maxItems = o.Count
+			}
+		}
+	}
+
+	var cache = &Cache{
+		countPieces: uint64(countPieces),
+		maxItems:    maxItems,
+	}
+
+	for i := 0; i < countPieces; i++ {
+		cache.pieces = append(cache.pieces, NewPiece(maxItems/countPieces))
+	}
+
+	// Add to manager
+	SharedManager.Add(cache)
+
+	return cache
+}
+
+func (this *Cache) Write(key string, value interface{}, expiredAt int64) (ok bool) {
+	if this.isDestroyed {
+		return
+	}
+
+	var currentTimestamp = utils.UnixTime()
+	if expiredAt <= currentTimestamp {
+		return
+	}
+
+	var maxExpiredAt = currentTimestamp + 30*86400
+	if expiredAt > maxExpiredAt {
+		expiredAt = maxExpiredAt
+	}
+	uint64Key := HashKey([]byte(key))
+	pieceIndex := uint64Key % this.countPieces
+	return this.pieces[pieceIndex].Add(uint64Key, &Item{
+		Value:     value,
+		expiredAt: expiredAt,
+	})
+}
+
+func (this *Cache) IncreaseInt64(key string, delta int64, expiredAt int64, extend bool) int64 {
+	if this.isDestroyed {
+		return 0
+	}
+
+	currentTimestamp := time.Now().Unix()
+	if expiredAt <= currentTimestamp {
+		return 0
+	}
+
+	maxExpiredAt := currentTimestamp + 30*86400
+	if expiredAt > maxExpiredAt {
+		expiredAt = maxExpiredAt
+	}
+	uint64Key := HashKey([]byte(key))
+	pieceIndex := uint64Key % this.countPieces
+	return this.pieces[pieceIndex].IncreaseInt64(uint64Key, delta, expiredAt, extend)
+}
+
+func (this *Cache) Read(key string) (item *Item) {
+	uint64Key := HashKey([]byte(key))
+	return this.pieces[uint64Key%this.countPieces].Read(uint64Key)
+}
+
+func (this *Cache) readIntKey(key uint64) (value *Item) {
+	return this.pieces[key%this.countPieces].Read(key)
+}
+
+func (this *Cache) Delete(key string) {
+	uint64Key := HashKey([]byte(key))
+	this.pieces[uint64Key%this.countPieces].Delete(uint64Key)
+}
+
+func (this *Cache) deleteIntKey(key uint64) {
+	this.pieces[key%this.countPieces].Delete(key)
+}
+
+func (this *Cache) Count() (count int) {
+	for _, piece := range this.pieces {
+		count += piece.Count()
+	}
+	return
+}
+
+func (this *Cache) GC() {
+	this.pieces[this.gcPieceIndex].GC()
+	newIndex := this.gcPieceIndex + 1
+	if newIndex >= int(this.countPieces) {
+		newIndex = 0
+	}
+	this.gcPieceIndex = newIndex
+}
+
+func (this *Cache) Clean() {
+	for _, piece := range this.pieces {
+		piece.Clean()
+	}
+}
+
+func (this *Cache) Destroy() {
+	SharedManager.Remove(this)
+
+	this.isDestroyed = true
+
+	for _, piece := range this.pieces {
+		piece.Destroy()
+	}
+}

internal/utils/ttlcache/cache_test.go

@@ -0,0 +1,246 @@
+package ttlcache
+
+import (
+	"github.com/TeaOSLab/EdgeAPI/internal/utils"
+	"github.com/TeaOSLab/EdgeAPI/internal/utils/testutils"
+	"github.com/iwind/TeaGo/assert"
+	"github.com/iwind/TeaGo/rands"
+	"github.com/iwind/TeaGo/types"
+	"runtime"
+	"strconv"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestNewCache(t *testing.T) {
+	var cache = NewCache()
+	cache.Write("a", 1, time.Now().Unix()+3600)
+	cache.Write("b", 2, time.Now().Unix()+3601)
+	cache.Write("a", 1, time.Now().Unix()+3602)
+	cache.Write("d", 1, time.Now().Unix()+1)
+
+	for _, piece := range cache.pieces {
+		if len(piece.m) > 0 {
+			for k, item := range piece.m {
+				t.Log(k, "=>", item.Value, item.expiredAt)
+			}
+		}
+	}
+	t.Log("a:", cache.Read("a"))
+	time.Sleep(2 * time.Second)
+	t.Log("d:", cache.Read("d")) // should be nil
+	t.Log("left:", cache.Count(), "items")
+}
+
+func TestCache_Memory(t *testing.T) {
+	testutils.StartMemoryStats(t)
+
+	var cache = NewCache()
+	var count = 2_000_000
+	for i := 0; i < count; i++ {
+		cache.Write("a"+strconv.Itoa(i), 1, time.Now().Unix()+3600)
+	}
+
+	t.Log(cache.Count())
+
+	time.Sleep(10 * time.Second)
+	for i := 0; i < count; i++ {
+		if i%2 == 0 {
+			cache.Delete("a" + strconv.Itoa(i))
+		}
+	}
+
+	t.Log(cache.Count())
+
+	cache.Count()
+
+	time.Sleep(10 * time.Second)
+}
+
+func TestCache_IncreaseInt64(t *testing.T) {
+	var a = assert.NewAssertion(t)
+
+	var cache = NewCache()
+	var unixTime = time.Now().Unix()
+
+	{
+		cache.IncreaseInt64("a", 1, unixTime+3600, false)
+		var item = cache.Read("a")
+		t.Log(item)
+		a.IsTrue(item.Value == int64(1))
+		a.IsTrue(item.expiredAt == unixTime+3600)
+	}
+	{
+		cache.IncreaseInt64("a", 1, unixTime+3600+1, true)
+		var item = cache.Read("a")
+		t.Log(item)
+		a.IsTrue(item.Value == int64(2))
+		a.IsTrue(item.expiredAt == unixTime+3600+1)
+	}
+	{
+		cache.Write("b", 1, time.Now().Unix()+3600+2)
+		t.Log(cache.Read("b"))
+	}
+	{
+		cache.IncreaseInt64("b", 1, time.Now().Unix()+3600+3, false)
+		t.Log(cache.Read("b"))
+	}
+}
+
+func TestCache_Read(t *testing.T) {
+	runtime.GOMAXPROCS(1)
+
+	var cache = NewCache(PiecesOption{Count: 32})
+
+	for i := 0; i < 10_000_000; i++ {
+		cache.Write("HELLO_WORLD_"+strconv.Itoa(i), i, time.Now().Unix()+int64(i%10240)+1)
+	}
+	time.Sleep(10 * time.Second)
+
+	total := 0
+	for _, piece := range cache.pieces {
+		//t.Log(len(piece.m), "keys")
+		total += len(piece.m)
+	}
+	t.Log(total, "total keys")
+
+	before := time.Now()
+	for i := 0; i < 10_240; i++ {
+		_ = cache.Read("HELLO_WORLD_" + strconv.Itoa(i))
+	}
+	t.Log(time.Since(before).Seconds()*1000, "ms")
+}
+
+func TestCache_GC(t *testing.T) {
+	var cache = NewCache(&PiecesOption{Count: 5})
+	cache.Write("a", 1, time.Now().Unix()+1)
+	cache.Write("b", 2, time.Now().Unix()+2)
+	cache.Write("c", 3, time.Now().Unix()+3)
+	cache.Write("d", 4, time.Now().Unix()+4)
+	cache.Write("e", 5, time.Now().Unix()+10)
+
+	go func() {
+		for i := 0; i < 1000; i++ {
+			cache.Write("f", 1, time.Now().Unix()+1)
+			time.Sleep(10 * time.Millisecond)
+		}
+	}()
+
+	for i := 0; i < 20; i++ {
+		cache.GC()
+		t.Log("items:", cache.Count())
+
+		if cache.Count() == 0 {
+			break
+		}
+		time.Sleep(1 * time.Second)
+	}
+
+	t.Log("now:", time.Now().Unix())
+	for _, p := range cache.pieces {
+		t.Log("expire list:", p.expiresList.Count(), p.expiresList)
+		for k, v := range p.m {
+			t.Log(k, v.Value, v.expiredAt)
+		}
+	}
+}
+
+func TestCache_GC2(t *testing.T) {
+	runtime.GOMAXPROCS(1)
+
+	var cache1 = NewCache(NewPiecesOption(32))
+	for i := 0; i < 1_000_000; i++ {
+		cache1.Write(strconv.Itoa(i), i, time.Now().Unix()+int64(rands.Int(0, 10)))
+	}
+
+	var cache2 = NewCache(NewPiecesOption(5))
+	for i := 0; i < 1_000_000; i++ {
+		cache2.Write(strconv.Itoa(i), i, time.Now().Unix()+int64(rands.Int(0, 10)))
+	}
+
+	for i := 0; i < 100; i++ {
+		t.Log(cache1.Count(), "items", cache2.Count(), "items")
+		time.Sleep(1 * time.Second)
+	}
+}
+
+func TestCacheDestroy(t *testing.T) {
+	var cache = NewCache()
+	t.Log("count:", SharedManager.Count())
+	cache.Destroy()
+	t.Log("count:", SharedManager.Count())
+}
+
+func BenchmarkNewCache(b *testing.B) {
+	runtime.GOMAXPROCS(1)
+
+	var cache = NewCache(NewPiecesOption(128))
+	for i := 0; i < 2_000_000; i++ {
+		cache.Write(strconv.Itoa(i), i, time.Now().Unix()+int64(rands.Int(10, 100)))
+	}
+	b.Log("start reading ...")
+
+	b.ResetTimer()
+
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			cache.Read(strconv.Itoa(rands.Int(0, 999999)))
+		}
+	})
+}
+
+func BenchmarkCache_Add(b *testing.B) {
+	runtime.GOMAXPROCS(1)
+
+	var cache = NewCache()
+	for i := 0; i < b.N; i++ {
+		cache.Write(strconv.Itoa(i), i, utils.UnixTime()+int64(i%1024))
+	}
+}
+
+func BenchmarkCache_Add_Parallel(b *testing.B) {
+	runtime.GOMAXPROCS(1)
+
+	var cache = NewCache()
+	var i int64
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			var j = atomic.AddInt64(&i, 1)
+			cache.Write(types.String(j), j, utils.UnixTime()+i%1024)
+		}
+	})
+}
+
+func BenchmarkNewCacheGC(b *testing.B) {
+	runtime.GOMAXPROCS(1)
+
+	var cache = NewCache(NewPiecesOption(1024))
+	for i := 0; i < 3_000_000; i++ {
+		cache.Write(strconv.Itoa(i), i, time.Now().Unix()+int64(rands.Int(0, 100)))
+	}
+	//b.Log(cache.pieces[0].Count())
+
+	b.ResetTimer()
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			cache.GC()
+		}
+	})
+}
+
+func BenchmarkNewCacheClean(b *testing.B) {
+	runtime.GOMAXPROCS(1)
+
+	var cache = NewCache(NewPiecesOption(128))
+	for i := 0; i < 3_000_000; i++ {
+		cache.Write(strconv.Itoa(i), i, time.Now().Unix()+int64(rands.Int(10, 100)))
+	}
+
+	b.ResetTimer()
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			cache.Clean()
+		}
+	})
+}

internal/utils/ttlcache/item.go

@@ -0,0 +1,6 @@
+package ttlcache
+
+type Item struct {
+	Value     interface{}
+	expiredAt int64
+}

internal/utils/ttlcache/manager.go

@@ -0,0 +1,60 @@
+// Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
+
+package ttlcache
+
+import (
+	"github.com/TeaOSLab/EdgeAPI/internal/goman"
+	"github.com/TeaOSLab/EdgeAPI/internal/zero"
+	"sync"
+	"time"
+)
+
+var SharedManager = NewManager()
+
+type Manager struct {
+	ticker *time.Ticker
+	locker sync.Mutex
+
+	cacheMap map[*Cache]zero.Zero
+}
+
+func NewManager() *Manager {
+	var manager = &Manager{
+		ticker:   time.NewTicker(2 * time.Second),
+		cacheMap: map[*Cache]zero.Zero{},
+	}
+
+	goman.New(func() {
+		manager.init()
+	})
+
+	return manager
+}
+
+func (this *Manager) init() {
+	for range this.ticker.C {
+		this.locker.Lock()
+		for cache := range this.cacheMap {
+			cache.GC()
+		}
+		this.locker.Unlock()
+	}
+}
+
+func (this *Manager) Add(cache *Cache) {
+	this.locker.Lock()
+	this.cacheMap[cache] = zero.New()
+	this.locker.Unlock()
+}
+
+func (this *Manager) Remove(cache *Cache) {
+	this.locker.Lock()
+	delete(this.cacheMap, cache)
+	this.locker.Unlock()
+}
+
+func (this *Manager) Count() int {
+	this.locker.Lock()
+	defer this.locker.Unlock()
+	return len(this.cacheMap)
+}

internal/utils/ttlcache/option.go

@@ -0,0 +1,20 @@
+package ttlcache
+
+type OptionInterface interface {
+}
+
+type PiecesOption struct {
+	Count int
+}
+
+func NewPiecesOption(count int) *PiecesOption {
+	return &PiecesOption{Count: count}
+}
+
+type MaxItemsOption struct {
+	Count int
+}
+
+func NewMaxItemsOption(count int) *MaxItemsOption {
+	return &MaxItemsOption{Count: count}
+}

internal/utils/ttlcache/piece.go

@@ -0,0 +1,138 @@
+package ttlcache
+
+import (
+	"github.com/TeaOSLab/EdgeAPI/internal/utils"
+	"github.com/TeaOSLab/EdgeAPI/internal/utils/expires"
+	"github.com/iwind/TeaGo/types"
+	"sync"
+	"time"
+)
+
+type Piece struct {
+	m           map[uint64]*Item
+	expiresList *expires.List
+	maxItems    int
+	lastGCTime  int64
+
+	locker sync.RWMutex
+}
+
+func NewPiece(maxItems int) *Piece {
+	return &Piece{
+		m:           map[uint64]*Item{},
+		expiresList: expires.NewSingletonList(),
+		maxItems:    maxItems,
+	}
+}
+
+func (this *Piece) Add(key uint64, item *Item) (ok bool) {
+	this.locker.Lock()
+	if len(this.m) >= this.maxItems {
+		this.locker.Unlock()
+		return
+	}
+	this.m[key] = item
+	this.locker.Unlock()
+
+	this.expiresList.Add(key, item.expiredAt)
+
+	return true
+}
+
+func (this *Piece) IncreaseInt64(key uint64, delta int64, expiredAt int64, extend bool) (result int64) {
+	this.locker.Lock()
+	item, ok := this.m[key]
+	if ok && item.expiredAt > time.Now().Unix() {
+		result = types.Int64(item.Value) + delta
+		item.Value = result
+		if extend {
+			item.expiredAt = expiredAt
+		}
+		this.expiresList.Add(key, expiredAt)
+	} else {
+		if len(this.m) < this.maxItems {
+			result = delta
+			this.m[key] = &Item{
+				Value:     delta,
+				expiredAt: expiredAt,
+			}
+			this.expiresList.Add(key, expiredAt)
+		}
+	}
+	this.locker.Unlock()
+
+	return
+}
+
+func (this *Piece) Delete(key uint64) {
+	this.expiresList.Remove(key)
+
+	this.locker.Lock()
+	delete(this.m, key)
+	this.locker.Unlock()
+}
+
+func (this *Piece) Read(key uint64) (item *Item) {
+	this.locker.RLock()
+	item = this.m[key]
+	if item != nil && item.expiredAt < utils.UnixTime() {
+		item = nil
+	}
+	this.locker.RUnlock()
+
+	return
+}
+
+func (this *Piece) Count() (count int) {
+	this.locker.RLock()
+	count = len(this.m)
+	this.locker.RUnlock()
+	return
+}
+
+func (this *Piece) GC() {
+	var currentTime = time.Now().Unix()
+	if this.lastGCTime == 0 {
+		this.lastGCTime = currentTime - 3600
+	}
+
+	var min = this.lastGCTime
+	var max = currentTime
+	if min > max {
+		// 过去的时间比现在大，则从这一秒重新开始
+		min = max
+	}
+
+	for i := min; i <= max; i++ {
+		var itemMap = this.expiresList.GC(i)
+		if len(itemMap) > 0 {
+			this.gcItemMap(itemMap)
+		}
+	}
+
+	this.lastGCTime = currentTime
+}
+
+func (this *Piece) Clean() {
+	this.locker.Lock()
+	this.m = map[uint64]*Item{}
+	this.locker.Unlock()
+
+	this.expiresList.Clean()
+}
+
+func (this *Piece) Destroy() {
+	this.locker.Lock()
+	this.m = nil
+	this.locker.Unlock()
+
+	this.expiresList.Clean()
+}
+
+func (this *Piece) gcItemMap(itemMap expires.ItemMap) {
+	this.locker.Lock()
+	for key := range itemMap {
+		delete(this.m, key)
+	}
+	this.locker.Unlock()
+}

internal/utils/ttlcache/piece_test.go

@@ -0,0 +1,60 @@
+package ttlcache
+
+import (
+	"github.com/iwind/TeaGo/rands"
+	"testing"
+	"time"
+)
+
+func TestPiece_Add(t *testing.T) {
+	piece := NewPiece(10)
+	piece.Add(1, &Item{expiredAt: time.Now().Unix() + 3600})
+	piece.Add(2, &Item{})
+	piece.Add(3, &Item{})
+	piece.Delete(3)
+	for key, item := range piece.m {
+		t.Log(key, item.Value)
+	}
+	t.Log(piece.Read(1))
+}
+
+func TestPiece_MaxItems(t *testing.T) {
+	piece := NewPiece(10)
+	for i := 0; i < 1000; i++ {
+		piece.Add(uint64(i), &Item{expiredAt: time.Now().Unix() + 3600})
+	}
+	t.Log(len(piece.m))
+}
+
+func TestPiece_GC(t *testing.T) {
+	piece := NewPiece(10)
+	piece.Add(1, &Item{Value: 1, expiredAt: time.Now().Unix() + 1})
+	piece.Add(2, &Item{Value: 2, expiredAt: time.Now().Unix() + 1})
+	piece.Add(3, &Item{Value: 3, expiredAt: time.Now().Unix() + 1})
+	t.Log("before gc ===")
+	for key, item := range piece.m {
+		t.Log(key, item.Value)
+	}
+
+	time.Sleep(1 * time.Second)
+	piece.GC()
+
+	t.Log("after gc ===")
+	for key, item := range piece.m {
+		t.Log(key, item.Value)
+	}
+}
+
+func TestPiece_GC2(t *testing.T) {
+	piece := NewPiece(10)
+	for i := 0; i < 10_000; i++ {
+		piece.Add(uint64(i), &Item{Value: 1, expiredAt: time.Now().Unix() + int64(rands.Int(1, 10))})
+	}
+
+	time.Sleep(1 * time.Second)
+
+	before := time.Now()
+	piece.GC()
+	t.Log(time.Since(before).Seconds()*1000, "ms")
+	t.Log(piece.Count())
+}

internal/utils/ttlcache/utils.go

@@ -0,0 +1,7 @@
+package ttlcache
+
+import "github.com/cespare/xxhash"
+
+func HashKey(key []byte) uint64 {
+	return xxhash.Sum64(key)
+}

internal/utils/ttlcache/utils_test.go

@@ -0,0 +1,13 @@
+package ttlcache
+
+import (
+	"runtime"
+	"testing"
+)
+
+func BenchmarkHashKey(b *testing.B) {
+	runtime.GOMAXPROCS(1)
+	for i := 0; i < b.N; i++ {
+		HashKey([]byte("HELLO,WORLDHELLO,WORLDHELLO,WORLDHELLO,WORLDHELLO,WORLDHELLO,WORLD"))
+	}
+}