优化缓存相关代码

internal/caches/open_file_pool_test.go

@@ -11,7 +11,7 @@ func TestOpenFilePool_Get(t *testing.T) {
 	var pool = caches.NewOpenFilePool("a")
 	t.Log(pool.Filename())
 	t.Log(pool.Get())
-	t.Log(pool.Put(caches.NewOpenFile(nil, nil, []byte{})))
+	t.Log(pool.Put(caches.NewOpenFile(nil, nil, nil, 0)))
 	t.Log(pool.Get())
 	t.Log(pool.Get())
 }

internal/caches/reader_file.go

@@ -67,17 +67,17 @@ func (this *FileReader) InitAutoDiscard(autoDiscard bool) error {
 
 	this.expiresAt = int64(binary.BigEndian.Uint32(buf[:SizeExpiresAt]))
 
-	status := types.Int(string(buf[SizeExpiresAt : SizeExpiresAt+SizeStatus]))
+	status := types.Int(string(buf[OffsetStatus : OffsetStatus+SizeStatus]))
 	if status < 100 || status > 999 {
 		return errors.New("invalid status")
 	}
 	this.status = status
 
 	// URL
-	urlLength := binary.BigEndian.Uint32(buf[SizeExpiresAt+SizeStatus : SizeExpiresAt+SizeStatus+SizeURLLength])
+	urlLength := binary.BigEndian.Uint32(buf[OffsetURLLength : OffsetURLLength+SizeURLLength])
 
 	// header
-	headerSize := int(binary.BigEndian.Uint32(buf[SizeExpiresAt+SizeStatus+SizeURLLength : SizeExpiresAt+SizeStatus+SizeURLLength+SizeHeaderLength]))
+	headerSize := int(binary.BigEndian.Uint32(buf[OffsetHeaderLength : OffsetHeaderLength+SizeHeaderLength]))
 	if headerSize == 0 {
 		return nil
 	}
@@ -86,7 +86,7 @@ func (this *FileReader) InitAutoDiscard(autoDiscard bool) error {
 
 	// body
 	this.bodyOffset = this.headerOffset + int64(headerSize)
-	bodySize := int(binary.BigEndian.Uint64(buf[SizeExpiresAt+SizeStatus+SizeURLLength+SizeHeaderLength : SizeExpiresAt+SizeStatus+SizeURLLength+SizeHeaderLength+SizeBodyLength]))
+	bodySize := int(binary.BigEndian.Uint64(buf[OffsetBodyLength : OffsetBodyLength+SizeBodyLength]))
 	if bodySize == 0 {
 		isOk = true
 		return nil
@@ -353,7 +353,9 @@ func (this *FileReader) Close() error {
 		if this.openFile != nil {
 			this.openFileCache.Put(this.fp.Name(), this.openFile)
 		} else {
-			this.openFileCache.Put(this.fp.Name(), NewOpenFile(this.fp, this.meta, this.header, this.LastModified()))
+			var cacheMeta = make([]byte, len(this.meta))
+			copy(cacheMeta, this.meta)
+			this.openFileCache.Put(this.fp.Name(), NewOpenFile(this.fp, cacheMeta, this.header, this.LastModified()))
 		}
 		return nil
 	}

internal/caches/reader_memory.go

@@ -25,7 +25,7 @@ func (this *MemoryReader) TypeName() string {
 }
 
 func (this *MemoryReader) ExpiresAt() int64 {
-	return this.item.ExpiredAt
+	return this.item.ExpiresAt
 }
 
 func (this *MemoryReader) Status() int {

internal/caches/reader_memory_test.go

@@ -4,7 +4,7 @@ import "testing"
 
 func TestMemoryReader_Header(t *testing.T) {
 	item := &MemoryItem{
-		ExpiredAt:   0,
+		ExpiresAt:   0,
 		HeaderValue: []byte("0123456789"),
 		BodyValue:   nil,
 		Status:      2000,
@@ -22,7 +22,7 @@ func TestMemoryReader_Header(t *testing.T) {
 
 func TestMemoryReader_Body(t *testing.T) {
 	item := &MemoryItem{
-		ExpiredAt:   0,
+		ExpiresAt:   0,
 		HeaderValue: nil,
 		BodyValue:   []byte("0123456789"),
 		Status:      2000,
@@ -40,7 +40,7 @@ func TestMemoryReader_Body(t *testing.T) {
 
 func TestMemoryReader_Body_Range(t *testing.T) {
 	item := &MemoryItem{
-		ExpiredAt:   0,
+		ExpiresAt:   0,
 		HeaderValue: nil,
 		BodyValue:   []byte("0123456789"),
 		Status:      2000,

internal/caches/storage_file.go

@@ -38,12 +38,18 @@ import (
 
 const (
 	SizeExpiresAt      = 4
+	OffsetExpiresAt    = 0
 	SizeStatus         = 3
+	OffsetStatus       = SizeExpiresAt
 	SizeURLLength      = 4
+	OffsetURLLength    = OffsetStatus + SizeStatus
 	SizeHeaderLength   = 4
+	OffsetHeaderLength = OffsetURLLength + SizeURLLength
 	SizeBodyLength     = 8
+	OffsetBodyLength   = OffsetHeaderLength + SizeHeaderLength
 
 	SizeMeta  = SizeExpiresAt + SizeStatus + SizeURLLength + SizeHeaderLength + SizeBodyLength
+	OffsetKey = SizeMeta
 )
 
 const (
@@ -56,6 +62,8 @@ const (
 var sharedWritingFileKeyMap = map[string]zero.Zero{} // key => bool
 var sharedWritingFileKeyLocker = sync.Mutex{}
 
+var maxOpenFiles = 2
+
 // FileStorage 文件缓存
 //   文件结构：
 //    [expires time] | [ status ] | [url length] | [header length] | [body length] | [url] [header data] [body data]
@@ -369,7 +377,16 @@ func (this *FileStorage) openReader(key string, allowMemory bool, useStale bool,
 }
 
 // OpenWriter 打开缓存文件等待写入
-func (this *FileStorage) OpenWriter(key string, expiredAt int64, status int, size int64, maxSize int64, isPartial bool) (Writer, error) {
+func (this *FileStorage) OpenWriter(key string, expiresAt int64, status int, size int64, maxSize int64, isPartial bool) (Writer, error) {
+	return this.openWriter(key, expiresAt, status, size, maxSize, isPartial, false)
+}
+
+// OpenFlushWriter 打开从其他媒介直接刷入的写入器
+func (this *FileStorage) OpenFlushWriter(key string, expiresAt int64, status int) (Writer, error) {
+	return this.openWriter(key, expiresAt, status, -1, -1, false, true)
+}
+
+func (this *FileStorage) openWriter(key string, expiredAt int64, status int, size int64, maxSize int64, isPartial bool, isFlushing bool) (Writer, error) {
 	// 是否正在退出
 	if teaconst.IsQuiting {
 		return nil, ErrWritingUnavailable
@@ -387,7 +404,7 @@ func (this *FileStorage) OpenWriter(key string, expiredAt int64, status int, siz
 		maxMemorySize = maxSize
 	}
 	var memoryStorage = this.memoryStorage
-	if !isPartial && memoryStorage != nil && ((size > 0 && size < maxMemorySize) || size < 0) {
+	if !isFlushing && !isPartial && memoryStorage != nil && ((size > 0 && size < maxMemorySize) || size < 0) {
 		writer, err := memoryStorage.OpenWriter(key, expiredAt, status, size, maxMemorySize, false)
 		if err == nil {
 			return writer, nil
@@ -407,6 +424,12 @@ func (this *FileStorage) OpenWriter(key string, expiredAt int64, status int, siz
 		sharedWritingFileKeyLocker.Unlock()
 		return nil, ErrFileIsWriting
 	}
+
+	if len(sharedWritingFileKeyMap) > maxOpenFiles {
+		sharedWritingFileKeyLocker.Unlock()
+		return nil, ErrTooManyOpenFiles
+	}
+
 	sharedWritingFileKeyMap[key] = zero.New()
 	sharedWritingFileKeyLocker.Unlock()
 	defer func() {
@@ -523,54 +546,28 @@ func (this *FileStorage) OpenWriter(key string, expiredAt int64, status int, siz
 		}
 
 		// 写入过期时间
-		bytes4 := make([]byte, 4)
+		var metaBytes = make([]byte, SizeMeta+len(key))
-		{
+		binary.BigEndian.PutUint32(metaBytes[OffsetExpiresAt:], uint32(expiredAt))
-			binary.BigEndian.PutUint32(bytes4, uint32(expiredAt))
-			_, err = writer.Write(bytes4)
-			if err != nil {
-				return nil, err
-			}
-		}
 
 		// 写入状态码
 		if status > 999 || status < 100 {
 			status = 200
 		}
-		_, err = writer.WriteString(strconv.Itoa(status))
+		copy(metaBytes[OffsetStatus:], strconv.Itoa(status))
-		if err != nil {
-			return nil, err
-		}
 
 		// 写入URL长度
-		{
+		binary.BigEndian.PutUint32(metaBytes[OffsetURLLength:], uint32(len(key)))
-			binary.BigEndian.PutUint32(bytes4, uint32(len(key)))
-			_, err = writer.Write(bytes4)
-			if err != nil {
-				return nil, err
-			}
-		}
 
 		// 写入Header Length
-		{
+		binary.BigEndian.PutUint32(metaBytes[OffsetHeaderLength:], uint32(0))
-			binary.BigEndian.PutUint32(bytes4, uint32(0))
-			_, err = writer.Write(bytes4)
-			if err != nil {
-				return nil, err
-			}
-		}
 
 		// 写入Body Length
-		{
+		binary.BigEndian.PutUint64(metaBytes[OffsetBodyLength:], uint64(0))
-			b := make([]byte, SizeBodyLength)
-			binary.BigEndian.PutUint64(b, uint64(0))
-			_, err = writer.Write(b)
-			if err != nil {
-				return nil, err
-			}
-		}
 
 		// 写入URL
-		_, err = writer.WriteString(key)
+		copy(metaBytes[OffsetKey:], key)
+
+		_, err = writer.Write(metaBytes)
 		if err != nil {
 			return nil, err
 		}

internal/caches/storage_interface.go

@@ -14,7 +14,10 @@ type StorageInterface interface {
 
 	// OpenWriter 打开缓存写入器等待写入
 	// size 和 maxSize 可能为-1
-	OpenWriter(key string, expiredAt int64, status int, size int64, maxSize int64, isPartial bool) (Writer, error)
+	OpenWriter(key string, expiresAt int64, status int, size int64, maxSize int64, isPartial bool) (Writer, error)
+
+	// OpenFlushWriter 打开从其他媒介直接刷入的写入器
+	OpenFlushWriter(key string, expiresAt int64, status int) (Writer, error)
 
 	// Delete 删除某个键值对应的缓存
 	Delete(key string) error

internal/caches/storage_memory.go

@@ -23,7 +23,7 @@ import (
 )
 
 type MemoryItem struct {
-	ExpiredAt   int64
+	ExpiresAt   int64
 	HeaderValue []byte
 	BodyValue   []byte
 	Status      int
@@ -32,7 +32,7 @@ type MemoryItem struct {
 }
 
 func (this *MemoryItem) IsExpired() bool {
-	return this.ExpiredAt < utils.UnixTime()
+	return this.ExpiresAt < utils.UnixTime()
 }
 
 type MemoryStorage struct {
@@ -118,7 +118,7 @@ func (this *MemoryStorage) OpenReader(key string, useStale bool, isPartial bool)
 		return nil, ErrNotFound
 	}
 
-	if useStale || (item.ExpiredAt > utils.UnixTime()) {
+	if useStale || (item.ExpiresAt > utils.UnixTime()) {
 		reader := NewMemoryReader(item)
 		err := reader.Init()
 		if err != nil {
@@ -160,7 +160,12 @@ func (this *MemoryStorage) OpenWriter(key string, expiredAt int64, status int, s
 	return this.openWriter(key, expiredAt, status, size, maxSize, true)
 }
 
-func (this *MemoryStorage) openWriter(key string, expiredAt int64, status int, size int64, maxSize int64, isDirty bool) (Writer, error) {
+// OpenFlushWriter 打开从其他媒介直接刷入的写入器
+func (this *MemoryStorage) OpenFlushWriter(key string, expiresAt int64, status int) (Writer, error) {
+	return this.openWriter(key, expiresAt, status, -1, -1, true)
+}
+
+func (this *MemoryStorage) openWriter(key string, expiresAt int64, status int, size int64, maxSize int64, isDirty bool) (Writer, error) {
 	// 待写入队列是否已满
 	if isDirty &&
 		this.parentStorage != nil &&
@@ -215,7 +220,7 @@ func (this *MemoryStorage) openWriter(key string, expiredAt int64, status int, s
 	}
 
 	isWriting = true
-	return NewMemoryWriter(this, key, expiredAt, status, isDirty, maxSize, func() {
+	return NewMemoryWriter(this, key, expiresAt, status, isDirty, maxSize, func() {
 		this.locker.Lock()
 		delete(this.writingKeyMap, key)
 		this.locker.Unlock()
@@ -471,7 +476,7 @@ func (this *MemoryStorage) flushItem(key string) {
 		return
 	}
 
-	writer, err := this.parentStorage.OpenWriter(key, item.ExpiredAt, item.Status, -1, -1, false)
+	writer, err := this.parentStorage.OpenFlushWriter(key, item.ExpiresAt, item.Status)
 	if err != nil {
 		if !CanIgnoreErr(err) {
 			remotelogs.Error("CACHE", "flush items failed: open writer failed: "+err.Error())
@@ -503,7 +508,7 @@ func (this *MemoryStorage) flushItem(key string) {
 	this.parentStorage.AddToList(&Item{
 		Type:       writer.ItemType(),
 		Key:        key,
-		ExpiredAt:  item.ExpiredAt,
+		ExpiredAt:  item.ExpiresAt,
 		HeaderSize: writer.HeaderSize(),
 		BodySize:   writer.BodySize(),
 	})

internal/caches/writer_memory.go

@@ -30,7 +30,7 @@ func NewMemoryWriter(memoryStorage *MemoryStorage, key string, expiredAt int64,
 		key:       key,
 		expiredAt: expiredAt,
 		item: &MemoryItem{
-			ExpiredAt:  expiredAt,
+			ExpiresAt:  expiredAt,
 			ModifiedAt: time.Now().Unix(),
 			Status:     status,
 		},

internal/nodes/http_writer.go

@@ -313,6 +313,8 @@ func (this *HTTPWriter) PrepareCache(resp *http.Response, size int64) {
 		if !caches.CanIgnoreErr(err) {
 			remotelogs.Error("HTTP_WRITER", "write cache failed: "+err.Error())
 		}
+
+		this.Header().Set("X-Cache", "BYPASS, open failed")
 		return
 	}
 	this.cacheWriter = cacheWriter