初步实验使用KV数据库（pebble）存储缓存索引

internal/caches/item.go

@@ -22,13 +22,14 @@ type Item struct {
 	Type       ItemType `json:"-"`
 	Key        string   `json:"1,omitempty"`
 	ExpiresAt  int64    `json:"2,omitempty"`
-	StaleAt    int64    `json:"-"`
+	StaleAt    int64    `json:"3,omitempty"`
 	HeaderSize int64    `json:"-"`
-	BodySize   int64    `json:"-"`
+	BodySize   int64    `json:"4,omitempty"`
 	MetaSize   int64    `json:"-"`
-	Host       string   `json:"-"`     // 主机名
-	ServerId   int64    `json:"3,omitempty"` // 服务ID
+	Host       string   `json:"-"`           // 主机名
+	ServerId   int64    `json:"5,omitempty"` // 服务ID
 	Week       int32    `json:"-"`
+	CreatedAt  int64    `json:"6,omitempty"`
 }
 
 func (this *Item) IsExpired() bool {

internal/caches/list_file_kv.go

@@ -0,0 +1,296 @@
+// Copyright 2024 GoEdge CDN goedge.cdn@gmail.com. All rights reserved. Official site: https://goedge.cn .
+
+package caches
+
+import (
+	"fmt"
+	"github.com/TeaOSLab/EdgeNode/internal/goman"
+	"github.com/TeaOSLab/EdgeNode/internal/utils/fnv"
+	"github.com/iwind/TeaGo/types"
+	"strings"
+	"testing"
+)
+
+const countKVStores = 10
+
+type KVFileList struct {
+	dir    string
+	stores [countKVStores]*KVListFileStore
+
+	onAdd    func(item *Item)
+	onRemove func(item *Item)
+}
+
+func NewKVFileList(dir string) *KVFileList {
+	dir = strings.TrimSuffix(dir, "/")
+
+	var stores = [countKVStores]*KVListFileStore{}
+	for i := 0; i < countKVStores; i++ {
+		stores[i] = NewKVListFileStore(dir + "/db-" + types.String(i) + ".store")
+	}
+
+	return &KVFileList{
+		dir:    dir,
+		stores: stores,
+	}
+}
+
+// Init 初始化
+func (this *KVFileList) Init() error {
+	for _, store := range this.stores {
+		err := store.Open()
+		if err != nil {
+			return fmt.Errorf("open store '"+store.Path()+"' failed: %w", err)
+		}
+	}
+
+	return nil
+}
+
+// Reset 重置数据
+func (this *KVFileList) Reset() error {
+	// do nothing
+	return nil
+}
+
+// Add 添加内容
+func (this *KVFileList) Add(hash string, item *Item) error {
+	err := this.getStore(hash).AddItem(hash, item)
+	if err != nil {
+		return err
+	}
+
+	if this.onAdd != nil {
+		this.onAdd(item)
+	}
+
+	return nil
+}
+
+// Exist 检查内容是否存在
+func (this *KVFileList) Exist(hash string) (bool, error) {
+	return this.getStore(hash).ExistItem(hash)
+}
+
+// ExistQuick 快速检查内容是否存在
+func (this *KVFileList) ExistQuick(hash string) (bool, error) {
+	return this.getStore(hash).ExistQuickItem(hash)
+}
+
+// CleanPrefix 清除某个前缀的缓存
+func (this *KVFileList) CleanPrefix(prefix string) error {
+	var group = goman.NewTaskGroup()
+	var lastErr error
+	for _, store := range this.stores {
+		var storeCopy = store
+		group.Run(func() {
+			err := storeCopy.CleanItemsWithPrefix(prefix)
+			if err != nil {
+				lastErr = err
+			}
+		})
+	}
+	group.Wait()
+	return lastErr
+}
+
+// CleanMatchKey 清除通配符匹配的Key
+func (this *KVFileList) CleanMatchKey(key string) error {
+	var group = goman.NewTaskGroup()
+	var lastErr error
+	for _, store := range this.stores {
+		var storeCopy = store
+		group.Run(func() {
+			err := storeCopy.CleanItemsWithWildcardKey(key)
+			if err != nil {
+				lastErr = err
+			}
+		})
+	}
+	group.Wait()
+	return lastErr
+}
+
+// CleanMatchPrefix 清除通配符匹配的前缀
+func (this *KVFileList) CleanMatchPrefix(prefix string) error {
+	var group = goman.NewTaskGroup()
+	var lastErr error
+	for _, store := range this.stores {
+		var storeCopy = store
+		group.Run(func() {
+			err := storeCopy.CleanItemsWithWildcardPrefix(prefix)
+			if err != nil {
+				lastErr = err
+			}
+		})
+	}
+	group.Wait()
+	return lastErr
+}
+
+// Remove 删除内容
+func (this *KVFileList) Remove(hash string) error {
+	err := this.getStore(hash).RemoveItem(hash)
+	if err != nil {
+		return err
+	}
+
+	if this.onRemove != nil {
+		// when remove file item, no any extra information needed
+		this.onRemove(nil)
+	}
+
+	return nil
+}
+
+// Purge 清理过期数据
+func (this *KVFileList) Purge(count int, callback func(hash string) error) (int, error) {
+	count /= countKVStores
+	if count <= 0 {
+		count = 100
+	}
+
+	var countFound = 0
+	var lastErr error
+	for _, store := range this.stores {
+		purgeCount, err := store.PurgeItems(count, callback)
+		countFound += purgeCount
+		if err != nil {
+			lastErr = err
+		}
+	}
+
+	return countFound, lastErr
+}
+
+// PurgeLFU 清理LFU数据
+func (this *KVFileList) PurgeLFU(count int, callback func(hash string) error) error {
+	count /= countKVStores
+	if count <= 0 {
+		count = 100
+	}
+
+	var lastErr error
+	for _, store := range this.stores {
+		err := store.PurgeLFUItems(count, callback)
+		if err != nil {
+			lastErr = err
+		}
+	}
+	return lastErr
+}
+
+// CleanAll 清除所有缓存
+func (this *KVFileList) CleanAll() error {
+	var group = goman.NewTaskGroup()
+	var lastErr error
+	for _, store := range this.stores {
+		var storeCopy = store
+		group.Run(func() {
+			err := storeCopy.RemoveAllItems()
+			if err != nil {
+				lastErr = err
+			}
+		})
+	}
+	group.Wait()
+	return lastErr
+}
+
+// Stat 统计
+func (this *KVFileList) Stat(check func(hash string) bool) (*Stat, error) {
+	var stat = &Stat{}
+
+	var group = goman.NewTaskGroup()
+
+	var lastErr error
+	for _, store := range this.stores {
+		var storeCopy = store
+		group.Run(func() {
+			storeStat, err := storeCopy.StatItems()
+			if err != nil {
+				lastErr = err
+				return
+			}
+
+			group.Lock()
+			stat.Size += storeStat.Size
+			stat.ValueSize += storeStat.ValueSize
+			stat.Count += storeStat.Count
+			group.Unlock()
+		})
+	}
+
+	group.Wait()
+
+	return stat, lastErr
+}
+
+// Count 总数量
+func (this *KVFileList) Count() (int64, error) {
+	var count int64
+
+	var group = goman.NewTaskGroup()
+
+	var lastErr error
+	for _, store := range this.stores {
+		var storeCopy = store
+		group.Run(func() {
+			countStoreItems, err := storeCopy.CountItems()
+			if err != nil {
+				lastErr = err
+				return
+			}
+
+			group.Lock()
+			count += countStoreItems
+			group.Unlock()
+		})
+	}
+
+	group.Wait()
+
+	return count, lastErr
+}
+
+// OnAdd 添加事件
+func (this *KVFileList) OnAdd(fn func(item *Item)) {
+	this.onAdd = fn
+}
+
+// OnRemove 删除事件
+func (this *KVFileList) OnRemove(fn func(item *Item)) {
+	this.onRemove = fn
+}
+
+// Close 关闭
+func (this *KVFileList) Close() error {
+	var lastErr error
+	for _, store := range this.stores {
+		err := store.Close()
+		if err != nil {
+			lastErr = err
+		}
+	}
+	return lastErr
+}
+
+// IncreaseHit 增加点击量
+func (this *KVFileList) IncreaseHit(hash string) error {
+	// do nothing
+	return nil
+}
+
+func (this *KVFileList) TestInspect(t *testing.T) error {
+	for _, store := range this.stores {
+		err := store.TestInspect(t)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (this *KVFileList) getStore(hash string) *KVListFileStore {
+	return this.stores[fnv.HashString(hash)%countKVStores]
+}

internal/caches/list_file_kv_objects.go

@@ -0,0 +1,66 @@
+// Copyright 2024 GoEdge CDN goedge.cdn@gmail.com. All rights reserved. Official site: https://goedge.cn .
+
+package caches
+
+import (
+	"encoding/binary"
+	"encoding/json"
+	"strings"
+)
+
+// ItemKVEncoder item encoder
+type ItemKVEncoder[T interface{ *Item }] struct {
+}
+
+func NewItemKVEncoder[T interface{ *Item }]() *ItemKVEncoder[T] {
+	return &ItemKVEncoder[T]{}
+}
+
+func (this *ItemKVEncoder[T]) Encode(value T) ([]byte, error) {
+	return json.Marshal(value)
+}
+
+func (this *ItemKVEncoder[T]) EncodeField(value T, fieldName string) ([]byte, error) {
+	switch fieldName {
+	case "createdAt":
+		var b = make([]byte, 4)
+		var createdAt = any(value).(*Item).CreatedAt
+		binary.BigEndian.PutUint32(b, uint32(createdAt))
+		return b, nil
+	case "staleAt":
+		var b = make([]byte, 4)
+		var staleAt = any(value).(*Item).StaleAt
+		if staleAt < 0 {
+			staleAt = 0
+		}
+		binary.BigEndian.PutUint32(b, uint32(staleAt))
+		return b, nil
+	case "serverId":
+		var b = make([]byte, 4)
+		var serverId = any(value).(*Item).ServerId
+		if serverId < 0 {
+			serverId = 0
+		}
+		binary.BigEndian.PutUint32(b, uint32(serverId))
+		return b, nil
+	case "key":
+		return []byte(any(value).(*Item).Key), nil
+	case "wildKey":
+		var key = any(value).(*Item).Key
+		var dotIndex = strings.Index(key, ".")
+		if dotIndex > 0 {
+			var slashIndex = strings.LastIndex(key[:dotIndex], "/")
+			if slashIndex > 0 {
+				key = key[:dotIndex][:slashIndex+1] + "*" + key[dotIndex:]
+			}
+		}
+
+		return []byte(key), nil
+	}
+	return nil, nil
+}
+
+func (this *ItemKVEncoder[T]) Decode(valueBytes []byte) (value T, err error) {
+	err = json.Unmarshal(valueBytes, &value)
+	return
+}

internal/caches/list_file_kv_objects_test.go

@@ -0,0 +1,69 @@
+// Copyright 2024 GoEdge CDN goedge.cdn@gmail.com. All rights reserved. Official site: https://goedge.cn .
+
+package caches_test
+
+import (
+	"github.com/TeaOSLab/EdgeNode/internal/caches"
+	"github.com/iwind/TeaGo/assert"
+	"testing"
+)
+
+func TestItemKVEncoder_EncodeField(t *testing.T) {
+	var a = assert.NewAssertion(t)
+
+	var encoder = caches.NewItemKVEncoder[*caches.Item]()
+	{
+		key, err := encoder.EncodeField(&caches.Item{
+			Key: "https://example.com/index.html",
+		}, "key")
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Log("key:", string(key))
+		a.IsTrue(string(key) == "https://example.com/index.html")
+	}
+
+	{
+		key, err := encoder.EncodeField(&caches.Item{
+			Key: "",
+		}, "wildKey")
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Log("key:", string(key))
+		a.IsTrue(string(key) == "")
+	}
+
+	{
+		key, err := encoder.EncodeField(&caches.Item{
+			Key: "example.com/index.html",
+		}, "wildKey")
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Log("key:", string(key))
+		a.IsTrue(string(key) == "example.com/index.html")
+	}
+
+	{
+		key, err := encoder.EncodeField(&caches.Item{
+			Key: "https://example.com/index.html",
+		}, "wildKey")
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Log("key:", string(key))
+		a.IsTrue(string(key) == "https://*.com/index.html")
+	}
+
+	{
+		key, err := encoder.EncodeField(&caches.Item{
+			Key: "https://www.example.com/index.html",
+		}, "wildKey")
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Log("key:", string(key))
+		a.IsTrue(string(key) == "https://*.example.com/index.html")
+	}
+}

internal/caches/list_file_kv_store.go

@@ -0,0 +1,480 @@
+// Copyright 2024 GoEdge CDN goedge.cdn@gmail.com. All rights reserved. Official site: https://goedge.cn .
+
+package caches
+
+import (
+	"errors"
+	"github.com/TeaOSLab/EdgeNode/internal/utils/fasttime"
+	"github.com/TeaOSLab/EdgeNode/internal/utils/kvstore"
+	"github.com/cockroachdb/pebble"
+	"github.com/iwind/TeaGo/types"
+	"regexp"
+	"strings"
+	"testing"
+)
+
+type KVListFileStore struct {
+	path     string
+	rawStore *kvstore.Store
+
+	// tables
+	itemsTable *kvstore.Table[*Item]
+
+	isReady bool
+}
+
+func NewKVListFileStore(path string) *KVListFileStore {
+	return &KVListFileStore{
+		path: path,
+	}
+}
+
+func (this *KVListFileStore) Open() error {
+	var reg = regexp.MustCompile(`^(.+)/([\w-]+)(\.store)$`)
+	var matches = reg.FindStringSubmatch(this.path)
+	if len(matches) != 4 {
+		return errors.New("invalid path '" + this.path + "'")
+	}
+	var dir = matches[1]
+	var name = matches[2]
+
+	rawStore, err := kvstore.OpenStoreDir(dir, name)
+	if err != nil {
+		return err
+	}
+	this.rawStore = rawStore
+
+	db, err := rawStore.NewDB("cache")
+	if err != nil {
+		return err
+	}
+
+	{
+		table, tableErr := kvstore.NewTable[*Item]("items", NewItemKVEncoder[*Item]())
+		if tableErr != nil {
+			return tableErr
+		}
+
+		err = table.AddFields("staleAt", "key", "wildKey", "createdAt")
+		if err != nil {
+			return err
+		}
+
+		db.AddTable(table)
+		this.itemsTable = table
+	}
+
+	this.isReady = true
+
+	return nil
+}
+
+func (this *KVListFileStore) Path() string {
+	return this.path
+}
+
+func (this *KVListFileStore) AddItem(hash string, item *Item) error {
+	if !this.isReady {
+		return nil
+	}
+
+	var currentTime = fasttime.Now().Unix()
+	if item.ExpiresAt <= currentTime {
+		return errors.New("invalid expires time '" + types.String(item.ExpiresAt) + "'")
+	}
+	if item.CreatedAt <= 0 {
+		item.CreatedAt = currentTime
+	}
+	if item.StaleAt <= 0 {
+		item.StaleAt = item.ExpiresAt + DefaultStaleCacheSeconds
+	}
+	return this.itemsTable.Set(hash, item)
+}
+
+func (this *KVListFileStore) ExistItem(hash string) (bool, error) {
+	if !this.isReady {
+		return false, nil
+	}
+
+	item, err := this.itemsTable.Get(hash)
+	if err != nil {
+		if kvstore.IsKeyNotFound(err) {
+			return false, nil
+		}
+		return false, err
+	}
+	if item == nil {
+		return false, nil
+	}
+
+	return item.ExpiresAt >= fasttime.NewFastTime().Unix(), nil
+}
+
+func (this *KVListFileStore) ExistQuickItem(hash string) (bool, error) {
+	if !this.isReady {
+		return false, nil
+	}
+
+	return this.itemsTable.Exist(hash)
+}
+
+func (this *KVListFileStore) RemoveItem(hash string) error {
+	if !this.isReady {
+		return nil
+	}
+
+	return this.itemsTable.Delete(hash)
+}
+
+func (this *KVListFileStore) RemoveAllItems() error {
+	if !this.isReady {
+		return nil
+	}
+
+	return this.itemsTable.Truncate()
+}
+
+func (this *KVListFileStore) PurgeItems(count int, callback func(hash string) error) (int, error) {
+	if !this.isReady {
+		return 0, nil
+	}
+
+	var countFound int
+	var currentTime = fasttime.Now().Unix()
+	var hashList []string
+	err := this.itemsTable.
+		Query().
+		FieldAsc("staleAt").
+		Limit(count).
+		FindAll(func(tx *kvstore.Tx[*Item], item kvstore.Item[*Item]) (goNext bool, err error) {
+			if item.Value == nil {
+				return true, nil
+			}
+			if item.Value.StaleAt < currentTime {
+				countFound++
+				hashList = append(hashList, item.Key)
+				return true, nil
+			}
+			return false, nil
+		})
+	if err != nil {
+		return 0, err
+	}
+
+	// delete items
+	if len(hashList) > 0 {
+		txErr := this.itemsTable.WriteTx(func(tx *kvstore.Tx[*Item]) error {
+			for _, hash := range hashList {
+				deleteErr := tx.Delete(hash)
+				if deleteErr != nil {
+					return deleteErr
+				}
+			}
+			return nil
+		})
+		if txErr != nil {
+			return 0, txErr
+		}
+
+		for _, hash := range hashList {
+			callbackErr := callback(hash)
+			if callbackErr != nil {
+				return 0, callbackErr
+			}
+		}
+	}
+
+	return countFound, nil
+}
+
+func (this *KVListFileStore) PurgeLFUItems(count int, callback func(hash string) error) error {
+	if !this.isReady {
+		return nil
+	}
+
+	var hashList []string
+	err := this.itemsTable.
+		Query().
+		FieldAsc("createdAt").
+		Limit(count).
+		FindAll(func(tx *kvstore.Tx[*Item], item kvstore.Item[*Item]) (goNext bool, err error) {
+			if item.Value != nil {
+				hashList = append(hashList, item.Key)
+			}
+			return true, nil
+		})
+	if err != nil {
+		return err
+	}
+
+	// delete items
+	if len(hashList) > 0 {
+		txErr := this.itemsTable.WriteTx(func(tx *kvstore.Tx[*Item]) error {
+			for _, hash := range hashList {
+				deleteErr := tx.Delete(hash)
+				if deleteErr != nil {
+					return deleteErr
+				}
+			}
+			return nil
+		})
+		if txErr != nil {
+			return txErr
+		}
+
+		for _, hash := range hashList {
+			callbackErr := callback(hash)
+			if callbackErr != nil {
+				return callbackErr
+			}
+		}
+	}
+
+	return nil
+}
+
+func (this *KVListFileStore) CleanItemsWithPrefix(prefix string) error {
+	if !this.isReady {
+		return nil
+	}
+
+	if len(prefix) == 0 {
+		return nil
+	}
+
+	var currentTime = fasttime.Now().Unix()
+
+	var fieldOffset []byte
+	const size = 1000
+	for {
+		var count int
+		err := this.itemsTable.
+			Query().
+			FieldPrefix("key", prefix).
+			FieldOffset(fieldOffset).
+			Limit(size).
+			ForUpdate().
+			FindAll(func(tx *kvstore.Tx[*Item], item kvstore.Item[*Item]) (goNext bool, err error) {
+				if item.Value == nil {
+					return true, nil
+				}
+
+				count++
+				fieldOffset = item.FieldKey
+
+				if item.Value.CreatedAt >= currentTime {
+					return true, nil
+				}
+				if item.Value.ExpiresAt == 0 {
+					return true, nil
+				}
+
+				item.Value.ExpiresAt = 0
+				item.Value.StaleAt = 0
+
+				setErr := tx.Set(item.Key, item.Value) // TODO improve performance
+				if setErr != nil {
+					return false, setErr
+				}
+
+				return true, nil
+			})
+		if err != nil {
+			return err
+		}
+
+		if count < size {
+			break
+		}
+	}
+
+	return nil
+}
+
+func (this *KVListFileStore) CleanItemsWithWildcardPrefix(prefix string) error {
+	if !this.isReady {
+		return nil
+	}
+
+	if len(prefix) == 0 {
+		return nil
+	}
+
+	var currentTime = fasttime.Now().Unix()
+
+	var fieldOffset []byte
+	const size = 1000
+	for {
+		var count int
+		err := this.itemsTable.
+			Query().
+			FieldPrefix("wildKey", prefix).
+			FieldOffset(fieldOffset).
+			Limit(size).
+			ForUpdate().
+			FindAll(func(tx *kvstore.Tx[*Item], item kvstore.Item[*Item]) (goNext bool, err error) {
+				if item.Value == nil {
+					return true, nil
+				}
+
+				count++
+				fieldOffset = item.FieldKey
+
+				if item.Value.CreatedAt >= currentTime {
+					return true, nil
+				}
+				if item.Value.ExpiresAt == 0 {
+					return true, nil
+				}
+
+				item.Value.ExpiresAt = 0
+				item.Value.StaleAt = 0
+
+				setErr := tx.Set(item.Key, item.Value) // TODO improve performance
+				if setErr != nil {
+					return false, setErr
+				}
+
+				return true, nil
+			})
+		if err != nil {
+			return err
+		}
+
+		if count < size {
+			break
+		}
+	}
+
+	return nil
+}
+
+func (this *KVListFileStore) CleanItemsWithWildcardKey(key string) error {
+	if !this.isReady {
+		return nil
+	}
+
+	if len(key) == 0 {
+		return nil
+	}
+
+	var currentTime = fasttime.Now().Unix()
+
+	for _, realKey := range []string{key, key + SuffixAll} {
+		var fieldOffset = append(this.itemsTable.FieldKey("wildKey"), '$')
+		fieldOffset = append(fieldOffset, realKey...)
+		const size = 1000
+
+		var wildKey string
+		if !strings.HasSuffix(realKey, SuffixAll) {
+			wildKey = string(append([]byte(realKey), 0, 0))
+		} else {
+			wildKey = realKey
+		}
+
+		for {
+			var count int
+			err := this.itemsTable.
+				Query().
+				FieldPrefix("wildKey", wildKey).
+				FieldOffset(fieldOffset).
+				Limit(size).
+				ForUpdate().
+				FindAll(func(tx *kvstore.Tx[*Item], item kvstore.Item[*Item]) (goNext bool, err error) {
+					if item.Value == nil {
+						return true, nil
+					}
+
+					count++
+					fieldOffset = item.FieldKey
+
+					if item.Value.CreatedAt >= currentTime {
+						return true, nil
+					}
+					if item.Value.ExpiresAt == 0 {
+						return true, nil
+					}
+
+					item.Value.ExpiresAt = 0
+					item.Value.StaleAt = 0
+
+					setErr := tx.Set(item.Key, item.Value) // TODO improve performance
+					if setErr != nil {
+						return false, setErr
+					}
+
+					return true, nil
+				})
+			if err != nil {
+				return err
+			}
+
+			if count < size {
+				break
+			}
+		}
+	}
+
+	return nil
+}
+
+func (this *KVListFileStore) CountItems() (int64, error) {
+	if !this.isReady {
+		return 0, nil
+	}
+
+	return this.itemsTable.Count()
+}
+
+func (this *KVListFileStore) StatItems() (*Stat, error) {
+	if !this.isReady {
+		return &Stat{}, nil
+	}
+
+	var stat = &Stat{}
+
+	err := this.itemsTable.
+		Query().
+		FindAll(func(tx *kvstore.Tx[*Item], item kvstore.Item[*Item]) (goNext bool, err error) {
+			if item.Value != nil {
+				stat.Size += item.Value.Size()
+				stat.ValueSize += item.Value.BodySize
+				stat.Count++
+			}
+			return true, nil
+		})
+	return stat, err
+}
+
+func (this *KVListFileStore) TestInspect(t *testing.T) error {
+	if !this.isReady {
+		return nil
+	}
+
+	it, err := this.rawStore.RawDB().NewIter(&pebble.IterOptions{})
+	if err != nil {
+		return err
+	}
+	defer func() {
+		_ = it.Close()
+	}()
+
+	for it.First(); it.Valid(); it.Next() {
+		valueBytes, valueErr := it.ValueAndErr()
+		if valueErr != nil {
+			return valueErr
+		}
+		t.Log(string(it.Key()), "=>", string(valueBytes))
+	}
+	return nil
+}
+
+func (this *KVListFileStore) Close() error {
+	this.isReady = false
+
+	if this.rawStore != nil {
+		return this.rawStore.Close()
+	}
+
+	return nil
+}

internal/caches/list_file_kv_test.go

@@ -0,0 +1,317 @@
+// Copyright 2024 GoEdge CDN goedge.cdn@gmail.com. All rights reserved. Official site: https://goedge.cn .
+
+package caches_test
+
+import (
+	"fmt"
+	"github.com/TeaOSLab/EdgeNode/internal/caches"
+	"github.com/TeaOSLab/EdgeNode/internal/utils/testutils"
+	"github.com/iwind/TeaGo/Tea"
+	_ "github.com/iwind/TeaGo/bootstrap"
+	stringutil "github.com/iwind/TeaGo/utils/string"
+	"math/rand"
+	"strconv"
+	"sync"
+	"testing"
+	"time"
+)
+
+var testingKVList *caches.KVFileList
+
+func TestMain(m *testing.M) {
+	m.Run()
+
+	if testingKVList != nil {
+		_ = testingKVList.Close()
+	}
+}
+
+func testOpenKVFileList(t *testing.T) *caches.KVFileList {
+	var list = caches.NewKVFileList(Tea.Root + "/data/stores/cache-stores")
+	err := list.Init()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	testingKVList = list
+	return list
+}
+
+func TestNewKVFileList(t *testing.T) {
+	_ = testOpenKVFileList(t)
+}
+
+func TestKVFileList_Add(t *testing.T) {
+	var list = testOpenKVFileList(t)
+
+	err := list.Add(stringutil.Md5("123456"), &caches.Item{
+		Type:       caches.ItemTypeFile,
+		Key:        "https://example.com/index.html",
+		ExpiresAt:  time.Now().Unix() + 60,
+		StaleAt:    0,
+		HeaderSize: 0,
+		BodySize:   4096,
+		MetaSize:   0,
+		Host:       "",
+		ServerId:   1,
+		Week:       0,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestKVFileList_Add_Many(t *testing.T) {
+	if !testutils.IsSingleTesting() {
+		return
+	}
+
+	var list = testOpenKVFileList(t)
+
+	const start = 0
+	const count = 1_000_000
+	const concurrent = 100
+
+	var before = time.Now()
+	defer func() {
+		var costSeconds = time.Since(before).Seconds()
+		t.Log("cost:", fmt.Sprintf("%.2fs", costSeconds), "qps:", fmt.Sprintf("%.2fK/s", float64(count)/1000/costSeconds))
+	}()
+
+	var wg = &sync.WaitGroup{}
+	wg.Add(concurrent)
+	for c := 0; c < concurrent; c++ {
+		go func(c int) {
+			defer wg.Done()
+
+			var segmentStart = start + count/concurrent*c
+			for i := segmentStart; i < segmentStart+count/concurrent; i++ {
+				err := list.Add(stringutil.Md5(strconv.Itoa(i)), &caches.Item{
+					Type:       caches.ItemTypeFile,
+					Key:        "https://www.example.com/index.html" + strconv.Itoa(i),
+					ExpiresAt:  time.Now().Unix() + 60,
+					StaleAt:    0,
+					HeaderSize: 0,
+					BodySize:   int64(rand.Int() % 1_000_000),
+					MetaSize:   0,
+					Host:       "",
+					ServerId:   1,
+					Week:       0,
+				})
+				if err != nil {
+					t.Log(err)
+				}
+			}
+		}(c)
+	}
+	wg.Wait()
+}
+
+func TestKVFileList_Add_Many_Suffix(t *testing.T) {
+	if !testutils.IsSingleTesting() {
+		return
+	}
+
+	var list = testOpenKVFileList(t)
+
+	const start = 0
+	const count = 1000
+	const concurrent = 100
+
+	var before = time.Now()
+	defer func() {
+		var costSeconds = time.Since(before).Seconds()
+		t.Log("cost:", fmt.Sprintf("%.2fs", costSeconds), "qps:", fmt.Sprintf("%.2fK/s", float64(count)/1000/costSeconds))
+	}()
+
+	var wg = &sync.WaitGroup{}
+	wg.Add(concurrent)
+	for c := 0; c < concurrent; c++ {
+		go func(c int) {
+			defer wg.Done()
+
+			var segmentStart = start + count/concurrent*c
+			for i := segmentStart; i < segmentStart+count/concurrent; i++ {
+				err := list.Add(stringutil.Md5(strconv.Itoa(i)+caches.SuffixAll), &caches.Item{
+					Type:       caches.ItemTypeFile,
+					Key:        "https://www.example.com/index.html" + strconv.Itoa(i) + caches.SuffixAll + "zip",
+					ExpiresAt:  time.Now().Unix() + 60,
+					StaleAt:    0,
+					HeaderSize: 0,
+					BodySize:   int64(rand.Int() % 1_000_000),
+					MetaSize:   0,
+					Host:       "",
+					ServerId:   1,
+					Week:       0,
+				})
+				if err != nil {
+					t.Log(err)
+				}
+			}
+		}(c)
+	}
+	wg.Wait()
+}
+
+func TestKVFileList_Exist(t *testing.T) {
+	var list = testOpenKVFileList(t)
+	for _, hash := range []string{
+		stringutil.Md5("123456"),
+		stringutil.Md5("654321"),
+	} {
+		b, err := list.Exist(hash)
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Log(hash, "=>", b)
+	}
+}
+
+func TestKVFileList_Remove(t *testing.T) {
+	var list = testOpenKVFileList(t)
+	for _, hash := range []string{
+		stringutil.Md5("123456"),
+		stringutil.Md5("654321"),
+	} {
+		err := list.Remove(hash)
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+func TestKVFileList_CleanAll(t *testing.T) {
+	var list = testOpenKVFileList(t)
+	err := list.CleanAll()
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestKVFileList_Inspect(t *testing.T) {
+	if !testutils.IsSingleTesting() {
+		return
+	}
+
+	var list = testOpenKVFileList(t)
+	err := list.TestInspect(t)
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestKVFileList_Purge(t *testing.T) {
+	var list = testOpenKVFileList(t)
+	var before = time.Now()
+	count, err := list.Purge(4_000, func(hash string) error {
+		//t.Log("hash:", hash)
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Log("cost:", fmt.Sprintf("%.2fms", time.Since(before).Seconds()*1000), "count:", count)
+}
+
+func TestKVFileList_PurgeLFU(t *testing.T) {
+	var list = testOpenKVFileList(t)
+	var before = time.Now()
+	err := list.PurgeLFU(20000, func(hash string) error {
+		t.Log("hash:", hash)
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Log("cost:", fmt.Sprintf("%.2fms", time.Since(before).Seconds()*1000))
+}
+
+func TestKVFileList_Count(t *testing.T) {
+	var list = testOpenKVFileList(t)
+	var before = time.Now()
+	count, err := list.Count()
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Log("cost:", fmt.Sprintf("%.2fms", time.Since(before).Seconds()*1000), "count:", count)
+}
+
+func TestKVFileList_Stat(t *testing.T) {
+	var list = testOpenKVFileList(t)
+	var before = time.Now()
+	stat, err := list.Stat(func(hash string) bool {
+		return true
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Log("cost:", fmt.Sprintf("%.2fms", time.Since(before).Seconds()*1000), "stat:", fmt.Sprintf("%+v", stat))
+}
+
+func TestKVFileList_CleanPrefix(t *testing.T) {
+	var list = testOpenKVFileList(t)
+	var before = time.Now()
+
+	defer func() {
+		var costSeconds = time.Since(before).Seconds()
+		t.Log("cost:", fmt.Sprintf("%.2fms", costSeconds*1000))
+	}()
+
+	err := list.CleanPrefix("https://www.example.com/index.html")
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestKVFileList_CleanMatchPrefix(t *testing.T) {
+	var list = testOpenKVFileList(t)
+	var before = time.Now()
+
+	defer func() {
+		var costSeconds = time.Since(before).Seconds()
+		t.Log("cost:", fmt.Sprintf("%.2fms", costSeconds*1000))
+	}()
+
+	err := list.CleanMatchPrefix("https://*.example.com/index.html")
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestKVFileList_CleanMatchKey(t *testing.T) {
+	var list = testOpenKVFileList(t)
+	var before = time.Now()
+
+	defer func() {
+		var costSeconds = time.Since(before).Seconds()
+		t.Log("cost:", fmt.Sprintf("%.2fms", costSeconds*1000))
+	}()
+
+	err := list.CleanMatchKey("https://*.example.com/index.html123")
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func BenchmarkKVFileList_Exist(b *testing.B) {
+	var list = caches.NewKVFileList(Tea.Root + "/data/stores/cache-stores")
+	err := list.Init()
+	if err != nil {
+		b.Fatal(err)
+	}
+
+	defer func() {
+		_ = list.Close()
+	}()
+
+	b.ResetTimer()
+
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			_, existErr := list.Exist(stringutil.Md5(strconv.Itoa(rand.Int() % 2_000_000)))
+			if existErr != nil {
+				b.Fatal(existErr)
+			}
+		}
+	})
+}

internal/caches/storage_file.go

@@ -258,12 +258,25 @@ func (this *FileStorage) Init() error {
 		return errors.New("[CACHE]cache storage dir can not be empty")
 	}
 
-	var list = NewSQLiteFileList(dir + "/p" + types.String(this.policy.Id) + "/.indexes")
-	err = list.Init()
-	if err != nil {
-		return err
+	// read list
+	var list ListInterface
+	var sqliteIndexesDir = dir + "/p" + types.String(this.policy.Id) + "/.indexes"
+	_, sqliteIndexesDirErr := os.Stat(sqliteIndexesDir)
+	if sqliteIndexesDirErr == nil || !teaconst.EnableKVCacheStore {
+		list = NewSQLiteFileList(sqliteIndexesDir)
+		err = list.Init()
+		if err != nil {
+			return err
+		}
+		list.(*SQLiteFileList).SetOldDir(dir + "/p" + types.String(this.policy.Id))
+	} else {
+		list = NewKVFileList(dir + "/p" + types.String(this.policy.Id) + "/.stores")
+		err = list.Init()
+		if err != nil {
+			return err
+		}
 	}
-	list.(*SQLiteFileList).SetOldDir(dir + "/p" + types.String(this.policy.Id))
+
 	this.list = list
 
 	// 检查目录是否存在
@@ -1608,7 +1621,8 @@ func (this *FileStorage) subDir(hash string) (dirPath string, dirIsFull bool) {
 // ScanGarbageCaches 清理目录中“失联”的缓存文件
 // “失联”为不在HashMap中的文件
 func (this *FileStorage) ScanGarbageCaches(fileCallback func(path string) error) error {
-	if !this.list.(*SQLiteFileList).HashMapIsLoaded() {
+	_, isSQLite := this.list.(*SQLiteFileList)
+	if isSQLite && !this.list.(*SQLiteFileList).HashMapIsLoaded() {
 		return errors.New("cache list is loading")
 	}
 
@@ -1641,8 +1655,8 @@ func (this *FileStorage) ScanGarbageCaches(fileCallback func(path string) error)
 				continue
 			}
 
-			dir2Matches, err := filepath.Glob(dir1 + "/*")
-			if err != nil {
+			dir2Matches, globErr := filepath.Glob(dir1 + "/*")
+			if globErr != nil {
 				// ignore error
 				continue
 			}
@@ -1665,8 +1679,8 @@ func (this *FileStorage) ScanGarbageCaches(fileCallback func(path string) error)
 					}
 				}
 
-				fileMatches, err := filepath.Glob(dir2 + "/*.cache")
-				if err != nil {
+				fileMatches, globDir2Err := filepath.Glob(dir2 + "/*.cache")
+				if globDir2Err != nil {
 					// ignore error
 					continue
 				}
@@ -1678,7 +1692,19 @@ func (this *FileStorage) ScanGarbageCaches(fileCallback func(path string) error)
 						continue
 					}
 
-					isReady, found := this.list.(*SQLiteFileList).ExistQuick(hash)
+					var isReady, found bool
+					switch rawList := this.list.(type) {
+					case *SQLiteFileList:
+						isReady, found = rawList.ExistQuick(hash)
+					case *KVFileList:
+						isReady = true
+						var checkErr error
+						found, checkErr = rawList.ExistQuick(hash)
+						if checkErr != nil {
+							return checkErr
+						}
+					}
+
 					if !isReady {
 						continue
 					}
@@ -1688,8 +1714,8 @@ func (this *FileStorage) ScanGarbageCaches(fileCallback func(path string) error)
 					}
 
 					// 检查文件正在被写入
-					stat, err := os.Stat(file)
-					if err != nil {
+					stat, statErr := os.Stat(file)
+					if statErr != nil {
 						continue
 					}
 					if fasttime.Now().Unix()-stat.ModTime().Unix() < 300 /** 5 minutes **/ {
@@ -1698,9 +1724,9 @@ func (this *FileStorage) ScanGarbageCaches(fileCallback func(path string) error)
 
 					if fileCallback != nil {
 						countFound++
-						err = fileCallback(file)
-						if err != nil {
-							return err
+						callbackErr := fileCallback(file)
+						if callbackErr != nil {
+							return callbackErr
 						}
 					}
 				}