优化过期列表管理

internal/utils/expires/id_key_map.go

@@ -58,3 +58,11 @@ func (this *IdKeyMap) DeleteKey(key string) {
 func (this *IdKeyMap) Len() int {
 	return len(this.idKeys)
 }
+
+func (this *IdKeyMap) IdKeys() map[int64]string {
+	return this.idKeys
+}
+
+func (this *IdKeyMap) KeyIds() map[string]int64 {
+	return this.keyIds
+}

internal/utils/expires/id_key_map_test.go

@@ -1,8 +1,9 @@
 // Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
 
-package expires
+package expires_test
 
 import (
+	"github.com/TeaOSLab/EdgeNode/internal/utils/expires"
 	"github.com/iwind/TeaGo/assert"
 	"github.com/iwind/TeaGo/logs"
 	"testing"
@@ -11,12 +12,12 @@ import (
 func TestNewIdKeyMap(t *testing.T) {
 	var a = assert.NewAssertion(t)
 
-	var m = NewIdKeyMap()
+	var m = expires.NewIdKeyMap()
 	m.Add(1, "1")
 	m.Add(1, "2")
 	m.Add(100, "100")
-	logs.PrintAsJSON(m.idKeys, t)
+	logs.PrintAsJSON(m.IdKeys(), t)
-	logs.PrintAsJSON(m.keyIds, t)
+	logs.PrintAsJSON(m.KeyIds(), t)
 
 	{
 		k, ok := m.Key(1)
@@ -36,11 +37,11 @@ func TestNewIdKeyMap(t *testing.T) {
 		a.IsFalse(ok)
 	}
 
-	logs.PrintAsJSON(m.idKeys, t)
+	logs.PrintAsJSON(m.IdKeys(), t)
-	logs.PrintAsJSON(m.keyIds, t)
+	logs.PrintAsJSON(m.KeyIds(), t)
 
 	m.DeleteId(100)
 
-	logs.PrintAsJSON(m.idKeys, t)
+	logs.PrintAsJSON(m.IdKeys(), t)
-	logs.PrintAsJSON(m.keyIds, t)
+	logs.PrintAsJSON(m.KeyIds(), t)
 }

internal/utils/expires/list.go

@@ -11,7 +11,7 @@ type List struct {
 	expireMap map[int64]ItemMap // expires timestamp => map[id]ItemMap
 	itemsMap  map[uint64]int64  // itemId => timestamp
 
-	locker sync.Mutex
+	mu sync.RWMutex
 
 	gcCallback      func(itemId uint64)
 	gcBatchCallback func(itemIds ItemMap)
@@ -42,8 +42,8 @@ func NewSingletonList() *List {
 // Add 添加条目
 // 如果条目已经存在，则覆盖
 func (this *List) Add(itemId uint64, expiresAt int64) {
-	this.locker.Lock()
+	this.mu.Lock()
-	defer this.locker.Unlock()
+	defer this.mu.Unlock()
 
 	if this.lastTimestamp == 0 || this.lastTimestamp > expiresAt {
 		this.lastTimestamp = expiresAt
@@ -72,14 +72,14 @@ func (this *List) Add(itemId uint64, expiresAt int64) {
 }
 
 func (this *List) Remove(itemId uint64) {
-	this.locker.Lock()
+	this.mu.Lock()
-	defer this.locker.Unlock()
+	defer this.mu.Unlock()
 	this.removeItem(itemId)
 }
 
 func (this *List) ExpiresAt(itemId uint64) int64 {
-	this.locker.Lock()
+	this.mu.RLock()
-	defer this.locker.Unlock()
+	defer this.mu.RUnlock()
 	return this.itemsMap[itemId]
 }
 
@@ -87,13 +87,10 @@ func (this *List) GC(timestamp int64) ItemMap {
 	if this.lastTimestamp > timestamp+1 {
 		return nil
 	}
-	this.locker.Lock()
 	var itemMap = this.gcItems(timestamp)
 	if len(itemMap) == 0 {
-		this.locker.Unlock()
 		return itemMap
 	}
-	this.locker.Unlock()
 
 	if this.gcCallback != nil {
 		for itemId := range itemMap {
@@ -108,16 +105,16 @@ func (this *List) GC(timestamp int64) ItemMap {
 }
 
 func (this *List) Clean() {
-	this.locker.Lock()
+	this.mu.Lock()
 	this.itemsMap = map[uint64]int64{}
 	this.expireMap = map[int64]ItemMap{}
-	this.locker.Unlock()
+	this.mu.Unlock()
 }
 
 func (this *List) Count() int {
-	this.locker.Lock()
+	this.mu.RLock()
 	var count = len(this.itemsMap)
-	this.locker.Unlock()
+	this.mu.RUnlock()
 	return count
 }
 
@@ -131,6 +128,18 @@ func (this *List) OnGCBatch(callback func(itemMap ItemMap)) *List {
 	return this
 }
 
+func (this *List) ExpireMap() map[int64]ItemMap {
+	return this.expireMap
+}
+
+func (this *List) ItemsMap() map[uint64]int64 {
+	return this.itemsMap
+}
+
+func (this *List) LastTimestamp() int64 {
+	return this.lastTimestamp
+}
+
 func (this *List) removeItem(itemId uint64) {
 	expiresAt, ok := this.itemsMap[itemId]
 	if !ok {
@@ -148,12 +157,18 @@ func (this *List) removeItem(itemId uint64) {
 }
 
 func (this *List) gcItems(timestamp int64) ItemMap {
+	this.mu.RLock()
 	expireItemsMap, ok := this.expireMap[timestamp]
+	this.mu.RUnlock()
+
 	if ok {
+		this.mu.Lock()
 		for itemId := range expireItemsMap {
 			delete(this.itemsMap, itemId)
 		}
 		delete(this.expireMap, timestamp)
+		this.mu.Unlock()
 	}
+
 	return expireItemsMap
 }

internal/utils/expires/list_test.go

@@ -1,87 +1,89 @@
-package expires
+package expires_test
 
 import (
+	"github.com/TeaOSLab/EdgeNode/internal/utils/expires"
 	"github.com/TeaOSLab/EdgeNode/internal/utils/fasttime"
 	"github.com/TeaOSLab/EdgeNode/internal/utils/testutils"
 	"github.com/iwind/TeaGo/assert"
 	"github.com/iwind/TeaGo/logs"
 	timeutil "github.com/iwind/TeaGo/utils/time"
 	"math"
+	"math/rand"
 	"runtime"
 	"testing"
 	"time"
 )
 
 func TestList_Add(t *testing.T) {
-	list := NewList()
+	var list = expires.NewList()
 	list.Add(1, time.Now().Unix())
 	t.Log("===BEFORE===")
-	logs.PrintAsJSON(list.expireMap, t)
+	logs.PrintAsJSON(list.ExpireMap(), t)
-	logs.PrintAsJSON(list.itemsMap, t)
+	logs.PrintAsJSON(list.ItemsMap(), t)
 
 	list.Add(1, time.Now().Unix()+1)
 	list.Add(2, time.Now().Unix()+1)
 	list.Add(3, time.Now().Unix()+2)
 	t.Log("===AFTER===")
-	logs.PrintAsJSON(list.expireMap, t)
+	logs.PrintAsJSON(list.ExpireMap(), t)
-	logs.PrintAsJSON(list.itemsMap, t)
+	logs.PrintAsJSON(list.ItemsMap(), t)
 }
 
 func TestList_Add_Overwrite(t *testing.T) {
 	var timestamp = time.Now().Unix()
 
-	list := NewList()
+	var list = expires.NewList()
 	list.Add(1, timestamp+1)
 	list.Add(1, timestamp+1)
 	list.Add(1, timestamp+2)
-	logs.PrintAsJSON(list.expireMap, t)
+	logs.PrintAsJSON(list.ExpireMap(), t)
-	logs.PrintAsJSON(list.itemsMap, t)
+	logs.PrintAsJSON(list.ItemsMap(), t)
 
 	var a = assert.NewAssertion(t)
-	a.IsTrue(len(list.itemsMap) == 1)
+	a.IsTrue(len(list.ItemsMap()) == 1)
-	a.IsTrue(len(list.expireMap) == 1)
+	a.IsTrue(len(list.ExpireMap()) == 1)
-	a.IsTrue(list.itemsMap[1] == timestamp+2)
+	a.IsTrue(list.ItemsMap()[1] == timestamp+2)
 }
 
 func TestList_Remove(t *testing.T) {
-	list := NewList()
+	var list = expires.NewList()
 	list.Add(1, time.Now().Unix()+1)
 	list.Remove(1)
-	logs.PrintAsJSON(list.expireMap, t)
+	logs.PrintAsJSON(list.ExpireMap(), t)
-	logs.PrintAsJSON(list.itemsMap, t)
+	logs.PrintAsJSON(list.ItemsMap(), t)
 }
 
 func TestList_GC(t *testing.T) {
 	var unixTime = time.Now().Unix()
 	t.Log("unixTime:", unixTime)
 
-	var list = NewList()
+	var list = expires.NewList()
 	list.Add(1, unixTime+1)
 	list.Add(2, unixTime+1)
 	list.Add(3, unixTime+2)
 	list.OnGC(func(itemId uint64) {
 		t.Log("gc:", itemId)
 	})
-	t.Log("last unixTime:", list.lastTimestamp)
+	t.Log("last unixTime:", list.LastTimestamp())
 	list.GC(time.Now().Unix() + 2)
-	logs.PrintAsJSON(list.expireMap, t)
+	logs.PrintAsJSON(list.ExpireMap(), t)
-	logs.PrintAsJSON(list.itemsMap, t)
+	logs.PrintAsJSON(list.ItemsMap(), t)
 
 	t.Log(list.Count())
 }
 
 func TestList_GC_Batch(t *testing.T) {
-	list := NewList()
+	var list = expires.NewList()
 	list.Add(1, time.Now().Unix()+1)
 	list.Add(2, time.Now().Unix()+1)
 	list.Add(3, time.Now().Unix()+2)
 	list.Add(4, time.Now().Unix()+2)
-	list.OnGCBatch(func(itemMap ItemMap) {
+	list.OnGCBatch(func(itemMap expires.ItemMap) {
 		t.Log("gc:", itemMap)
 	})
 	list.GC(time.Now().Unix() + 2)
-	logs.PrintAsJSON(list.expireMap, t)
+	logs.PrintAsJSON(list.ExpireMap(), t)
-	logs.PrintAsJSON(list.itemsMap, t)
+	logs.PrintAsJSON(list.ItemsMap(), t)
 }
 
 func TestList_Start_GC(t *testing.T) {
@@ -89,7 +91,7 @@ func TestList_Start_GC(t *testing.T) {
 		return
 	}
 
-	list := NewList()
+	var list = expires.NewList()
 	list.Add(1, time.Now().Unix()+1)
 	list.Add(2, time.Now().Unix()+1)
 	list.Add(3, time.Now().Unix()+2)
@@ -105,14 +107,14 @@ func TestList_Start_GC(t *testing.T) {
 	})
 
 	go func() {
-		SharedManager.Add(list)
+		expires.SharedManager.Add(list)
 	}()
 
 	time.Sleep(20 * time.Second)
 }
 
 func TestList_ManyItems(t *testing.T) {
-	list := NewList()
+	var list = expires.NewList()
 	for i := 0; i < 1_000; i++ {
 		list.Add(uint64(i), time.Now().Unix())
 	}
@@ -135,13 +137,12 @@ func TestList_Memory(t *testing.T) {
 		return
 	}
 
-	var list = NewList()
+	var list = expires.NewList()
 
 	testutils.StartMemoryStats(t, func() {
 		t.Log(list.Count(), "items")
 	})
 
-
 	for i := 0; i < 10_000_000; i++ {
 		list.Add(uint64(i), time.Now().Unix()+1800)
 	}
@@ -194,6 +195,15 @@ func TestList_Map_Performance(t *testing.T) {
 	}
 }
 
+func BenchmarkList_Add(b *testing.B) {
+	var list = expires.NewList()
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			list.Add(rand.Uint64(), fasttime.Now().Unix()+int64(rand.Int()%10_000_000))
+		}
+	})
+}
+
 func Benchmark_Map_Uint64(b *testing.B) {
 	runtime.GOMAXPROCS(1)
 	var timestamp = uint64(time.Now().Unix())
@@ -214,14 +224,14 @@ func Benchmark_Map_Uint64(b *testing.B) {
 }
 
 func BenchmarkList_GC(b *testing.B) {
-	runtime.GOMAXPROCS(1)
+	runtime.GOMAXPROCS(4)
 
-	var lists = []*List{}
+	var lists = []*expires.List{}
 
 	for m := 0; m < 1_000; m++ {
-		var list = NewList()
+		var list = expires.NewList()
 		for j := 0; j < 10_000; j++ {
-			list.Add(uint64(j), fasttime.Now().Unix()+100)
+			list.Add(uint64(j), fasttime.Now().Unix()+int64(rand.Int()%1_000_000))
 		}
 		lists = append(lists, list)
 	}
@@ -233,7 +243,7 @@ func BenchmarkList_GC(b *testing.B) {
 	b.RunParallel(func(pb *testing.PB) {
 		for pb.Next() {
 			for _, list := range lists {
-				list.GC(timestamp)
+				list.GC(timestamp + int64(rand.Int()%1_000_000))
 			}
 		}
 	})