Skip to content

Commit b6db83e

Browse files
j2rong4cnSuyunmeng
j2rong4cn
and
Suyunmeng
authored
refactor: unify stream caching via HybridCache (#2460)
* add LinearMemory * replace mmap with LinearMemory * remove unused code * add GuardedMemory; add `min_free_memoryMB` conf * add HybridCache and StreamBuffer * log * rename SizedReadWriterAt to Section * 重构 FileStream，改用 HybridCache * 重构 HybridCache，更新方法名并添加回滚功能；优化请求和流处理逻辑 * 重构 StreamSectionReader 接口，使用HybridCache * 在 NewGuardedMemory 函数中添加了对 LinearMemory 的最终化处理，以确保内存释放 * . * 优化检查逻辑 * 重命名 * 改进、重命名 * 修复 * 添加测试 * 移除HybridCacheReader并引入DynamicReadAtSeeker * 重构缓存读取逻辑，简化代码并引入ReadFromN方法 * 优化缓存配置注释并修复下载器部分大小限制逻辑 * 优化中断逻辑 * 优化下载器代码 * 修复bug * HybridCache添加多文件缓存模式 * 优化下载器并发，添加测试 * 修复bug * 重命名+注释 * . * fix(net): always cleanup downloader on interrupt * fix(net): guard chunk enqueue with context cancel * fix(net): update interrupt logic and add download interrupt test * fix(test): update concurrency limit in high concurrency test * refactor(buffer): simplify ReadAt logic * refactor(config): update memory configuration logic * . --------- Co-authored-by: Suyunmeng <[email protected]>
1 parent 0726d16 commit b6db83e

24 files changed

Lines changed: 1767 additions & 651 deletions

File tree

‎drivers/teldrive/types.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,7 @@ type chunkTask struct {
5252
fileName string
5353
chunkSize int64
5454
reader io.ReadSeeker
55-
ss stream.StreamSectionReaderIF
55+
ss stream.StreamSectionReader
5656
}
5757

5858
type CopyManager struct {

‎internal/bootstrap/config.go‎

Lines changed: 34 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package bootstrap
22

33
import (
4+
"math"
45
"net/url"
56
"os"
67
"path/filepath"
@@ -96,27 +97,46 @@ func InitConfig() {
9697
confFromEnv()
9798
}
9899

99-
if conf.Conf.MaxConcurrency > 0 {
100-
net.DefaultConcurrencyLimit = &net.ConcurrencyLimit{Limit: conf.Conf.MaxConcurrency}
100+
if conf.Conf.MaxConcurrency > math.MaxInt32 {
101+
net.DefaultConcurrencyLimit = &net.ConcurrencyLimit{Limit: math.MaxInt32}
102+
} else if conf.Conf.MaxConcurrency > 0 {
103+
net.DefaultConcurrencyLimit = &net.ConcurrencyLimit{Limit: uint32(conf.Conf.MaxConcurrency)}
101104
}
102-
if conf.Conf.MaxBufferLimit < 0 {
103-
m, _ := mem.VirtualMemory()
104-
if m != nil {
105-
conf.MaxBufferLimit = max(int(float64(m.Total)*0.05), 4*utils.MB)
106-
conf.MaxBufferLimit -= conf.MaxBufferLimit % utils.MB
105+
106+
memStat, _ := mem.VirtualMemory()
107+
if memStat != nil {
108+
log.Infof("total memory: %dMB, available: %dMB", memStat.Total>>20, memStat.Available>>20)
109+
if conf.Conf.MinFreeMemory < 0 {
110+
conf.MinFreeMemory = 0
111+
log.Info("disable memory cache")
107112
} else {
108-
conf.MaxBufferLimit = 16 * utils.MB
113+
if conf.Conf.MinFreeMemory < 16 {
114+
t := (memStat.Total >> 20) / 10
115+
conf.MinFreeMemory = max(16, min(t, 1024)) << 20
116+
} else {
117+
conf.MinFreeMemory = uint64(conf.Conf.MinFreeMemory) << 20
118+
}
119+
log.Infof("min free memory: %dMB", conf.MinFreeMemory>>20)
109120
}
121+
122+
if conf.Conf.MaxBlockLimit < 4 {
123+
t := (memStat.Total >> 20) * 3 / 100
124+
conf.MaxBlockLimit = max(4, min(uint64(t), 64)) << 20
125+
} else {
126+
conf.MaxBlockLimit = uint64(conf.Conf.MaxBlockLimit) << 20
127+
}
128+
log.Infof("max block limit: %dMB", conf.MaxBlockLimit>>20)
110129
} else {
111-
conf.MaxBufferLimit = conf.Conf.MaxBufferLimit * utils.MB
130+
conf.MinFreeMemory = 0
131+
log.Warn("failed to get memory info, disable memory cache")
112132
}
113-
log.Infof("max buffer limit: %dMB", conf.MaxBufferLimit/utils.MB)
114-
if conf.Conf.MmapThreshold > 0 {
115-
conf.MmapThreshold = conf.Conf.MmapThreshold * utils.MB
133+
134+
if conf.Conf.CacheThreshold > 0 {
135+
conf.CacheThreshold = uint64(conf.Conf.CacheThreshold) << 20
116136
} else {
117-
conf.MmapThreshold = 0
137+
conf.CacheThreshold = 0
118138
}
119-
log.Infof("mmap threshold: %dMB", conf.Conf.MmapThreshold)
139+
log.Infof("cache threshold: %dMB", conf.CacheThreshold>>20)
120140

121141
if len(conf.Conf.Log.Filter.Filters) == 0 {
122142
conf.Conf.Log.Filter.Enable = false

‎internal/cache/file.go‎

Lines changed: 186 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,186 @@
1+
package cache
2+
3+
import (
4+
"errors"
5+
"io"
6+
"os"
7+
8+
"github.com/OpenListTeam/OpenList/v4/internal/conf"
9+
"github.com/OpenListTeam/OpenList/v4/pkg/buffer"
10+
)
11+
12+
type FileCache interface {
13+
buffer.Block
14+
io.Closer
15+
Truncate(size int64) error
16+
}
17+
18+
type singleFileCache struct {
19+
*os.File
20+
size int64
21+
}
22+
23+
func (s *singleFileCache) Size() int64 {
24+
return s.size
25+
}
26+
27+
func (s *singleFileCache) Truncate(size int64) error {
28+
if size <= s.size {
29+
return nil
30+
}
31+
err := s.File.Truncate(size)
32+
if err == nil {
33+
s.size = size
34+
}
35+
return err
36+
}
37+
38+
func (s *singleFileCache) Close() error {
39+
err := s.File.Close()
40+
_ = os.Remove(s.File.Name())
41+
return err
42+
}
43+
44+
type fileBlock struct {
45+
file *os.File
46+
size int64
47+
written int64
48+
}
49+
50+
type MultiFileCache struct {
51+
blocks []*fileBlock
52+
size int64
53+
}
54+
55+
func (s *MultiFileCache) Size() int64 {
56+
return s.size
57+
}
58+
59+
func (m *MultiFileCache) Close() error {
60+
var errs []error
61+
for _, c := range m.blocks {
62+
if err := c.file.Close(); err != nil {
63+
errs = append(errs, err)
64+
}
65+
_ = os.Remove(c.file.Name())
66+
}
67+
clear(m.blocks)
68+
m.blocks = m.blocks[:0]
69+
return errors.Join(errs...)
70+
}
71+
72+
func (m *MultiFileCache) Truncate(size int64) error {
73+
if size <= m.size {
74+
return nil
75+
}
76+
f, err := os.CreateTemp(conf.Conf.TempDir, "file-*")
77+
if err != nil {
78+
return err
79+
}
80+
m.blocks = append(m.blocks, &fileBlock{file: f, size: size - m.size})
81+
m.size = size
82+
return nil
83+
}
84+
85+
func (m *MultiFileCache) ReadAt(p []byte, off int64) (n int, err error) {
86+
if len(p) == 0 {
87+
return 0, nil
88+
}
89+
if off < 0 || off >= m.size {
90+
return 0, io.EOF
91+
}
92+
93+
for _, c := range m.blocks {
94+
if off >= c.size {
95+
off -= c.size
96+
continue
97+
}
98+
99+
canRead := min(len(p)-n, int(c.size-off))
100+
if canRead <= 0 {
101+
break
102+
}
103+
104+
filled := 0
105+
106+
if off < c.written {
107+
fileReadable := min(canRead, int(c.written-off))
108+
nn, fileErr := c.file.ReadAt(p[n:n+fileReadable], off)
109+
n += nn
110+
filled = nn
111+
if fileErr != nil && !errors.Is(fileErr, io.EOF) {
112+
return n, fileErr
113+
}
114+
}
115+
116+
if n == len(p) {
117+
return n, nil
118+
}
119+
120+
if zeroFill := canRead - filled; zeroFill > 0 {
121+
clear(p[n : n+zeroFill])
122+
n += zeroFill
123+
}
124+
125+
if n == len(p) {
126+
return n, nil
127+
}
128+
off = 0
129+
}
130+
131+
return n, io.EOF
132+
}
133+
134+
func (m *MultiFileCache) WriteAt(p []byte, off int64) (n int, err error) {
135+
if len(p) == 0 {
136+
return 0, nil
137+
}
138+
if off < 0 || off >= m.size {
139+
return 0, io.ErrShortWrite
140+
}
141+
142+
for _, b := range m.blocks {
143+
if off >= b.size {
144+
off -= b.size
145+
continue
146+
}
147+
148+
canWrite := min(len(p)-n, int(b.size-off))
149+
if canWrite <= 0 {
150+
break
151+
}
152+
153+
nn, fileErr := b.file.WriteAt(p[n:n+canWrite], off)
154+
if end := off + int64(nn); end > b.written {
155+
b.written = end
156+
}
157+
n += nn
158+
if fileErr != nil {
159+
return n, fileErr
160+
}
161+
if nn < canWrite {
162+
return n, io.ErrShortWrite
163+
}
164+
if n == len(p) {
165+
return n, nil
166+
}
167+
off = 0
168+
}
169+
170+
return n, io.ErrShortWrite
171+
}
172+
173+
func NewFileCache(blockSize int64) (FileCache, error) {
174+
f, err := os.CreateTemp(conf.Conf.TempDir, "file-*")
175+
if err != nil {
176+
return nil, err
177+
}
178+
err = f.Truncate(blockSize)
179+
if err == nil {
180+
return &singleFileCache{File: f, size: blockSize}, nil
181+
}
182+
return &MultiFileCache{
183+
blocks: []*fileBlock{{file: f, size: blockSize}},
184+
size: blockSize,
185+
}, nil
186+
}

‎internal/cache/file_test.go‎

Lines changed: 101 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,101 @@
1+
package cache_test
2+
3+
import (
4+
"bytes"
5+
"errors"
6+
"io"
7+
"os"
8+
"reflect"
9+
"testing"
10+
11+
"github.com/OpenListTeam/OpenList/v4/internal/cache"
12+
"github.com/OpenListTeam/OpenList/v4/internal/conf"
13+
)
14+
15+
func TestFile(t *testing.T) {
16+
f, err := os.CreateTemp("", "writeat-*")
17+
if err != nil {
18+
t.Error(err)
19+
return
20+
}
21+
defer os.Remove(f.Name())
22+
defer f.Close()
23+
t.Run("ReadAt", func(t *testing.T) {
24+
_, err := f.ReadAt(make([]byte, 1), 20)
25+
if err != nil && !errors.Is(err, io.EOF) {
26+
t.Error(err)
27+
}
28+
})
29+
t.Run("WriteAt", func(t *testing.T) {
30+
n, err := f.WriteAt([]byte("abc"), 20)
31+
if err != nil {
32+
t.Errorf("write n=%d err=%v", n, err)
33+
return
34+
}
35+
stat, err := f.Stat()
36+
if err != nil {
37+
t.Errorf("stat err=%v", err)
38+
return
39+
}
40+
if stat.Size() != 23 {
41+
t.Fatalf("unexpected size: got %d want 23", stat.Size())
42+
}
43+
44+
b := make([]byte, stat.Size())
45+
rn, rerr := f.ReadAt(b, 0)
46+
if rn != len(b) || rerr != nil {
47+
t.Fatalf("read n=%d err=%v", rn, rerr)
48+
}
49+
want := append(make([]byte, 20), []byte("abc")...)
50+
if !reflect.DeepEqual(b, want) {
51+
t.Fatalf("unexpected content: got %v want %v", b, want)
52+
}
53+
})
54+
}
55+
56+
func TestMultiFileCache(t *testing.T) {
57+
prevConf := conf.Conf
58+
t.Cleanup(func() {
59+
conf.Conf = prevConf
60+
})
61+
conf.Conf = &conf.Config{}
62+
f := cache.MultiFileCache{}
63+
defer f.Close()
64+
t.Run("ReadAt", func(t *testing.T) {
65+
_, err := f.ReadAt(make([]byte, 1), 20)
66+
if err != nil && !errors.Is(err, io.EOF) {
67+
t.Error(err)
68+
}
69+
})
70+
t.Run("WriteAt", func(t *testing.T) {
71+
err := f.Truncate(15)
72+
if err != nil {
73+
t.Errorf("truncate err=%v", err)
74+
return
75+
}
76+
n, err := f.WriteAt([]byte("abc"), 10)
77+
if err != nil {
78+
t.Errorf("write n=%d err=%v", n, err)
79+
return
80+
}
81+
82+
err = f.Truncate(30)
83+
if err != nil {
84+
t.Errorf("truncate err=%v", err)
85+
return
86+
}
87+
_, _ = f.WriteAt([]byte("123"), 15)
88+
89+
b := append(make([]byte, 17), []byte("def")...)
90+
b[0] = 'a'
91+
rn, rerr := f.ReadAt(b, 8)
92+
if rn != len(b) || rerr != nil {
93+
t.Fatalf("read n=%d err=%v", rn, rerr)
94+
}
95+
want := []byte{0, 0, 'a', 'b', 'c', 0, 0, '1', '2', '3'}
96+
want = append(want, make([]byte, 10)...)
97+
if !bytes.Equal(b, want) {
98+
t.Fatalf("unexpected content: got %v want %v", b, want)
99+
}
100+
})
101+
}

‎internal/conf/config.go‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -120,8 +120,9 @@ type Config struct {
120120
DistDir string `json:"dist_dir"`
121121
Log LogConfig `json:"log" envPrefix:"LOG_"`
122122
DelayedStart int `json:"delayed_start" env:"DELAYED_START"`
123-
MaxBufferLimit int `json:"max_buffer_limitMB" env:"MAX_BUFFER_LIMIT_MB"`
124-
MmapThreshold int `json:"mmap_thresholdMB" env:"MMAP_THRESHOLD_MB"`
123+
MinFreeMemory int `json:"min_free_memory" env:"MIN_FREE_MEMORY"`
124+
MaxBlockLimit int `json:"max_block_limit" env:"MAX_BLOCK_LIMIT"`
125+
CacheThreshold int `json:"cache_threshold" env:"CACHE_THRESHOLD"`
125126
MaxConnections int `json:"max_connections" env:"MAX_CONNECTIONS"`
126127
MaxConcurrency int `json:"max_concurrency" env:"MAX_CONCURRENCY"`
127128
TlsInsecureSkipVerify bool `json:"tls_insecure_skip_verify" env:"TLS_INSECURE_SKIP_VERIFY"`
@@ -178,8 +179,7 @@ func DefaultConfig(dataDir string) *Config {
178179
},
179180
},
180181
},
181-
MaxBufferLimit: -1,
182-
MmapThreshold: 4,
182+
CacheThreshold: 4,
183183
MaxConnections: 0,
184184
MaxConcurrency: 64,
185185
TlsInsecureSkipVerify: false,

0 commit comments

Comments
 (0)