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

读取文件时增加线程数限制

Browse Source
This commit is contained in:
刘祥超
2024-04-29 22:01:55 +08:00
parent 1cede74db3
commit 7febc6aaf3
5 changed files with 133 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
internal/caches/errors.go
View File

@@ -55,3 +55,7 @@ func IsCapacityError(err error) bool {
var capacityErr *CapacityError
return errors.As(err, &capacityErr)
}
func IsBusyError(err error) bool {
return err != nil && errors.Is(err, ErrServerIsBusy)
}

13
internal/caches/reader_file.go
View File

@@ -3,6 +3,7 @@ package caches
import (
"encoding/binary"
"errors"
fsutils "github.com/TeaOSLab/EdgeNode/internal/utils/fs"
rangeutils "github.com/TeaOSLab/EdgeNode/internal/utils/ranges"
"github.com/iwind/TeaGo/types"
"io"
@@ -174,7 +175,9 @@ func (this *FileReader) ReadHeader(buf []byte, callback ReaderFunc) error {
var headerSize = this.headerSize
for {
fsutils.ReaderLimiter.Ack()
n, err := this.fp.Read(buf)
fsutils.ReaderLimiter.Release()
if n > 0 {
if n < headerSize {
goNext, e := callback(n)
@@ -236,7 +239,9 @@ func (this *FileReader) ReadBody(buf []byte, callback ReaderFunc) error {
}
for {
fsutils.ReaderLimiter.Ack()
n, err := this.fp.Read(buf)
fsutils.ReaderLimiter.Release()
if n > 0 {
goNext, e := callback(n)
if e != nil {
@@ -267,7 +272,9 @@ func (this *FileReader) Read(buf []byte) (n int, err error) {
return
}
fsutils.ReaderLimiter.Ack()
n, err = this.fp.Read(buf)
fsutils.ReaderLimiter.Release()
if err != nil && err != io.EOF {
_ = this.discard()
}
@@ -299,13 +306,17 @@ func (this *FileReader) ReadBodyRange(buf []byte, start int64, end int64, callba
isOk = true
return ErrInvalidRange
}
fsutils.ReaderLimiter.Ack()
_, err := this.fp.Seek(offset, io.SeekStart)
fsutils.ReaderLimiter.Release()
if err != nil {
return err
}
for {
fsutils.ReaderLimiter.Ack()
n, err := this.fp.Read(buf)
fsutils.ReaderLimiter.Release()
if n > 0 {
var n2 = int(end-offset) + 1
if n2 <= n {
@@ -375,7 +386,9 @@ func (this *FileReader) Close() error {
}
func (this *FileReader) readToBuff(fp *os.File, buf []byte) (ok bool, err error) {
fsutils.ReaderLimiter.Ack()
n, err := fp.Read(buf)
fsutils.ReaderLimiter.Release()
if err != nil {
return false, err
}

8
internal/caches/storage_file.go
View File

@@ -393,7 +393,11 @@ func (this *FileStorage) openReader(key string, allowMemory bool, useStale bool,
// 尝试通过MMAP读取
if estimatedSize > 0 {
if !fsutils.ReaderLimiter.TryAck() {
return nil, ErrServerIsBusy
}
reader, err := this.tryMMAPReader(isPartial, estimatedSize, path)
fsutils.ReaderLimiter.Release()
if err != nil {
return nil, err
}
@@ -412,7 +416,11 @@ func (this *FileStorage) openReader(key string, allowMemory bool, useStale bool,
var err error
if openFile == nil {
if !fsutils.ReaderLimiter.TryAck() {
return nil, ErrServerIsBusy
}
fp, err = os.OpenFile(path, os.O_RDONLY, 0444)
fsutils.ReaderLimiter.Release()
if err != nil {
if !os.IsNotExist(err) {
return nil, err