!42 lru fix bug

Merge pull request !42 from K-on/feat-lru-driver

conf/public_conf_test.go

@@ -7,30 +7,18 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestLoadConf(t *testing.T) {
-	type args struct {
-		path string
-	}
-	tests := []struct {
-		name    string
-		args    args
-		wantErr bool
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if err := LoadConf(tt.args.path); (err != nil) != tt.wantErr {
-				t.Errorf("LoadConf() error = %v, wantErr %v", err, tt.wantErr)
-			}
-		})
-	}
-
-}
-
 func TestConf(t *testing.T) {
+	// 外部导入 conf.yaml 需要导入 conf 包
+
+	// get 使用, 读取 public_conf 配置文件
+	h := viper.Get("storage.host")
+	require.Equal(t, h, "127.0.0.1")
+
+	h = viper.Get("env")
+	require.Equal(t, h, "dev")
+
+	// set 使用
 	viper.Set("host", "1222")
 	host := viper.GetString("host")
-
 	require.Equal(t, host, "1222")
 }

conf/wheat-cache.yaml

@@ -6,3 +6,10 @@ storage:
   host: '127.0.0.1'
   port: 5890
   timeOut: 2  # second
+
+# clearSize and maxSize must be Int
+lruCache:
+  clearSize: "512MB"
+  maxSize: "1GB"
+  eventDriverSize: 2000
+  workTime: 1

doc/LRU设计.md

@@ -2,6 +2,6 @@
 
 EventP : 生产事件
 
-EventQ : 队列事件
+EventQ : 事件队列
 
 Event CP : 清理事件

makefile

@@ -31,3 +31,7 @@ gen-protobuf:
 .PHONY: gen-middleware
 gen-middleware:
 	@python3 ./shell/gen_middleware.py
+
+.PHONY: init-conf
+init-conf:
+	@python3 ./shell/init_conf.py

pkg/event/define.go

