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

优化Partial Content兼容性

Browse Source
This commit is contained in:
GoEdgeLab
2022-11-20 18:07:46 +08:00
parent ee926e4dda
commit 3a276a23ef
8 changed files with 143 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
internal/caches/list_file_db.go
View File

@@ -14,6 +14,7 @@ import (
"github.com/iwind/TeaGo/types"
timeutil "github.com/iwind/TeaGo/utils/time"
"os"
"path/filepath"
"runtime"
"strings"
"time"
@@ -108,7 +109,7 @@ func (this *FileListDB) Open(dbPath string) error {
this.writeBatch = dbs.NewBatch(writeDB, 4)
this.writeBatch.OnFail(func(err error) {
remotelogs.Warn("LIST_FILE_DB", "run batch failed: "+err.Error())
remotelogs.Warn("LIST_FILE_DB", "run batch failed: "+err.Error()+" ("+filepath.Base(this.dbPath)+")")
})
goman.New(func() {

34
internal/caches/partial_ranges.go
View File

@@ -7,19 +7,21 @@ import (
"encoding/json"
"github.com/iwind/TeaGo/types"
"os"
"strconv"
)
// PartialRanges 内容分区范围定义
type PartialRanges struct {
ExpiresAt int64 `json:"expiresAt"` // 过期时间
Ranges [][2]int64 `json:"ranges"`
Version int `json:"version"` // 版本号
Ranges [][2]int64 `json:"ranges"` // 范围
BodySize int64 `json:"bodySize"` // 总长度
}
// NewPartialRanges 获取新对象
func NewPartialRanges(expiresAt int64) *PartialRanges {
return &PartialRanges{
Ranges: [][2]int64{},
ExpiresAt: expiresAt,
Version: 1,
}
}
@@ -35,9 +37,11 @@ func NewPartialRangesFromData(data []byte) (*PartialRanges, error) {
var colonIndex = bytes.IndexRune(line, ':')
if colonIndex > 0 {
switch string(line[:colonIndex]) {
case "e":
rs.ExpiresAt = types.Int64(line[colonIndex+1:])
case "r":
case "v": // 版本号
rs.Version = types.Int(line[colonIndex+1:])
case "b": // 总长度
rs.BodySize = types.Int64(line[colonIndex+1:])
case "r": // 范围信息
var commaIndex = bytes.IndexRune(line, ',')
if commaIndex > 0 {
rs.Ranges = append(rs.Ranges, [2]int64{types.Int64(line[colonIndex+1 : commaIndex]), types.Int64(line[commaIndex+1:])})
@@ -53,16 +57,18 @@ func NewPartialRangesFromData(data []byte) (*PartialRanges, error) {
}
// NewPartialRangesFromJSON 从JSON中解析范围
func newPartialRangesFromJSON(data []byte) (*PartialRanges, error) {
func NewPartialRangesFromJSON(data []byte) (*PartialRanges, error) {
var rs = NewPartialRanges(0)
err := json.Unmarshal(data, &rs)
if err != nil {
return nil, err
}
rs.Version = 0
return rs, nil
}
// NewPartialRangesFromFile 从文件中加载范围信息
func NewPartialRangesFromFile(path string) (*PartialRanges, error) {
data, err := os.ReadFile(path)
if err != nil {
@@ -74,7 +80,7 @@ func NewPartialRangesFromFile(path string) (*PartialRanges, error) {
// 兼容老的JSON格式
if data[0] == '{' {
return newPartialRangesFromJSON(data)
return NewPartialRangesFromJSON(data)
}
// 新的格式
@@ -151,13 +157,15 @@ func (this *PartialRanges) Nearest(begin int64, end int64) (r [2]int64, ok bool)
// 转换为字符串
func (this *PartialRanges) String() string {
var s = "e:" + types.String(this.ExpiresAt) + "\n"
var s = "v:" + strconv.Itoa(this.Version) + "\n" + // version
"b:" + this.formatInt64(this.BodySize) + "\n" // bodySize
for _, r := range this.Ranges {
s += "r:" + types.String(r[0]) + "," + types.String(r[1]) + "\n"
s += "r:" + this.formatInt64(r[0]) + "," + this.formatInt64(r[1]) + "\n" // range
}
return s
}
// Bytes 将内容转换为字节
func (this *PartialRanges) Bytes() []byte {
return []byte(this.String())
}
@@ -167,6 +175,7 @@ func (this *PartialRanges) WriteToFile(path string) error {
return os.WriteFile(path, this.Bytes(), 0666)
}
// Max 获取最大位置
func (this *PartialRanges) Max() int64 {
if len(this.Ranges) > 0 {
return this.Ranges[len(this.Ranges)-1][1]
@@ -174,6 +183,7 @@ func (this *PartialRanges) Max() int64 {
return 0
}
// Reset 重置范围信息
func (this *PartialRanges) Reset() {
this.Ranges = [][2]int64{}
}
@@ -230,3 +240,7 @@ func (this *PartialRanges) max(n1 int64, n2 int64) int64 {
}
return n2
}
func (this *PartialRanges) formatInt64(i int64) string {
return strconv.FormatInt(i, 10)
}

46
internal/caches/partial_ranges_test.go
View File

@@ -149,7 +149,6 @@ func TestNewPartialRanges_Large_Range(t *testing.T) {
func TestPartialRanges_Encode_JSON(t *testing.T) {
var r = caches.NewPartialRanges(0)
r.ExpiresAt = time.Now().Unix()
for i := 0; i < 10; i++ {
r.Ranges = append(r.Ranges, [2]int64{int64(i * 100), int64(i*100 + 100)})
}
@@ -164,7 +163,7 @@ func TestPartialRanges_Encode_JSON(t *testing.T) {
func TestPartialRanges_Encode_String(t *testing.T) {
var r = caches.NewPartialRanges(0)
r.ExpiresAt = time.Now().Unix()
r.BodySize = 1024
for i := 0; i < 10; i++ {
r.Ranges = append(r.Ranges, [2]int64{int64(i * 100), int64(i*100 + 100)})
}
@@ -180,6 +179,36 @@ func TestPartialRanges_Encode_String(t *testing.T) {
logs.PrintAsJSON(r2, t)
}
func TestPartialRanges_Version(t *testing.T) {
{
ranges, err := caches.NewPartialRangesFromData([]byte(`e:1668928495
r:0,1048576
r:1140260864,1140295164`))
if err != nil {
t.Fatal(err)
}
t.Log("version:", ranges.Version)
}
{
ranges, err := caches.NewPartialRangesFromData([]byte(`e:1668928495
r:0,1048576
r:1140260864,1140295164
v:0
`))
if err != nil {
t.Fatal(err)
}
t.Log("version:", ranges.Version)
}
{
ranges, err := caches.NewPartialRangesFromJSON([]byte(`{}`))
if err != nil {
t.Fatal(err)
}
t.Log("version:", ranges.Version)
}
}
func BenchmarkNewPartialRanges(b *testing.B) {
for i := 0; i < b.N; i++ {
var r = caches.NewPartialRanges(0)
@@ -188,3 +217,16 @@ func BenchmarkNewPartialRanges(b *testing.B) {
}
}
}
func BenchmarkPartialRanges_String(b *testing.B) {
var r = caches.NewPartialRanges(0)
r.BodySize = 1024
for i := 0; i < 10; i++ {
r.Ranges = append(r.Ranges, [2]int64{int64(i * 100), int64(i*100 + 100)})
}
b.ResetTimer()
for i := 0; i < b.N; i++ {
_ = r.String()
}
}

14
internal/caches/storage_file.go
View File

@@ -556,13 +556,23 @@ func (this *FileStorage) openWriter(key string, expiredAt int64, status int, hea
_ = partialReader.Close()
if err == nil && partialReader.bodyOffset > 0 {
partialRanges = partialReader.Ranges()
if bodySize > 0 && partialRanges != nil && partialRanges.BodySize > 0 && bodySize != partialRanges.BodySize {
_ = this.removeCacheFile(tmpPath)
} else {
isNewCreated = false
partialBodyOffset = partialReader.bodyOffset
}
} else {
_ = this.removeCacheFile(tmpPath)
}
}
}
if isNewCreated {
err = this.list.Remove(hash)
if err != nil {
return nil, err
}
}
if partialRanges == nil {
partialRanges = NewPartialRanges(expiredAt)
}
@@ -617,7 +627,7 @@ func (this *FileStorage) openWriter(key string, expiredAt int64, status int, hea
}
var metaBodySize int64 = -1
var metaHeaderSize int = -1
var metaHeaderSize = -1
if isNewCreated {
// 写入meta
// 从v0.5.8开始不再在meta中写入Key
@@ -650,7 +660,7 @@ func (this *FileStorage) openWriter(key string, expiredAt int64, status int, hea
isOk = true
if isPartial {
return NewPartialFileWriter(writer, key, expiredAt, isNewCreated, isPartial, partialBodyOffset, partialRanges, func() {
return NewPartialFileWriter(writer, key, expiredAt, metaHeaderSize, metaBodySize, isNewCreated, isPartial, partialBodyOffset, partialRanges, func() {
sharedWritingFileKeyLocker.Lock()
delete(sharedWritingFileKeyMap, key)
if len(sharedWritingFileKeyMap) == 0 {

28
internal/caches/writer_partial_file.go
View File

@@ -13,8 +13,13 @@ import (
type PartialFileWriter struct {
rawWriter *os.File
key string
metaHeaderSize int
headerSize int64
metaBodySize int64
bodySize int64
expiredAt int64
endFunc func()
once sync.Once
@@ -27,7 +32,7 @@ type PartialFileWriter struct {
rangePath string
}
func NewPartialFileWriter(rawWriter *os.File, key string, expiredAt int64, isNew bool, isPartial bool, bodyOffset int64, ranges *PartialRanges, endFunc func()) *PartialFileWriter {
func NewPartialFileWriter(rawWriter *os.File, key string, expiredAt int64, metaHeaderSize int, metaBodySize int64, isNew bool, isPartial bool, bodyOffset int64, ranges *PartialRanges, endFunc func()) *PartialFileWriter {
return &PartialFileWriter{
key: key,
rawWriter: rawWriter,
@@ -38,6 +43,8 @@ func NewPartialFileWriter(rawWriter *os.File, key string, expiredAt int64, isNew
bodyOffset: bodyOffset,
ranges: ranges,
rangePath: partialRangesFilePath(rawWriter.Name()),
metaHeaderSize: metaHeaderSize,
metaBodySize: metaBodySize,
}
}
@@ -71,7 +78,11 @@ func (this *PartialFileWriter) AppendHeader(data []byte) error {
// WriteHeaderLength 写入Header长度数据
func (this *PartialFileWriter) 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))
_, err := this.rawWriter.Seek(SizeExpiresAt+SizeStatus+SizeURLLength, io.SeekStart)
if err != nil {
@@ -110,8 +121,13 @@ func (this *PartialFileWriter) WriteAt(offset int64, data []byte) error {
}
if this.bodyOffset == 0 {
this.bodyOffset = SizeMeta + int64(len(this.key)) + this.headerSize
var keyLength = 0
if this.ranges.Version == 0 { // 以往的版本包含有Key
keyLength = len(this.key)
}
this.bodyOffset = SizeMeta + int64(keyLength) + this.headerSize
}
_, err := this.rawWriter.WriteAt(data, this.bodyOffset+offset)
if err != nil {
return err
@@ -129,7 +145,10 @@ func (this *PartialFileWriter) SetBodyLength(bodyLength int64) {
// WriteBodyLength 写入Body长度数据
func (this *PartialFileWriter) WriteBodyLength(bodyLength int64) error {
bytes8 := make([]byte, 8)
if this.metaBodySize > 0 && this.metaBodySize == bodyLength {
return nil
}
var bytes8 = make([]byte, 8)
binary.BigEndian.PutUint64(bytes8, uint64(bodyLength))
_, err := this.rawWriter.Seek(SizeExpiresAt+SizeStatus+SizeURLLength+SizeHeaderLength, io.SeekStart)
if err != nil {
@@ -150,6 +169,7 @@ func (this *PartialFileWriter) Close() error {
this.endFunc()
})
this.ranges.BodySize = this.bodySize
err := this.ranges.WriteToFile(this.rangePath)
if err != nil {
_ = this.rawWriter.Close()

2
internal/caches/writer_partial_file_test.go
View File

@@ -27,7 +27,7 @@ func TestPartialFileWriter_Write(t *testing.T) {
t.Fatal(err)
}
var ranges = caches.NewPartialRanges(0)
var writer = caches.NewPartialFileWriter(fp, "test", time.Now().Unix()+86500, true, true, 0, ranges, func() {
var writer = caches.NewPartialFileWriter(fp, "test", time.Now().Unix()+86500, -1, -1, true, true, 0, ranges, func() {
t.Log("end")
})
_, err = writer.WriteHeader([]byte("header"))

8
internal/nodes/http_request_cache.go
View File

@@ -628,14 +628,14 @@ func (this *HTTPRequest) tryPartialReader(storage caches.StorageInterface, key s
}()
// 检查范围
const maxFirstSpan = 16 << 20 // TODO 可以在缓存策略中设置此值
//const maxFirstSpan = 16 << 20 // TODO 可以在缓存策略中设置此值
for index, r := range ranges {
// 没有指定结束字节时，自动指定一个
if r.Start() >= 0 && r.End() == -1 {
// 没有指定结束位置时，自动指定一个
/**if r.Start() >= 0 && r.End() == -1 {
if partialReader.MaxLength() > r.Start()+maxFirstSpan {
r[1] = r.Start() + maxFirstSpan
}
}
}**/
r1, ok := r.Convert(partialReader.MaxLength())
if !ok {
return nil, nil

14
internal/nodes/http_writer.go
View File

@@ -303,7 +303,19 @@ func (this *HTTPWriter) PrepareCache(resp *http.Response, size int64) {
if this.isPartial {
cacheKey += caches.SuffixPartial
}
cacheWriter, err := storage.OpenWriter(cacheKey, expiresAt, this.StatusCode(), this.calculateHeaderLength(), size, cacheRef.MaxSizeBytes(), this.isPartial)
// 待写入尺寸
var totalSize = size
if totalSize < 0 && this.isPartial {
var contentRange = resp.Header.Get("Content-Range")
if len(contentRange) > 0 {
_, partialTotalSize := httpRequestParseContentRangeHeader(contentRange)
if partialTotalSize > 0 {
totalSize = partialTotalSize
}
}
}
cacheWriter, err := storage.OpenWriter(cacheKey, expiresAt, this.StatusCode(), this.calculateHeaderLength(), totalSize, cacheRef.MaxSizeBytes(), this.isPartial)
if err != nil {
if err == caches.ErrEntityTooLarge && addStatusHeader {
this.Header().Set("X-Cache", "BYPASS, entity too large")