ChenKaiLiuG/GoEdgeNode
1
0
Fork 0
Code Issues Actions Packages Projects Releases Wiki Activity

使用新的方法控制缓存并发写入速度

Browse Source
This commit is contained in:
刘祥超
2023-07-29 09:29:36 +08:00
parent 3793d684fa
commit 33bb06fbc3
12 changed files with 116 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
internal/caches/errors.go
View File

@@ -12,7 +12,7 @@ var (
ErrEntityTooLarge = errors.New("entity too large")
ErrWritingUnavailable = errors.New("writing unavailable")
ErrWritingQueueFull = errors.New("writing queue full")
ErrTooManyOpenFiles = errors.New("too many open files")
ErrServerIsBusy = errors.New("server is busy")
)
// CapacityError 容量错误
@@ -38,7 +38,7 @@ func CanIgnoreErr(err error) bool {
err == ErrEntityTooLarge ||
err == ErrWritingUnavailable ||
err == ErrWritingQueueFull ||
err == ErrTooManyOpenFiles {
err == ErrServerIsBusy {
return true
}
_, ok := err.(*CapacityError)

52
internal/caches/max_open_files.go
View File

@@ -1,52 +0,0 @@
// Copyright 2022 Liuxiangchao iwind.liu@gmail.com. All rights reserved. Official site: https://goedge.cn .
package caches
import (
"github.com/TeaOSLab/EdgeNode/internal/goman"
"sync/atomic"
"time"
)
const (
modeSlow int32 = 1
modeFast int32 = 2
)
// MaxOpenFiles max open files manager
type MaxOpenFiles struct {
ticker *time.Ticker
mode int32
}
func NewMaxOpenFiles() *MaxOpenFiles {
var f = &MaxOpenFiles{}
f.ticker = time.NewTicker(1 * time.Second)
f.init()
return f
}
func (this *MaxOpenFiles) init() {
goman.New(func() {
for range this.ticker.C {
// reset mode
atomic.StoreInt32(&this.mode, modeFast)
}
})
}
func (this *MaxOpenFiles) Fast() {
atomic.AddInt32(&this.mode, modeFast)
}
func (this *MaxOpenFiles) FinishAll() {
this.Fast()
}
func (this *MaxOpenFiles) Slow() {
atomic.StoreInt32(&this.mode, modeSlow)
}
func (this *MaxOpenFiles) Next() bool {
return atomic.LoadInt32(&this.mode) != modeSlow
}

35
internal/caches/max_open_files_test.go
View File

@@ -1,35 +0,0 @@
// Copyright 2022 Liuxiangchao iwind.liu@gmail.com. All rights reserved. Official site: https://goedge.cn .
package caches_test
import (
"github.com/TeaOSLab/EdgeNode/internal/caches"
"testing"
"time"
)
func TestNewMaxOpenFiles(t *testing.T) {
var maxOpenFiles = caches.NewMaxOpenFiles()
maxOpenFiles.Fast()
t.Log("fast:", maxOpenFiles.Next())
maxOpenFiles.Slow()
t.Log("slow:", maxOpenFiles.Next())
time.Sleep(1*time.Second + 1*time.Millisecond)
t.Log("slow 1 second:", maxOpenFiles.Next())
maxOpenFiles.Slow()
t.Log("slow:", maxOpenFiles.Next())
maxOpenFiles.Slow()
t.Log("slow:", maxOpenFiles.Next())
time.Sleep(1 * time.Second)
t.Log("slow 1 second:", maxOpenFiles.Next())
maxOpenFiles.Slow()
t.Log("slow:", maxOpenFiles.Next())
maxOpenFiles.Fast()
t.Log("fast:", maxOpenFiles.Next())
}

24
internal/caches/storage_file.go
View File

@@ -61,9 +61,6 @@ const (
var sharedWritingFileKeyMap = map[string]zero.Zero{} // key => bool
var sharedWritingFileKeyLocker = sync.Mutex{}
var maxOpenFiles = NewMaxOpenFiles()
const maxOpenFilesSlowCost = 1000 * time.Microsecond // us
const protectingLoadWhenDump = false
// FileStorage 文件缓存
@@ -445,9 +442,9 @@ func (this *FileStorage) openWriter(key string, expiredAt int64, status int, hea
return nil, ErrFileIsWriting
}
if !isFlushing && !maxOpenFiles.Next() {
if !isFlushing && !fsutils.WriteReady() {
sharedWritingFileKeyLocker.Unlock()
return nil, ErrTooManyOpenFiles
return nil, ErrServerIsBusy
}
sharedWritingFileKeyMap[key] = zero.New()
@@ -456,9 +453,6 @@ func (this *FileStorage) openWriter(key string, expiredAt int64, status int, hea
if !isOk {
sharedWritingFileKeyLocker.Lock()
delete(sharedWritingFileKeyMap, key)
if len(sharedWritingFileKeyMap) == 0 {
maxOpenFiles.FinishAll()
}
sharedWritingFileKeyLocker.Unlock()
}
}()
@@ -558,7 +552,6 @@ func (this *FileStorage) openWriter(key string, expiredAt int64, status int, hea
if isNewCreated && existsFile {
flags |= os.O_TRUNC
}
var before = time.Now()
writer, err := os.OpenFile(tmpPath, flags, 0666)
if err != nil {
// TODO 检查在各个系统中的稳定性
@@ -572,13 +565,6 @@ func (this *FileStorage) openWriter(key string, expiredAt int64, status int, hea
return nil, err
}
}
if !isFlushing {
if time.Since(before) >= maxOpenFilesSlowCost {
maxOpenFiles.Slow()
} else {
maxOpenFiles.Fast()
}
}
var removeOnFailure = true
defer func() {
@@ -639,18 +625,12 @@ func (this *FileStorage) openWriter(key string, expiredAt int64, status int, hea
return NewPartialFileWriter(writer, key, expiredAt, metaHeaderSize, metaBodySize, isNewCreated, isPartial, partialBodyOffset, partialRanges, func() {
sharedWritingFileKeyLocker.Lock()
delete(sharedWritingFileKeyMap, key)
if len(sharedWritingFileKeyMap) == 0 {
maxOpenFiles.FinishAll()
}
sharedWritingFileKeyLocker.Unlock()
}), nil
} else {
return NewFileWriter(this, writer, key, expiredAt, metaHeaderSize, metaBodySize, maxSize, func() {
sharedWritingFileKeyLocker.Lock()
delete(sharedWritingFileKeyMap, key)
if len(sharedWritingFileKeyMap) == 0 {
maxOpenFiles.FinishAll()
}
sharedWritingFileKeyLocker.Unlock()
}), nil
}

4
internal/caches/storage_memory.go
View File

@@ -488,10 +488,8 @@ func (this *MemoryStorage) startFlush() {
if err == nil && loadStat != nil {
if loadStat.Load1 > 10 {
writeDelayMS = 100
} else if loadStat.Load1 > 3 {
} else if loadStat.Load1 > 5 {
writeDelayMS = 50
} else if loadStat.Load1 > 2 {
writeDelayMS = 10
} else {
writeDelayMS = 0
}

5
internal/caches/writer_file.go
View File

@@ -3,6 +3,7 @@ package caches
import (
"encoding/binary"
"errors"
fsutils "github.com/TeaOSLab/EdgeNode/internal/utils/fs"
"github.com/iwind/TeaGo/types"
"io"
"os"
@@ -42,7 +43,9 @@ func NewFileWriter(storage StorageInterface, rawWriter *os.File, key string, exp
// WriteHeader 写入数据
func (this *FileWriter) WriteHeader(data []byte) (n int, err error) {
fsutils.WriteBegin()
n, err = this.rawWriter.Write(data)
fsutils.WriteEnd()
this.headerSize += int64(n)
if err != nil {
_ = this.Discard()
@@ -72,7 +75,9 @@ func (this *FileWriter) WriteHeaderLength(headerLength int) error {
// Write 写入数据
func (this *FileWriter) Write(data []byte) (n int, err error) {
fsutils.WriteBegin()
n, err = this.rawWriter.Write(data)
fsutils.WriteEnd()
this.bodySize += int64(n)
if this.maxSize > 0 && this.bodySize > this.maxSize {

9
internal/caches/writer_partial_file.go
View File

@@ -4,6 +4,7 @@ package caches
import (
"encoding/binary"
fsutils "github.com/TeaOSLab/EdgeNode/internal/utils/fs"
"github.com/iwind/TeaGo/types"
"io"
"os"
@@ -53,7 +54,9 @@ func (this *PartialFileWriter) WriteHeader(data []byte) (n int, err error) {
if !this.isNew {
return
}
fsutils.WriteBegin()
n, err = this.rawWriter.Write(data)
fsutils.WriteEnd()
this.headerSize += int64(n)
if err != nil {
_ = this.Discard()
@@ -62,7 +65,9 @@ func (this *PartialFileWriter) WriteHeader(data []byte) (n int, err error) {
}
func (this *PartialFileWriter) AppendHeader(data []byte) error {
fsutils.WriteBegin()
_, err := this.rawWriter.Write(data)
fsutils.WriteEnd()
if err != nil {
_ = this.Discard()
} else {
@@ -99,7 +104,9 @@ func (this *PartialFileWriter) WriteHeaderLength(headerLength int) error {
// Write 写入数据
func (this *PartialFileWriter) Write(data []byte) (n int, err error) {
fsutils.WriteBegin()
n, err = this.rawWriter.Write(data)
fsutils.WriteEnd()
this.bodySize += int64(n)
if err != nil {
_ = this.Discard()
@@ -128,7 +135,9 @@ func (this *PartialFileWriter) WriteAt(offset int64, data []byte) error {
this.bodyOffset = SizeMeta + int64(keyLength) + this.headerSize
}
fsutils.WriteBegin()
_, err := this.rawWriter.WriteAt(data, this.bodyOffset+offset)
fsutils.WriteEnd()
if err != nil {
return err
}