@@ -27,3 +27,4 @@ type ProduceInterface interface {
 type ConsumerInterface interface {
 	Receive(ctx context.Context) *Event
 }
+

pkg/lru/define.go

@@ -1 +1,22 @@
 package lru
+
+import "sync"
+
+type SingleWorkFunc func() interface{}
+
+const (
+	OptionEventName  = "operateEvent"
+	CleanEventName = "clearEvent"
+	WorkFuncEventKey = "workFunc"
+)
+
+var (
+	lruCacheOnce sync.Once
+	lruCache *singleCache
+)
+
+const (
+	lruMaxSize = 1*1024*1024*1024*8
+	lruClearSize = 0.5*1024*1024*1024*8
+	lruEventDriver = 2000
+)

pkg/lru/lru.go

@@ -0,0 +1,131 @@
+package lru
+
+import (
+	"container/list"
+	_ "gitee.com/timedb/wheatCache/conf"
+	"gitee.com/timedb/wheatCache/pkg/errorx"
+	"gitee.com/timedb/wheatCache/pkg/event"
+	"gitee.com/timedb/wheatCache/pkg/structure"
+	"gitee.com/timedb/wheatCache/pkg/util"
+	"github.com/spf13/viper"
+	"sync/atomic"
+)
+
+type keyBaseValue struct {
+	key string
+	val structure.KeyBaseInterface
+}
+
+type singleCache struct {
+	maxsize   int64 //最大的长度
+	clearSize int64 // 清理长度
+	nowSize   int64 // 现在的长度
+	li        *list.List
+	lruMap    map[string]*list.Element
+
+	lruDriver   event.DriverInterface
+	lruConsumer event.ConsumerInterface
+	lruCleanProduce  event.ProduceInterface // 发送清理事件
+}
+
+// UpdateLruSize 更新现在的长度
+func (lru *singleCache) UpdateLruSize(length int64) {
+	atomic.AddInt64(&lru.nowSize, length)
+}
+
+func cacheInit() (int64, int64, event.DriverInterface) {
+	maxSize := viper.GetString("lruCache.maxSize")
+	retMaxSize, maxErr:= util.ParseSizeToBit(maxSize)
+	if maxErr != nil{
+		return 0, 0, nil
+	}
+	if retMaxSize == 0{
+		retMaxSize = lruMaxSize
+	}
+
+	clearSize := viper.GetString("lruCache.clearSize")
+	retClearSize, clearErr := util.ParseSizeToBit(clearSize)
+	if clearErr != nil{
+		return 0, 0, nil
+	}
+	if retClearSize == 0{
+		retClearSize = lruClearSize
+	}
+
+	maxDriver := viper.GetInt("lruCache.eventDriverSize")
+	if maxDriver == 0{
+		maxDriver = lruEventDriver
+	}
+	lruDriver := event.NewDriver(maxDriver)
+	return retMaxSize, retClearSize, lruDriver
+}
+
+// NewLRUCache lru初始化
+func NewLRUCache() *singleCache {
+	maxSize, clearSize, lruDrivers := cacheInit()
+	lruCacheOnce.Do(func() {
+		_, _, lruDriver := cacheInit()
+		lru := &singleCache{
+			maxsize:     maxSize,
+			clearSize:   clearSize,
+			nowSize:     0,
+			li:          list.New(),
+			lruMap:      make(map[string]*list.Element),
+			lruDriver:   lruDriver,
+			lruConsumer: event.NewConsumer(lruDrivers),
+			lruCleanProduce:  event.NewProduce(lruDrivers),
+		}
+		lruCache = lru
+		go lru.lruSingleWork()
+	})
+	return lruCache
+}
+
+// GetDriver 获取驱动
+func (lru *singleCache) GetDriver() event.DriverInterface {
+	return lru.lruDriver
+}
+
+//Add 增加
+func (lru *singleCache) Add(key string, val structure.KeyBaseInterface) {
+
+	keyBaseVal := &keyBaseValue{
+		key: key,
+		val: val,
+	}
+	if elVal, ok := lru.lruMap[key]; ok {
+		lru.li.MoveToFront(elVal)
+		elVal.Value = keyBaseVal
+		return
+	}
+	valEl := lru.li.PushFront(keyBaseVal)
+	lru.lruMap[key] = valEl
+	//增加大小
+	lru.UpdateLruSize(valEl.Value.(*keyBaseValue).val.SizeByte())
+}
+
+// Get 查找key对应的value
+func (lru *singleCache) Get(key string) (structure.KeyBaseInterface, bool) {
+
+	if lru.lruMap == nil {
+		return nil, false
+	}
+	if elVal, ok := lru.lruMap[key]; ok {
+		lru.li.MoveToFront(elVal)
+		return elVal.Value.(*keyBaseValue).val, true
+	}
+	return nil, false
+}
+
+//Del 删除机制
+func (lru *singleCache) Del() error {
+	if lru.lruMap == nil {
+		return errorx.New("lru is nil")
+	}
+	data := lru.li.Back()
+	delete(lru.lruMap, data.Value.(*keyBaseValue).key)
+	//删除大小
+	lru.UpdateLruSize(-1 * data.Value.(*keyBaseValue).val.SizeByte())
+	lru.li.Remove(data)
+	return nil
+}

pkg/lru/lru_test.go

@@ -0,0 +1,24 @@
+package lru
+
+import (
+	"fmt"
+	"gitee.com/timedb/wheatCache/pkg/structure/stringx"
+	"github.com/stretchr/testify/require"
+	"testing"
+)
+
+func TestNewLRUCache(t *testing.T) {
+	cache := NewLRUCache()
+	v1 := stringx.NewStringSingle()
+	v2 := stringx.NewStringSingle()
+	v3 := stringx.NewStringSingle()
+	cache.Add("1", v1)
+	cache.Add("2", v2)
+	cache.Add("3", v3)
+	cache.Add("1", v1)
+	fmt.Println(cache.nowSize)
+	cache.Del()
+	fmt.Println(cache.nowSize)
+	_, isTrue := cache.Get("1")
+	require.Equal(t, isTrue, true)
+}

pkg/lru/woker_test.go

@@ -0,0 +1,85 @@
+package lru
+
+import (
+	"context"
+	"gitee.com/timedb/wheatCache/pkg/errorx"
+	"gitee.com/timedb/wheatCache/pkg/event"
+	"gitee.com/timedb/wheatCache/pkg/proto"
+	"gitee.com/timedb/wheatCache/pkg/structure"
+	"gitee.com/timedb/wheatCache/pkg/structure/stringx"
+	"github.com/stretchr/testify/require"
+	"testing"
+	"time"
+)
+
+func TestWorker(t *testing.T) {
+	ctx := context.Background()
+	lru := NewLRUCache()
+	produce := event.NewProduce(lru.GetDriver())
+	workEvent := event.NewEvent(OptionEventName)
+	workEvent.SetValue(WorkFuncEventKey, event.EventWorkFunc(func() (interface{}, error) {
+		v1 := stringx.NewStringSingle()
+		key := "v1"
+		res, err := v1.Set(&proto.SetRequest{
+			Val: "123",
+		})
+		if err != nil {
+			return nil, err
+		}
+		lru.Add(key, v1)
+		return res.Result, nil
+	}))
+	workEvent.InitWaitEvent()
+	produce.Call(ctx,workEvent)
+	res, err := workEvent.StartWaitEvent(2 * time.Second)
+	require.NoError(t, err)
+	require.Equal(t, res, "123")
+
+	workEvent.InitWaitEvent()
+	workEvent.SetValue(WorkFuncEventKey, event.EventWorkFunc(func() (interface{}, error) {
+		v2, ok := lru.Get("v1")
+		if !ok{
+			return nil, errorx.New("no this key")
+		}
+		switch v2.(type) {
+		case structure.StringXInterface:
+			res, err := v2.(structure.StringXInterface).Get(&proto.GetRequest{
+			})
+			if err != nil {
+				return nil, err
+			}
+			return res.Result, nil
+		default:
+			return nil, errorx.New("no this type")
+		}
+	}))
+	produce.Call(ctx, workEvent)
+	res, err = workEvent.StartWaitEvent(2 * time.Second)
+	require.NoError(t, err)
+	require.Equal(t, res, "123")
+
+	workEvent.InitWaitEvent()
+	workEvent.SetValue(WorkFuncEventKey, event.EventWorkFunc(func() (interface{}, error) {
+		lru.Del()
+		v2, ok := lru.Get("v1")
+		if !ok{
+			return nil, nil
+		}
+		switch v2.(type) {
+		case structure.StringXInterface:
+			res, err := v2.(structure.StringXInterface).Get(&proto.GetRequest{
+			})
+			if err != nil {
+				return nil, err
+			}
+			return res.Result, nil
+		default:
+			return nil, errorx.New("no this type")
+		}
+	}))
+	produce.Call(ctx, workEvent)
+	res, err = workEvent.StartWaitEvent(2 * time.Second)
+	require.Equal(t, err, nil)
+	require.Equal(t, res, nil)
+
+}

pkg/lru/worker.go

@@ -0,0 +1,30 @@
+package lru
+
+import (
+	"context"
+	"gitee.com/timedb/wheatCache/pkg/event"
+	"log"
+)
+
+func lruCleanWork() {
+
+}
+
+func (lru *singleCache) lruSingleWork() interface{} {
+	ctx := context.Background()
+	for {
+		workEvent := lru.lruConsumer.Receive(ctx)
+
+		workFunc, ok := workEvent.GetValue(WorkFuncEventKey)
+		if !ok {
+			continue
+		}
+
+		switch workFunc.(type) {
+		case event.EventWorkFunc:
+			workEvent.ExecWorkAndSendResult(workFunc.(event.EventWorkFunc))
+		default:
+			log.Print("this is debug ")
+		}
+	}
+}

pkg/util/memory.go

@@ -0,0 +1,32 @@
+package util
+
+import (
+	"gitee.com/timedb/wheatCache/pkg/errorx"
+	"regexp"
+	"strconv"
+	"strings"
+)
+
+// ParseSizeToBit
+// 支持MB, GB, KB, 格式 "5KB" 或者 "5kb"等等
+func ParseSizeToBit(size string) (int64, error) {
+	sizes := regexp.MustCompile("^\\d+")
+	sizeRes := sizes.FindAllString(size, 1)
+	unit := strings.Split(size, sizeRes[0])
+	Res, _ := strconv.ParseInt(sizeRes[0], 10, 64)
+	sizeType := strings.ToUpper(unit[1])
+	switch {
+	case sizeType == "BIT" || sizeType == "B":
+		return Res * 8, nil
+	case sizeType == "KB":
+		return Res * 1024 * 8, nil
+	case sizeType =="MB":
+		return Res * 1024 * 1024 * 8, nil
+	case sizeType == "GB":
+		return Res * 1024 *1024 * 1024 * 8, nil
+	default:
+		return 0, errorx.New("your size is wrong")
+	}
+}
+
+

pkg/util/memory_test.go

@@ -0,0 +1,18 @@
+package util
+
+import (
+	"github.com/stretchr/testify/require"
+	"testing"
+)
+
+func TestParseSizeToBit(t *testing.T) {
+	require.Equal(t, ParseSizeToBit("18Kb"), int64(18*1024*8))
+	require.Equal(t, ParseSizeToBit("18KB"), int64(18*1024*8))
+	require.Equal(t, ParseSizeToBit("18mB"), int64(18*1024*1024*8))
+	require.Equal(t, ParseSizeToBit("18gb"), int64(18*1024*1024*1024*8))
+
+	require.Equal(t, ParseSizeToBit("18b"), int64(18*8))
+	require.Equal(t, ParseSizeToBit("18B"), int64(18*8))
+	require.Equal(t, ParseSizeToBit("18bit"), int64(18*8))
+	require.Equal(t, ParseSizeToBit("18BIt"), int64(18*8))
+}

shell/init_conf.py

@@ -0,0 +1,22 @@
+import os
+from pathlib import Path
+import shutil
+
+sysPath = os.getcwd()
+confInPath = f"{sysPath}/conf/wheat-cache.yaml"
+confOutPath = "/etc/wheat-cache"
+
+
+def check_and_make_conf_dir():
+    conf_dir = Path(confOutPath)
+    if not conf_dir.is_dir():
+        os.makedirs(confOutPath)
+
+
+def copy_conf():
+    shutil.copy(confInPath, confOutPath)
+
+
+if __name__ == '__main__':
+    check_and_make_conf_dir()
+    copy_conf()