Skip to content

Commit 57d5ed1

Browse files
j2rong4cnj2rong4cn
committed
优化、修复bug
* 解决alias并发响应时Link返回MFile线程不安全
1 parent acaa4ad commit 57d5ed1

6 files changed

Lines changed: 109 additions & 84 deletions

File tree

‎drivers/alias/driver.go‎

Lines changed: 52 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -78,10 +78,18 @@ func (d *Alias) Get(ctx context.Context, path string) (model.Obj, error) {
7878
return nil, errs.ObjectNotFound
7979
}
8080
for _, dst := range dsts {
81-
obj, err := d.get(ctx, path, dst, sub)
82-
if err == nil {
83-
return obj, nil
81+
obj, err := fs.Get(ctx, stdpath.Join(dst, sub), &fs.GetArgs{NoLog: true})
82+
if err != nil {
83+
continue
8484
}
85+
return &model.Object{
86+
Path: path,
87+
Name: obj.GetName(),
88+
Size: obj.GetSize(),
89+
Modified: obj.ModTime(),
90+
IsFolder: obj.IsDir(),
91+
HashInfo: obj.GetHash(),
92+
}, nil
8593
}
8694
return nil, errs.ObjectNotFound
8795
}
@@ -99,7 +107,27 @@ func (d *Alias) List(ctx context.Context, dir model.Obj, args model.ListArgs) ([
99107
var objs []model.Obj
100108
fsArgs := &fs.ListArgs{NoLog: true, Refresh: args.Refresh}
101109
for _, dst := range dsts {
102-
tmp, err := d.list(ctx, dst, sub, fsArgs)
110+
tmp, err := fs.List(ctx, stdpath.Join(dst, sub), fsArgs)
111+
if err == nil {
112+
tmp, err = utils.SliceConvert(tmp, func(obj model.Obj) (model.Obj, error) {
113+
thumb, ok := model.GetThumb(obj)
114+
objRes := model.Object{
115+
Name: obj.GetName(),
116+
Size: obj.GetSize(),
117+
Modified: obj.ModTime(),
118+
IsFolder: obj.IsDir(),
119+
}
120+
if !ok {
121+
return &objRes, nil
122+
}
123+
return &model.ObjThumb{
124+
Object: objRes,
125+
Thumbnail: model.Thumbnail{
126+
Thumbnail: thumb,
127+
},
128+
}, nil
129+
})
130+
}
103131
if err == nil {
104132
objs = append(objs, tmp...)
105133
}
@@ -113,42 +141,50 @@ func (d *Alias) Link(ctx context.Context, file model.Obj, args model.LinkArgs) (
113141
if !ok {
114142
return nil, errs.ObjectNotFound
115143
}
144+
// proxy || ftp,s3
145+
if common.GetApiUrl(ctx) == "" {
146+
args.Redirect = false
147+
}
116148
for _, dst := range dsts {
117149
reqPath := stdpath.Join(dst, sub)
118150
link, fi, err := d.link(ctx, reqPath, args)
119151
if err != nil {
120152
continue
121153
}
122154
if link == nil {
123-
// 重定向
155+
// 重定向且需要通过代理
124156
return &model.Link{
125157
URL: fmt.Sprintf("%s/p%s?sign=%s",
126158
common.GetApiUrl(ctx),
127159
utils.EncodePath(reqPath, true),
128160
sign.Sign(reqPath)),
129161
}, nil
130162
}
131-
132-
if args.Redirect || link.MFile == nil && d.DownloadConcurrency <= 0 && d.DownloadPartSize <= 0 {
133-
// 不修改link的字段，可直接返回
163+
if args.Redirect {
134164
return link, nil
135165
}
136166

137167
resultLink := &model.Link{
138168
URL: link.URL,
139169
Header: link.Header,
140170
RangeReader: link.RangeReader,
141-
Concurrency: d.DownloadConcurrency,
142-
PartSize: d.DownloadPartSize * utils.KB,
171+
MFile: link.MFile,
172+
Concurrency: link.Concurrency,
173+
PartSize: link.PartSize,
143174
ContentLength: link.ContentLength,
144175
SyncClosers: utils.NewSyncClosers(link),
145176
}
146-
if link.MFile != nil {
147-
resultLink.RangeReader = &model.FileRangeReader{
148-
// MFile并发响应 线程不安全
149-
// 包装成RangeReader 使用io.ReaderAt接口实现线程安全
150-
RangeReaderIF: stream.GetRangeReaderFromMFile(fi.GetSize(), link.MFile),
151-
}
177+
if resultLink.ContentLength == 0 {
178+
resultLink.ContentLength = fi.GetSize()
179+
}
180+
if resultLink.MFile != nil {
181+
return resultLink, nil
182+
}
183+
if d.DownloadConcurrency > 0 {
184+
resultLink.Concurrency = d.DownloadConcurrency
185+
}
186+
if d.DownloadPartSize > 0 {
187+
resultLink.PartSize = d.DownloadPartSize * utils.KB
152188
}
153189
return resultLink, nil
154190
}

‎drivers/alias/util.go‎

Lines changed: 1 addition & 44 deletions
Original file line numberDiff line numberDiff line change
@@ -54,55 +54,12 @@ func (d *Alias) getRootAndPath(path string) (string, string) {
5454
return parts[0], parts[1]
5555
}
5656

57-
func (d *Alias) get(ctx context.Context, path string, dst, sub string) (model.Obj, error) {
58-
obj, err := fs.Get(ctx, stdpath.Join(dst, sub), &fs.GetArgs{NoLog: true})
59-
if err != nil {
60-
return nil, err
61-
}
62-
return &model.Object{
63-
Path: path,
64-
Name: obj.GetName(),
65-
Size: obj.GetSize(),
66-
Modified: obj.ModTime(),
67-
IsFolder: obj.IsDir(),
68-
HashInfo: obj.GetHash(),
69-
}, nil
70-
}
71-
72-
func (d *Alias) list(ctx context.Context, dst, sub string, args *fs.ListArgs) ([]model.Obj, error) {
73-
objs, err := fs.List(ctx, stdpath.Join(dst, sub), args)
74-
// the obj must implement the model.SetPath interface
75-
// return objs, err
76-
if err != nil {
77-
return nil, err
78-
}
79-
return utils.SliceConvert(objs, func(obj model.Obj) (model.Obj, error) {
80-
thumb, ok := model.GetThumb(obj)
81-
objRes := model.Object{
82-
Name: obj.GetName(),
83-
Size: obj.GetSize(),
84-
Modified: obj.ModTime(),
85-
IsFolder: obj.IsDir(),
86-
}
87-
if !ok {
88-
return &objRes, nil
89-
}
90-
return &model.ObjThumb{
91-
Object: objRes,
92-
Thumbnail: model.Thumbnail{
93-
Thumbnail: thumb,
94-
},
95-
}, nil
96-
})
97-
}
98-
9957
func (d *Alias) link(ctx context.Context, reqPath string, args model.LinkArgs) (*model.Link, model.Obj, error) {
10058
storage, reqActualPath, err := op.GetStorageAndActualPath(reqPath)
10159
if err != nil {
10260
return nil, nil, err
10361
}
104-
// proxy || ftp,s3
105-
if !args.Redirect || len(common.GetApiUrl(ctx)) == 0 {
62+
if !args.Redirect {
10663
return op.Link(ctx, storage, reqActualPath, args)
10764
}
10865
obj, err := fs.Get(ctx, reqPath, &fs.GetArgs{NoLog: true})

‎drivers/local/driver.go‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -245,12 +245,13 @@ func (d *Local) Link(ctx context.Context, file model.Obj, args model.LinkArgs) (
245245
if err != nil {
246246
return nil, err
247247
}
248+
link.ContentLength = file.GetSize()
248249
link.MFile = open
249250
}
250251
link.AddIfCloser(link.MFile)
251252
if link.MFile != nil && !d.Config().OnlyLinkMFile {
252253
link.RangeReader = &model.FileRangeReader{
253-
RangeReaderIF: stream.GetRangeReaderFromMFile(file.GetSize(), link.MFile),
254+
RangeReaderIF: stream.GetRangeReaderFromMFile(link.ContentLength, link.MFile),
254255
}
255256
link.MFile = nil
256257
}

‎internal/op/archive.go‎

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -372,11 +372,16 @@ func DriverExtract(ctx context.Context, storage driver.Driver, path string, args
372372
}
373373

374374
var forget any
375+
var linkM *extractLink
375376
fn := func() (*extractLink, error) {
376377
link, err := driverExtract(ctx, storage, path, args)
377378
if err != nil {
378379
return nil, errors.Wrapf(err, "failed extract archive")
379380
}
381+
if link.MFile != nil && forget != nil {
382+
linkM = link
383+
return nil, errLinkMFileCache
384+
}
380385
if link.Link.Expiration != nil {
381386
extractCache.Set(key, link, cache.WithEx[*extractLink](*link.Link.Expiration))
382387
}
@@ -406,11 +411,18 @@ func DriverExtract(ctx context.Context, storage driver.Driver, path string, args
406411
link.AcquireReference()
407412
}
408413
}
414+
if err == errLinkMFileCache {
415+
if linkM != nil {
416+
return linkM.Link, linkM.Obj, nil
417+
}
418+
forget = nil
419+
link, err = fn()
420+
}
409421

410422
if err != nil {
411423
return nil, nil, err
412424
}
413-
return link.Link, link.Obj, err
425+
return link.Link, link.Obj, nil
414426
}
415427

416428
func driverExtract(ctx context.Context, storage driver.Driver, path string, args model.ArchiveInnerArgs) (*extractLink, error) {

‎internal/op/fs.go‎

Lines changed: 17 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ package op
22

33
import (
44
"context"
5+
stderrors "errors"
56
stdpath "path"
67
"slices"
78
"strings"
@@ -250,6 +251,7 @@ func GetUnwrap(ctx context.Context, storage driver.Driver, path string) (model.O
250251

251252
var linkCache = cache.NewMemCache(cache.WithShards[*model.Link](16))
252253
var linkG = singleflight.Group[*model.Link]{Remember: true}
254+
var errLinkMFileCache = stderrors.New("ErrLinkMFileCache")
253255

254256
// Link get link, if is an url. should have an expiry time
255257
func Link(ctx context.Context, storage driver.Driver, path string, args model.LinkArgs) (*model.Link, model.Obj, error) {
@@ -292,11 +294,16 @@ func Link(ctx context.Context, storage driver.Driver, path string, args model.Li
292294
}
293295

294296
var forget any
297+
var linkM *model.Link
295298
fn := func() (*model.Link, error) {
296299
link, err := storage.Link(ctx, file, args)
297300
if err != nil {
298301
return nil, errors.Wrapf(err, "failed get link")
299302
}
303+
if link.MFile != nil && forget != nil {
304+
linkM = link
305+
return nil, errLinkMFileCache
306+
}
300307
if link.Expiration != nil {
301308
linkCache.Set(key, link, cache.WithEx[*model.Link](*link.Expiration))
302309
}
@@ -326,11 +333,19 @@ func Link(ctx context.Context, storage driver.Driver, path string, args model.Li
326333
link.AcquireReference()
327334
}
328335
}
336+
337+
if err == errLinkMFileCache {
338+
if linkM != nil {
339+
return linkM, file, nil
340+
}
341+
forget = nil
342+
link, err = fn()
343+
}
344+
329345
if err != nil {
330346
return nil, nil, err
331347
}
332-
333-
return link, file, err
348+
return link, file, nil
334349
}
335350

336351
// Other api

‎pkg/utils/io.go‎

Lines changed: 24 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,9 @@ import (
66
"errors"
77
"fmt"
88
"io"
9+
"math"
910
"sync"
11+
"sync/atomic"
1012
"time"
1113

1214
log "github.com/sirupsen/logrus"
@@ -191,56 +193,58 @@ type SyncClosersIF interface {
191193

192194
type SyncClosers struct {
193195
closers []io.Closer
194-
mu sync.Mutex
195-
ref int
196+
ref atomic.Int32
196197
}
197198

198199
var _ SyncClosersIF = (*SyncClosers)(nil)
199200

200201
func (c *SyncClosers) AcquireReference() bool {
201-
c.mu.Lock()
202-
defer c.mu.Unlock()
203-
if len(c.closers) == 0 {
204-
return false
202+
ref := c.ref.Add(1)
203+
if ref > 0 {
204+
log.Debugf("SyncClosers.AcquireReference %p,ref=%d\n", c, ref)
205+
return true
205206
}
206-
c.ref++
207-
log.Debugf("SyncClosers.AcquireReference %p,ref=%d\n", c, c.ref)
208-
return true
207+
c.ref.Store(math.MinInt16)
208+
return false
209209
}
210210

211211
func (c *SyncClosers) Close() error {
212-
c.mu.Lock()
213-
defer c.mu.Unlock()
214-
defer log.Debugf("SyncClosers.Close %p,ref=%d\n", c, c.ref)
215-
if c.ref > 1 {
216-
c.ref--
212+
ref := c.ref.Add(-1)
213+
if ref < -1 {
214+
c.ref.Store(math.MinInt16)
217215
return nil
218216
}
219-
c.ref = 0
217+
log.Debugf("SyncClosers.Close %p,ref=%d\n", c, ref+1)
218+
if ref > 0 {
219+
return nil
220+
}
221+
c.ref.Store(math.MinInt16)
220222

221223
var errs []error
222224
for _, closer := range c.closers {
223225
if closer != nil {
224226
errs = append(errs, closer.Close())
225227
}
226228
}
227-
c.closers = c.closers[:0]
229+
clear(c.closers)
228230
return errors.Join(errs...)
229231
}
230232

231233
func (c *SyncClosers) Add(closer io.Closer) {
232234
if closer != nil {
233-
c.mu.Lock()
235+
if c.ref.Load() < 0 {
236+
panic("Not reusable")
237+
}
234238
c.closers = append(c.closers, closer)
235-
c.mu.Unlock()
236239
}
237240
}
238241

239242
func (c *SyncClosers) AddIfCloser(a any) {
240243
if closer, ok := a.(io.Closer); ok {
241-
c.mu.Lock()
244+
if c.ref.Load() < 0 {
245+
panic("Not reusable")
246+
}
242247
c.closers = append(c.closers, closer)
243-
c.mu.Unlock()
244248
}
245249
}
246250

0 commit comments

Comments
 (0)