文件缓存策略支持二级缓存（内存 | 文件）

internal/caches/item.go

@@ -2,7 +2,15 @@ package caches
 
 import "time"
 
+type ItemType = int
+
+const (
+	ItemTypeFile   ItemType = 1
+	ItemTypeMemory ItemType = 2
+)
+
 type Item struct {
+	Type       ItemType
 	Key        string
 	ExpiredAt  int64
 	HeaderSize int64

internal/caches/storage_file.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
+	"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs/shared"
 	"github.com/TeaOSLab/EdgeNode/internal/events"
 	"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
 	"github.com/TeaOSLab/EdgeNode/internal/utils"
@@ -43,9 +44,10 @@ var (
 //   文件结构：
 //    [expires time] | [ status ] | [url length] | [header length] | [body length] | [url] [header data] [body data]
 type FileStorage struct {
-	policy      *serverconfigs.HTTPCachePolicy
-	cacheConfig *serverconfigs.HTTPFileCacheStorage
-	totalSize   int64
+	policy        *serverconfigs.HTTPCachePolicy
+	cacheConfig   *serverconfigs.HTTPFileCacheStorage // 二级缓存
+	memoryStorage *MemoryStorage                      // 一级缓存
+	totalSize     int64
 
 	list   *List
 	locker sync.RWMutex
@@ -77,6 +79,7 @@ func (this *FileStorage) Init() error {
 	defer this.locker.Unlock()
 
 	before := time.Now()
+	cacheDir := ""
 	defer func() {
 		// 统计
 		count := 0
@@ -98,7 +101,7 @@ func (this *FileStorage) Init() error {
 		} else if size > 1024 {
 			sizeMB = fmt.Sprintf("%.3f K", float64(size)/1024)
 		}
-		remotelogs.Println("CACHE", "init policy "+strconv.FormatInt(this.policy.Id, 10)+", cost: "+fmt.Sprintf("%.2f", cost)+" ms, count: "+strconv.Itoa(count)+", size: "+sizeMB)
+		remotelogs.Println("CACHE", "init policy "+strconv.FormatInt(this.policy.Id, 10)+" from '"+cacheDir+"', cost: "+fmt.Sprintf("%.2f", cost)+" ms, count: "+strconv.Itoa(count)+", size: "+sizeMB)
 	}()
 
 	// 配置
@@ -112,6 +115,7 @@ func (this *FileStorage) Init() error {
 		return err
 	}
 	this.cacheConfig = cacheConfig
+	cacheDir = cacheConfig.Dir
 
 	if !filepath.IsAbs(this.cacheConfig.Dir) {
 		this.cacheConfig.Dir = Tea.Root + Tea.DS + this.cacheConfig.Dir
@@ -142,10 +146,49 @@ func (this *FileStorage) Init() error {
 		return err
 	}
 
+	// 加载内存缓存
+	if this.cacheConfig.MemoryPolicy != nil {
+		memoryCapacity := this.cacheConfig.MemoryPolicy.Capacity
+		if memoryCapacity != nil && memoryCapacity.Count > 0 {
+			memoryPolicy := &serverconfigs.HTTPCachePolicy{
+				Id:          this.policy.Id,
+				IsOn:        this.policy.IsOn,
+				Name:        this.policy.Name,
+				Description: this.policy.Description,
+				Capacity:    memoryCapacity,
+				MaxKeys:     this.policy.MaxKeys,
+				MaxSize:     &shared.SizeCapacity{Count: 128, Unit: shared.SizeCapacityUnitMB}, // TODO 将来可以修改
+				Type:        serverconfigs.CachePolicyStorageMemory,
+				Options:     this.policy.Options,
+				Life:        this.policy.Life,
+				MinLife:     this.policy.MinLife,
+				MaxLife:     this.policy.MaxLife,
+			}
+			err = memoryPolicy.Init()
+			if err != nil {
+				return err
+			}
+			memoryStorage := NewMemoryStorage(memoryPolicy)
+			err = memoryStorage.Init()
+			if err != nil {
+				return err
+			}
+			this.memoryStorage = memoryStorage
+		}
+	}
+
 	return nil
 }
 
 func (this *FileStorage) OpenReader(key string) (Reader, error) {
+	// 先尝试内存缓存
+	if this.memoryStorage != nil {
+		reader, err := this.memoryStorage.OpenReader(key)
+		if err == nil {
+			return reader, err
+		}
+	}
+
 	hash, path := this.keyPath(key)
 	if !this.list.Exist(hash) {
 		return nil, ErrNotFound
@@ -173,6 +216,14 @@ func (this *FileStorage) OpenReader(key string) (Reader, error) {
 
 // 打开缓存文件等待写入
 func (this *FileStorage) OpenWriter(key string, expiredAt int64, status int) (Writer, error) {
+	// 先尝试内存缓存
+	if this.memoryStorage != nil {
+		writer, err := this.memoryStorage.OpenWriter(key, expiredAt, status)
+		if err == nil {
+			return writer, nil
+		}
+	}
+
 	// 检查是否超出最大值
 	if this.policy.MaxKeys > 0 && this.list.Count() > this.policy.MaxKeys {
 		return nil, errors.New("write file cache failed: too many keys in cache storage")
@@ -291,6 +342,13 @@ func (this *FileStorage) OpenWriter(key string, expiredAt int64, status int) (Wr
 
 // 添加到List
 func (this *FileStorage) AddToList(item *Item) {
+	if this.memoryStorage != nil {
+		if item.Type == ItemTypeMemory {
+			this.memoryStorage.AddToList(item)
+			return
+		}
+	}
+
 	item.MetaSize = SizeMeta
 	hash := stringutil.Md5(item.Key)
 	this.list.Add(hash, item)
@@ -301,6 +359,11 @@ func (this *FileStorage) Delete(key string) error {
 	this.locker.Lock()
 	defer this.locker.Unlock()
 
+	// 先尝试内存缓存
+	if this.memoryStorage != nil {
+		_ = this.memoryStorage.Delete(key)
+	}
+
 	hash, path := this.keyPath(key)
 	this.list.Remove(hash)
 	err := os.Remove(path)
@@ -325,6 +388,11 @@ func (this *FileStorage) CleanAll() error {
 	this.locker.Lock()
 	defer this.locker.Unlock()
 
+	// 先尝试内存缓存
+	if this.memoryStorage != nil {
+		_ = this.memoryStorage.CleanAll()
+	}
+
 	this.list.Reset()
 
 	// 删除缓存和目录
@@ -378,6 +446,11 @@ func (this *FileStorage) Purge(keys []string, urlType string) error {
 	this.locker.Lock()
 	defer this.locker.Unlock()
 
+	// 先尝试内存缓存
+	if this.memoryStorage != nil {
+		_ = this.memoryStorage.Purge(keys, urlType)
+	}
+
 	// 目录
 	if urlType == "dir" {
 		resultKeys := []string{}
@@ -412,6 +485,11 @@ func (this *FileStorage) Stop() {
 	this.locker.Lock()
 	defer this.locker.Unlock()
 
+	// 先尝试内存缓存
+	if this.memoryStorage != nil {
+		this.memoryStorage.Stop()
+	}
+
 	this.list.Reset()
 	if this.ticker != nil {
 		this.ticker.Stop()
@@ -518,6 +596,7 @@ func (this *FileStorage) decodeFile(path string) (*Item, error) {
 	}()
 
 	item := &Item{
+		Type:     ItemTypeFile,
 		MetaSize: SizeMeta,
 	}
 

internal/caches/storage_memory.go

@@ -17,6 +17,7 @@ type MemoryItem struct {
 	HeaderValue []byte
 	BodyValue   []byte
 	Status      int
+	IsDone      bool
 }
 
 type MemoryStorage struct {
@@ -67,15 +68,14 @@ func (this *MemoryStorage) OpenReader(key string) (Reader, error) {
 	hash := this.hash(key)
 
 	this.locker.RLock()
+	defer this.locker.RUnlock()
+
 	item := this.valuesMap[hash]
-	if item == nil {
-		this.locker.RUnlock()
+	if item == nil || !item.IsDone {
 		return nil, ErrNotFound
 	}
 
 	if item.ExpiredAt > utils.UnixTime() {
-		this.locker.RUnlock()
-
 		reader := NewMemoryReader(item)
 		err := reader.Init()
 		if err != nil {
@@ -83,7 +83,6 @@ func (this *MemoryStorage) OpenReader(key string) (Reader, error) {
 		}
 		return reader, nil
 	}
-	this.locker.RUnlock()
 
 	_ = this.Delete(key)
 
@@ -190,7 +189,7 @@ func (this *MemoryStorage) hash(key string) uint64 {
 
 // 清理任务
 func (this *MemoryStorage) purgeLoop() {
-	this.list.Purge(1000, func(hash string) {
+	this.list.Purge(2048, func(hash string) {
 		uintHash, err := strconv.ParseUint(hash, 10, 64)
 		if err == nil {
 			this.locker.Lock()

internal/caches/writer.go

@@ -25,4 +25,7 @@ type Writer interface {
 
 	// 过期时间
 	ExpiredAt() int64
+
+	// 内容类型
+	ItemType() ItemType
 }

internal/caches/writer_file.go

@@ -126,3 +126,8 @@ func (this *FileWriter) ExpiredAt() int64 {
 func (this *FileWriter) Key() string {
 	return this.key
 }
+
+// 内容类型
+func (this *FileWriter) ItemType() ItemType {
+	return ItemTypeFile
+}

internal/caches/writer_gzip.go

@@ -67,3 +67,8 @@ func (this *gzipWriter) HeaderSize() int64 {
 func (this *gzipWriter) BodySize() int64 {
 	return this.rawWriter.BodySize()
 }
+
+// 内容类型
+func (this *gzipWriter) ItemType() ItemType {
+	return this.rawWriter.ItemType()
+}

internal/caches/writer_memory.go

@@ -14,10 +14,13 @@ type MemoryWriter struct {
 	headerSize     int64
 	bodySize       int64
 	status         int
+
+	hash uint64
+	item *MemoryItem
 }
 
 func NewMemoryWriter(m map[uint64]*MemoryItem, key string, expiredAt int64, status int, locker *sync.RWMutex) *MemoryWriter {
-	return &MemoryWriter{
+	w := &MemoryWriter{
 		m:              m,
 		key:            key,
 		expiredAt:      expiredAt,
@@ -25,16 +28,20 @@ func NewMemoryWriter(m map[uint64]*MemoryItem, key string, expiredAt int64, stat
 		isFirstWriting: true,
 		status:         status,
 	}
+	w.hash = w.calculateHash(key)
+
+	return w
 }
 
 // 写入数据
 func (this *MemoryWriter) WriteHeader(data []byte) (n int, err error) {
 	this.headerSize += int64(len(data))
 
-	hash := this.hash(this.key)
 	this.locker.Lock()
-	item, ok := this.m[hash]
+	item, ok := this.m[this.hash]
 	if ok {
+		item.IsDone = false
+
 		// 第一次写先清空
 		if this.isFirstWriting {
 			item.HeaderValue = nil
@@ -43,13 +50,13 @@ func (this *MemoryWriter) WriteHeader(data []byte) (n int, err error) {
 		}
 		item.HeaderValue = append(item.HeaderValue, data...)
 	} else {
-		item := &MemoryItem{}
+		item = &MemoryItem{}
 		item.HeaderValue = append([]byte{}, data...)
 		item.ExpiredAt = this.expiredAt
 		item.Status = this.status
-		this.m[hash] = item
 		this.isFirstWriting = false
 	}
+	this.item = item
 	this.locker.Unlock()
 	return len(data), nil
 }
@@ -58,10 +65,11 @@ func (this *MemoryWriter) WriteHeader(data []byte) (n int, err error) {
 func (this *MemoryWriter) Write(data []byte) (n int, err error) {
 	this.bodySize += int64(len(data))
 
-	hash := this.hash(this.key)
 	this.locker.Lock()
-	item, ok := this.m[hash]
+	item, ok := this.m[this.hash]
 	if ok {
+		item.IsDone = false
+
 		// 第一次写先清空
 		if this.isFirstWriting {
 			item.HeaderValue = nil
@@ -70,13 +78,13 @@ func (this *MemoryWriter) Write(data []byte) (n int, err error) {
 		}
 		item.BodyValue = append(item.BodyValue, data...)
 	} else {
-		item := &MemoryItem{}
+		item = &MemoryItem{}
 		item.BodyValue = append([]byte{}, data...)
 		item.ExpiredAt = this.expiredAt
 		item.Status = this.status
-		this.m[hash] = item
 		this.isFirstWriting = false
 	}
+	this.item = item
 	this.locker.Unlock()
 	return len(data), nil
 }
@@ -92,14 +100,22 @@ func (this *MemoryWriter) BodySize() int64 {
 
 // 关闭
 func (this *MemoryWriter) Close() error {
+	if this.item == nil {
+		return nil
+	}
+
+	this.locker.Lock()
+	this.item.IsDone = true
+	this.m[this.hash] = this.item
+	this.locker.Unlock()
+
 	return nil
 }
 
 // 丢弃
 func (this *MemoryWriter) Discard() error {
-	hash := this.hash(this.key)
 	this.locker.Lock()
-	delete(this.m, hash)
+	delete(this.m, this.hash)
 	this.locker.Unlock()
 	return nil
 }
@@ -114,7 +130,12 @@ func (this *MemoryWriter) ExpiredAt() int64 {
 	return this.expiredAt
 }
 
+// 内容类型
+func (this *MemoryWriter) ItemType() ItemType {
+	return ItemTypeMemory
+}
+
 // 计算Key Hash
-func (this *MemoryWriter) hash(key string) uint64 {
+func (this *MemoryWriter) calculateHash(key string) uint64 {
 	return xxhash.Sum64String(key)
 }

internal/nodes/api_stream.go

@@ -182,6 +182,7 @@ func (this *APIStream) handleWriteCache(message *pb.NodeStreamMessage) error {
 		return err
 	}
 	storage.AddToList(&caches.Item{
+		Type:       writer.ItemType(),
 		Key:        msg.Key,
 		ExpiredAt:  expiredAt,
 		HeaderSize: writer.HeaderSize(),
@@ -445,6 +446,7 @@ func (this *APIStream) handlePreheatCache(message *pb.NodeStreamMessage) error {
 						err = writer.Close()
 						if err == nil {
 							storage.AddToList(&caches.Item{
+								Type:      writer.ItemType(),
 								Key:       key,
 								ExpiredAt: expiredAt,
 							})

internal/nodes/http_writer.go

@@ -215,6 +215,7 @@ func (this *HTTPWriter) Close() {
 		err := this.cacheWriter.Close()
 		if err == nil {
 			this.cacheStorage.AddToList(&caches.Item{
+				Type:       this.cacheWriter.ItemType(),
 				Key:        this.cacheWriter.Key(),
 				ExpiredAt:  this.cacheWriter.ExpiredAt(),
 				HeaderSize: this.cacheWriter.HeaderSize(),