TRKJ/EdgeNode
2
0
Fork 0
mirror of https://github.com/TeaOSLab/EdgeNode.git synced 2025-11-04 16:00:25 +08:00
Code Issues Packages Projects Releases Wiki Activity

减少文件缓存写入次数

Browse Source
This commit is contained in:
GoEdgeLab
2022-11-19 17:23:45 +08:00
parent 6cc70dc8e5
commit 00cce8572f
9 changed files with 101 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
internal/caches/reader_file_test.go
View File

@@ -19,7 +19,7 @@ func TestFileReader(t *testing.T) {
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
_, path := storage.keyPath("my-key") _, path, _ := storage.keyPath("my-key")
fp, err := os.Open(path) fp, err := os.Open(path)
if err != nil { if err != nil {
@@ -105,7 +105,7 @@ func TestFileReader_Range(t *testing.T) {
} }
_ = writer.Close()**/ _ = writer.Close()**/
_, path := storage.keyPath("my-number") _, path, _ := storage.keyPath("my-number")
fp, err := os.Open(path) fp, err := os.Open(path)
if err != nil { if err != nil {

33
internal/caches/storage_file.go
View File

@@ -50,7 +50,6 @@ const (
OffsetBodyLength = OffsetHeaderLength + SizeHeaderLength OffsetBodyLength = OffsetHeaderLength + SizeHeaderLength
SizeMeta = SizeExpiresAt + SizeStatus + SizeURLLength + SizeHeaderLength + SizeBodyLength SizeMeta = SizeExpiresAt + SizeStatus + SizeURLLength + SizeHeaderLength + SizeBodyLength
OffsetKey = SizeMeta
) )
const ( const (
@@ -415,16 +414,16 @@ func (this *FileStorage) openReader(key string, allowMemory bool, useStale bool,
} }
// OpenWriter 打开缓存文件等待写入 // OpenWriter 打开缓存文件等待写入
func (this *FileStorage) OpenWriter(key string, expiresAt int64, status int, size int64, maxSize int64, isPartial bool) (Writer, error) { func (this *FileStorage) OpenWriter(key string, expiresAt int64, status int, headerSize int, bodySize int64, maxSize int64, isPartial bool) (Writer, error) {
return this.openWriter(key, expiresAt, status, size, maxSize, isPartial, false) return this.openWriter(key, expiresAt, status, headerSize, bodySize, maxSize, isPartial, false)
} }
// OpenFlushWriter 打开从其他媒介直接刷入的写入器 // OpenFlushWriter 打开从其他媒介直接刷入的写入器
func (this *FileStorage) OpenFlushWriter(key string, expiresAt int64, status int) (Writer, error) { func (this *FileStorage) OpenFlushWriter(key string, expiresAt int64, status int, headerSize int, bodySize int64) (Writer, error) {
return this.openWriter(key, expiresAt, status, -1, -1, false, true) return this.openWriter(key, expiresAt, status, headerSize, bodySize, -1, false, true)
} }
func (this *FileStorage) openWriter(key string, expiredAt int64, status int, size int64, maxSize int64, isPartial bool, isFlushing bool) (Writer, error) { func (this *FileStorage) openWriter(key string, expiredAt int64, status int, headerSize int, bodySize int64, maxSize int64, isPartial bool, isFlushing bool) (Writer, error) {
// 是否正在退出 // 是否正在退出
if teaconst.IsQuiting { if teaconst.IsQuiting {
return nil, ErrWritingUnavailable return nil, ErrWritingUnavailable
@@ -442,8 +441,8 @@ func (this *FileStorage) openWriter(key string, expiredAt int64, status int, siz
maxMemorySize = maxSize maxMemorySize = maxSize
} }
var memoryStorage = this.memoryStorage var memoryStorage = this.memoryStorage
if !isFlushing && !isPartial && memoryStorage != nil && ((size > 0 && size < maxMemorySize) || size < 0) { if !isFlushing && !isPartial && memoryStorage != nil && ((bodySize > 0 && bodySize < maxMemorySize) || bodySize < 0) {
writer, err := memoryStorage.OpenWriter(key, expiredAt, status, size, maxMemorySize, false) writer, err := memoryStorage.OpenWriter(key, expiredAt, status, headerSize, bodySize, maxMemorySize, false)
if err == nil { if err == nil {
return writer, nil return writer, nil
} }
@@ -608,6 +607,8 @@ func (this *FileStorage) openWriter(key string, expiredAt int64, status int, siz
return nil, ErrFileIsWriting return nil, ErrFileIsWriting
} }
var metaBodySize int64 = -1
var metaHeaderSize int = -1
if isNewCreated { if isNewCreated {
// 写入meta // 写入meta
// 从v0.5.8开始不再在meta中写入Key // 从v0.5.8开始不再在meta中写入Key
@@ -620,6 +621,18 @@ func (this *FileStorage) openWriter(key string, expiredAt int64, status int, siz
} }
copy(metaBytes[OffsetStatus:], strconv.Itoa(status)) copy(metaBytes[OffsetStatus:], strconv.Itoa(status))
// 写入Header Length
if headerSize > 0 {
binary.BigEndian.PutUint32(metaBytes[OffsetHeaderLength:], uint32(headerSize))
metaHeaderSize = headerSize
}
// 写入Body Length
if bodySize > 0 {
binary.BigEndian.PutUint64(metaBytes[OffsetBodyLength:], uint64(bodySize))
metaBodySize = bodySize
}
_, err = writer.Write(metaBytes) _, err = writer.Write(metaBytes)
if err != nil { if err != nil {
return nil, err return nil, err
@@ -642,7 +655,7 @@ func (this *FileStorage) openWriter(key string, expiredAt int64, status int, siz
sharedWritingFileKeyLocker.Unlock() sharedWritingFileKeyLocker.Unlock()
}), nil }), nil
} else { } else {
return NewFileWriter(this, writer, key, expiredAt, -1, func() { return NewFileWriter(this, writer, key, expiredAt, metaHeaderSize, metaBodySize, -1, func() {
sharedWritingFileKeyLocker.Lock() sharedWritingFileKeyLocker.Lock()
delete(sharedWritingFileKeyMap, key) delete(sharedWritingFileKeyMap, key)
if len(sharedWritingFileKeyMap) == 0 { if len(sharedWritingFileKeyMap) == 0 {
@@ -1140,7 +1153,7 @@ func (this *FileStorage) hotLoop() {
expiresAt = bestExpiresAt expiresAt = bestExpiresAt
} }
writer, err := memoryStorage.openWriter(item.Key, expiresAt, reader.Status(), reader.BodySize(), -1, false) writer, err := memoryStorage.openWriter(item.Key, expiresAt, reader.Status(), types.Int(reader.HeaderSize()), reader.BodySize(), -1, false)
if err != nil { if err != nil {
if !CanIgnoreErr(err) { if !CanIgnoreErr(err) {
remotelogs.Error("CACHE", "transfer hot item failed: "+err.Error()) remotelogs.Error("CACHE", "transfer hot item failed: "+err.Error())

14
internal/caches/storage_file_test.go
View File

@@ -62,7 +62,7 @@ func TestFileStorage_OpenWriter(t *testing.T) {
header := []byte("Header") header := []byte("Header")
body := []byte("This is Body") body := []byte("This is Body")
writer, err := storage.OpenWriter("my-key", time.Now().Unix()+86400, 200, -1, -1, false) writer, err := storage.OpenWriter("my-key", time.Now().Unix()+86400, 200, -1, -1, -1, false)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
@@ -100,7 +100,7 @@ func TestFileStorage_OpenWriter_Partial(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
writer, err := storage.OpenWriter("my-key", time.Now().Unix()+86400, 200, -1, -1, true) writer, err := storage.OpenWriter("my-key", time.Now().Unix()+86400, 200, -1, -1, -1, true)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
@@ -139,7 +139,7 @@ func TestFileStorage_OpenWriter_HTTP(t *testing.T) {
t.Log(time.Since(now).Seconds()*1000, "ms") t.Log(time.Since(now).Seconds()*1000, "ms")
}() }()
writer, err := storage.OpenWriter("my-http-response", time.Now().Unix()+86400, 200, -1, -1, false) writer, err := storage.OpenWriter("my-http-response", time.Now().Unix()+86400, 200, -1, -1, -1, false)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
@@ -212,7 +212,7 @@ func TestFileStorage_Concurrent_Open_DifferentFile(t *testing.T) {
go func(i int) { go func(i int) {
defer wg.Done() defer wg.Done()
writer, err := storage.OpenWriter("abc"+strconv.Itoa(i), time.Now().Unix()+3600, 200, -1, -1, false) writer, err := storage.OpenWriter("abc"+strconv.Itoa(i), time.Now().Unix()+3600, 200, -1, -1, -1, false)
if err != nil { if err != nil {
if err != ErrFileIsWriting { if err != ErrFileIsWriting {
t.Error(err) t.Error(err)
@@ -267,7 +267,7 @@ func TestFileStorage_Concurrent_Open_SameFile(t *testing.T) {
go func(i int) { go func(i int) {
defer wg.Done() defer wg.Done()
writer, err := storage.OpenWriter("abc"+strconv.Itoa(0), time.Now().Unix()+3600, 200, -1, -1, false) writer, err := storage.OpenWriter("abc"+strconv.Itoa(0), time.Now().Unix()+3600, 200, -1, -1, -1, false)
if err != nil { if err != nil {
if err != ErrFileIsWriting { if err != ErrFileIsWriting {
t.Error(err) t.Error(err)
@@ -522,7 +522,7 @@ func TestFileStorage_DecodeFile(t *testing.T) {
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
_, path := storage.keyPath("my-key") _, path, _ := storage.keyPath("my-key")
t.Log(path) t.Log(path)
} }
@@ -569,6 +569,6 @@ func BenchmarkFileStorage_KeyPath(b *testing.B) {
} }
for i := 0; i < b.N; i++ { for i := 0; i < b.N; i++ {
_, _ = storage.keyPath(strconv.Itoa(i)) _, _, _ = storage.keyPath(strconv.Itoa(i))
} }
} }

4
internal/caches/storage_interface.go
View File

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

18
internal/caches/storage_memory.go
View File

@@ -149,7 +149,7 @@ func (this *MemoryStorage) OpenReader(key string, useStale bool, isPartial bool)
} }
// OpenWriter 打开缓存写入器等待写入 // OpenWriter 打开缓存写入器等待写入
func (this *MemoryStorage) OpenWriter(key string, expiredAt int64, status int, size int64, maxSize int64, isPartial bool) (Writer, error) { func (this *MemoryStorage) OpenWriter(key string, expiredAt int64, status int, headerSize int, bodySize int64, maxSize int64, isPartial bool) (Writer, error) {
if this.ignoreKeys.Has(key) { if this.ignoreKeys.Has(key) {
return nil, ErrEntityTooLarge return nil, ErrEntityTooLarge
} }
@@ -158,15 +158,15 @@ func (this *MemoryStorage) OpenWriter(key string, expiredAt int64, status int, s
if isPartial { if isPartial {
return nil, ErrFileIsWriting return nil, ErrFileIsWriting
} }
return this.openWriter(key, expiredAt, status, size, maxSize, true) return this.openWriter(key, expiredAt, status, headerSize, bodySize, maxSize, true)
} }
// OpenFlushWriter 打开从其他媒介直接刷入的写入器 // OpenFlushWriter 打开从其他媒介直接刷入的写入器
func (this *MemoryStorage) OpenFlushWriter(key string, expiresAt int64, status int) (Writer, error) { func (this *MemoryStorage) OpenFlushWriter(key string, expiresAt int64, status int, headerSize int, bodySize int64) (Writer, error) {
return this.openWriter(key, expiresAt, status, -1, -1, true) return this.openWriter(key, expiresAt, status, headerSize, bodySize, -1, true)
} }
func (this *MemoryStorage) openWriter(key string, expiresAt int64, status int, size int64, maxSize int64, isDirty bool) (Writer, error) { func (this *MemoryStorage) openWriter(key string, expiresAt int64, status int, headerSize int, bodySize int64, maxSize int64, isDirty bool) (Writer, error) {
// 待写入队列是否已满 // 待写入队列是否已满
if isDirty && if isDirty &&
this.parentStorage != nil && this.parentStorage != nil &&
@@ -207,10 +207,10 @@ func (this *MemoryStorage) openWriter(key string, expiresAt int64, status int, s
return nil, NewCapacityError("write memory cache failed: too many keys in cache storage") return nil, NewCapacityError("write memory cache failed: too many keys in cache storage")
} }
capacityBytes := this.memoryCapacityBytes() capacityBytes := this.memoryCapacityBytes()
if size < 0 { if bodySize < 0 {
size = 0 bodySize = 0
} }
if capacityBytes > 0 && capacityBytes <= this.totalSize+size { if capacityBytes > 0 && capacityBytes <= this.totalSize+bodySize {
return nil, NewCapacityError("write memory cache failed: over memory size: " + strconv.FormatInt(capacityBytes, 10) + ", current size: " + strconv.FormatInt(this.totalSize, 10) + " bytes") return nil, NewCapacityError("write memory cache failed: over memory size: " + strconv.FormatInt(capacityBytes, 10) + ", current size: " + strconv.FormatInt(this.totalSize, 10) + " bytes")
} }
@@ -481,7 +481,7 @@ func (this *MemoryStorage) flushItem(key string) {
return return
} }
writer, err := this.parentStorage.OpenFlushWriter(key, item.ExpiresAt, item.Status) writer, err := this.parentStorage.OpenFlushWriter(key, item.ExpiresAt, item.Status, len(item.HeaderValue), int64(len(item.BodyValue)))
if err != nil { if err != nil {
if !CanIgnoreErr(err) { if !CanIgnoreErr(err) {
remotelogs.Error("CACHE", "flush items failed: open writer failed: "+err.Error()) remotelogs.Error("CACHE", "flush items failed: open writer failed: "+err.Error())

22
internal/caches/storage_memory_test.go
View File

@@ -16,7 +16,7 @@ import (
func TestMemoryStorage_OpenWriter(t *testing.T) { func TestMemoryStorage_OpenWriter(t *testing.T) {
storage := NewMemoryStorage(&serverconfigs.HTTPCachePolicy{}, nil) storage := NewMemoryStorage(&serverconfigs.HTTPCachePolicy{}, nil)
writer, err := storage.OpenWriter("abc", time.Now().Unix()+60, 200, -1, -1, false) writer, err := storage.OpenWriter("abc", time.Now().Unix()+60, 200, -1, -1, -1, false)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
@@ -63,7 +63,7 @@ func TestMemoryStorage_OpenWriter(t *testing.T) {
} }
} }
writer, err = storage.OpenWriter("abc", time.Now().Unix()+60, 200, -1, -1, false) writer, err = storage.OpenWriter("abc", time.Now().Unix()+60, 200, -1, -1, -1, false)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
@@ -104,7 +104,7 @@ func TestMemoryStorage_OpenReaderLock(t *testing.T) {
func TestMemoryStorage_Delete(t *testing.T) { func TestMemoryStorage_Delete(t *testing.T) {
storage := NewMemoryStorage(&serverconfigs.HTTPCachePolicy{}, nil) storage := NewMemoryStorage(&serverconfigs.HTTPCachePolicy{}, nil)
{ {
writer, err := storage.OpenWriter("abc", time.Now().Unix()+60, 200, -1, -1, false) writer, err := storage.OpenWriter("abc", time.Now().Unix()+60, 200, -1, -1, -1, false)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
@@ -112,7 +112,7 @@ func TestMemoryStorage_Delete(t *testing.T) {
t.Log(len(storage.valuesMap)) t.Log(len(storage.valuesMap))
} }
{ {
writer, err := storage.OpenWriter("abc1", time.Now().Unix()+60, 200, -1, -1, false) writer, err := storage.OpenWriter("abc1", time.Now().Unix()+60, 200, -1, -1, -1, false)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
@@ -127,7 +127,7 @@ func TestMemoryStorage_Stat(t *testing.T) {
storage := NewMemoryStorage(&serverconfigs.HTTPCachePolicy{}, nil) storage := NewMemoryStorage(&serverconfigs.HTTPCachePolicy{}, nil)
expiredAt := time.Now().Unix() + 60 expiredAt := time.Now().Unix() + 60
{ {
writer, err := storage.OpenWriter("abc", expiredAt, 200, -1, -1, false) writer, err := storage.OpenWriter("abc", expiredAt, 200, -1, -1, -1, false)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
@@ -140,7 +140,7 @@ func TestMemoryStorage_Stat(t *testing.T) {
}) })
} }
{ {
writer, err := storage.OpenWriter("abc1", expiredAt, 200, -1, -1, false) writer, err := storage.OpenWriter("abc1", expiredAt, 200, -1, -1, -1, false)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
@@ -164,7 +164,7 @@ func TestMemoryStorage_CleanAll(t *testing.T) {
storage := NewMemoryStorage(&serverconfigs.HTTPCachePolicy{}, nil) storage := NewMemoryStorage(&serverconfigs.HTTPCachePolicy{}, nil)
expiredAt := time.Now().Unix() + 60 expiredAt := time.Now().Unix() + 60
{ {
writer, err := storage.OpenWriter("abc", expiredAt, 200, -1, -1, false) writer, err := storage.OpenWriter("abc", expiredAt, 200, -1, -1, -1, false)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
@@ -176,7 +176,7 @@ func TestMemoryStorage_CleanAll(t *testing.T) {
}) })
} }
{ {
writer, err := storage.OpenWriter("abc1", expiredAt, 200, -1, -1, false) writer, err := storage.OpenWriter("abc1", expiredAt, 200, -1, -1, -1, false)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
@@ -199,7 +199,7 @@ func TestMemoryStorage_Purge(t *testing.T) {
storage := NewMemoryStorage(&serverconfigs.HTTPCachePolicy{}, nil) storage := NewMemoryStorage(&serverconfigs.HTTPCachePolicy{}, nil)
expiredAt := time.Now().Unix() + 60 expiredAt := time.Now().Unix() + 60
{ {
writer, err := storage.OpenWriter("abc", expiredAt, 200, -1, -1, false) writer, err := storage.OpenWriter("abc", expiredAt, 200, -1, -1, -1, false)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
@@ -211,7 +211,7 @@ func TestMemoryStorage_Purge(t *testing.T) {
}) })
} }
{ {
writer, err := storage.OpenWriter("abc1", expiredAt, 200, -1, -1, false) writer, err := storage.OpenWriter("abc1", expiredAt, 200, -1, -1, -1, false)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
@@ -242,7 +242,7 @@ func TestMemoryStorage_Expire(t *testing.T) {
for i := 0; i < 1000; i++ { for i := 0; i < 1000; i++ {
expiredAt := time.Now().Unix() + int64(rands.Int(0, 60)) expiredAt := time.Now().Unix() + int64(rands.Int(0, 60))
key := "abc" + strconv.Itoa(i) key := "abc" + strconv.Itoa(i)
writer, err := storage.OpenWriter(key, expiredAt, 200, -1, -1, false) writer, err := storage.OpenWriter(key, expiredAt, 200, -1, -1, -1, false)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }

21
internal/caches/writer_file.go
View File

@@ -14,15 +14,20 @@ type FileWriter struct {
storage StorageInterface storage StorageInterface
rawWriter *os.File rawWriter *os.File
key string key string
metaHeaderSize int
headerSize int64 headerSize int64
metaBodySize int64 // 写入前的内容长度
bodySize int64 bodySize int64
expiredAt int64 expiredAt int64
maxSize int64 maxSize int64
endFunc func() endFunc func()
once sync.Once once sync.Once
} }
func NewFileWriter(storage StorageInterface, rawWriter *os.File, key string, expiredAt int64, maxSize int64, endFunc func()) *FileWriter { func NewFileWriter(storage StorageInterface, rawWriter *os.File, key string, expiredAt int64, metaHeaderSize int, metaBodySize int64, maxSize int64, endFunc func()) *FileWriter {
return &FileWriter{ return &FileWriter{
storage: storage, storage: storage,
key: key, key: key,
@@ -30,6 +35,8 @@ func NewFileWriter(storage StorageInterface, rawWriter *os.File, key string, exp
expiredAt: expiredAt, expiredAt: expiredAt,
maxSize: maxSize, maxSize: maxSize,
endFunc: endFunc, endFunc: endFunc,
metaHeaderSize: metaHeaderSize,
metaBodySize: metaBodySize,
} }
} }
@@ -45,7 +52,10 @@ func (this *FileWriter) WriteHeader(data []byte) (n int, err error) {
// WriteHeaderLength 写入Header长度数据 // WriteHeaderLength 写入Header长度数据
func (this *FileWriter) WriteHeaderLength(headerLength int) error { func (this *FileWriter) WriteHeaderLength(headerLength int) error {
bytes4 := make([]byte, 4) if this.metaHeaderSize > 0 && this.metaHeaderSize == headerLength {
return nil
}
var bytes4 = make([]byte, 4)
binary.BigEndian.PutUint32(bytes4, uint32(headerLength)) binary.BigEndian.PutUint32(bytes4, uint32(headerLength))
_, err := this.rawWriter.Seek(SizeExpiresAt+SizeStatus+SizeURLLength, io.SeekStart) _, err := this.rawWriter.Seek(SizeExpiresAt+SizeStatus+SizeURLLength, io.SeekStart)
if err != nil { if err != nil {
@@ -88,7 +98,10 @@ func (this *FileWriter) WriteAt(offset int64, data []byte) error {
// WriteBodyLength 写入Body长度数据 // WriteBodyLength 写入Body长度数据
func (this *FileWriter) WriteBodyLength(bodyLength int64) error { func (this *FileWriter) WriteBodyLength(bodyLength int64) error {
bytes8 := make([]byte, 8) if this.metaBodySize >= 0 && bodyLength == this.metaBodySize {
return nil
}
var bytes8 = make([]byte, 8)
binary.BigEndian.PutUint64(bytes8, uint64(bodyLength)) binary.BigEndian.PutUint64(bytes8, uint64(bodyLength))
_, err := this.rawWriter.Seek(SizeExpiresAt+SizeStatus+SizeURLLength+SizeHeaderLength, io.SeekStart) _, err := this.rawWriter.Seek(SizeExpiresAt+SizeStatus+SizeURLLength+SizeHeaderLength, io.SeekStart)
if err != nil { if err != nil {
@@ -109,7 +122,7 @@ func (this *FileWriter) Close() error {
this.endFunc() this.endFunc()
}) })
path := this.rawWriter.Name() var path = this.rawWriter.Name()
err := this.WriteHeaderLength(types.Int(this.headerSize)) err := this.WriteHeaderLength(types.Int(this.headerSize))
if err != nil { if err != nil {

2
internal/nodes/api_stream.go
View File

@@ -178,7 +178,7 @@ func (this *APIStream) handleWriteCache(message *pb.NodeStreamMessage) error {
} }
expiredAt := time.Now().Unix() + msg.LifeSeconds expiredAt := time.Now().Unix() + msg.LifeSeconds
writer, err := storage.OpenWriter(msg.Key, expiredAt, 200, int64(len(msg.Value)), -1, false) writer, err := storage.OpenWriter(msg.Key, expiredAt, 200, -1, int64(len(msg.Value)), -1, false)
if err != nil { if err != nil {
this.replyFail(message.RequestId, "prepare writing failed: "+err.Error()) this.replyFail(message.RequestId, "prepare writing failed: "+err.Error())
return err return err

16
internal/nodes/http_writer.go
View File

@@ -303,7 +303,7 @@ func (this *HTTPWriter) PrepareCache(resp *http.Response, size int64) {
if this.isPartial { if this.isPartial {
cacheKey += caches.SuffixPartial cacheKey += caches.SuffixPartial
} }
cacheWriter, err := storage.OpenWriter(cacheKey, expiresAt, this.StatusCode(), size, cacheRef.MaxSizeBytes(), this.isPartial) cacheWriter, err := storage.OpenWriter(cacheKey, expiresAt, this.StatusCode(), this.calculateHeaderLength(), size, cacheRef.MaxSizeBytes(), this.isPartial)
if err != nil { if err != nil {
if err == caches.ErrEntityTooLarge && addStatusHeader { if err == caches.ErrEntityTooLarge && addStatusHeader {
this.Header().Set("X-Cache", "BYPASS, entity too large") this.Header().Set("X-Cache", "BYPASS, entity too large")
@@ -638,7 +638,7 @@ func (this *HTTPWriter) PrepareCompression(resp *http.Response, size int64) {
cacheKey += this.cacheReaderSuffix cacheKey += this.cacheReaderSuffix
} }
compressionCacheWriter, err := this.cacheStorage.OpenWriter(cacheKey+caches.SuffixCompression+compressionEncoding, expiredAt, this.StatusCode(), -1, cacheRef.MaxSizeBytes(), false) compressionCacheWriter, err := this.cacheStorage.OpenWriter(cacheKey+caches.SuffixCompression+compressionEncoding, expiredAt, this.StatusCode(), this.calculateHeaderLength(), -1, cacheRef.MaxSizeBytes(), false)
if err != nil { if err != nil {
return return
} }
@@ -964,7 +964,7 @@ func (this *HTTPWriter) finishWebP() {
expiredAt = this.cacheWriter.ExpiredAt() expiredAt = this.cacheWriter.ExpiredAt()
} }
webpCacheWriter, _ = this.cacheStorage.OpenWriter(cacheKey, expiredAt, this.StatusCode(), -1, -1, false) webpCacheWriter, _ = this.cacheStorage.OpenWriter(cacheKey, expiredAt, this.StatusCode(), -1, -1, -1, false)
if webpCacheWriter != nil { if webpCacheWriter != nil {
// 写入Header // 写入Header
for k, v := range this.Header() { for k, v := range this.Header() {
@@ -1179,3 +1179,13 @@ func (this *HTTPWriter) finishRequest() {
_ = this.rawReader.Close() _ = this.rawReader.Close()
} }
} }
// 计算Header长度
func (this *HTTPWriter) calculateHeaderLength() (result int) {
for k, v := range this.Header() {
for _, v1 := range v {
result += len(k) + 1 /**:**/ + len(v1) + 1 /**\n**/
}
}
return
}