Skip to content

Commit 8cf1518

Browse files
j2rong4cnxrgzs
j2rong4cn
and
xrgzs
authored
perf: optimize upload (#554)
* pref(115,123): optimize upload * chore * aliyun_open, google_drive * fix bug * chore * cloudreve, cloudreve_v4, onedrive, onedrive_app * chore(conf): add `max_buffer_limit` option * 123pan multithread upload * doubao * google_drive * chore * chore * chore: 计算分片数量的代码 * MaxBufferLimit自动挡 * MaxBufferLimit自动挡 * 189pc * errorgroup添加Lifecycle * 查缺补漏 * Conf.MaxBufferLimit单位为MB * 。 --------- Co-authored-by: MadDogOwner <[email protected]>
1 parent c8f2aaa commit 8cf1518

File tree

43 files changed

+771
-459
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

43 files changed

+771
-459
lines changed

‎drivers/115_open/upload.go‎

Lines changed: 14 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import (
99
sdk "github.com/OpenListTeam/115-sdk-go"
1010
"github.com/OpenListTeam/OpenList/v4/internal/driver"
1111
"github.com/OpenListTeam/OpenList/v4/internal/model"
12+
streamPkg "github.com/OpenListTeam/OpenList/v4/internal/stream"
1213
"github.com/OpenListTeam/OpenList/v4/pkg/utils"
1314
"github.com/aliyun/aliyun-oss-go-sdk/oss"
1415
"github.com/avast/retry-go"
@@ -69,9 +70,6 @@ func (d *Open115) singleUpload(ctx context.Context, tempF model.File, tokenResp
6970
// }
7071

7172
func (d *Open115) multpartUpload(ctx context.Context, stream model.FileStreamer, up driver.UpdateProgress, tokenResp *sdk.UploadGetTokenResp, initResp *sdk.UploadInitResp) error {
72-
fileSize := stream.GetSize()
73-
chunkSize := calPartSize(fileSize)
74-
7573
ossClient, err := oss.New(tokenResp.Endpoint, tokenResp.AccessKeyId, tokenResp.AccessKeySecret, oss.SecurityToken(tokenResp.SecurityToken))
7674
if err != nil {
7775
return err
@@ -86,9 +84,15 @@ func (d *Open115) multpartUpload(ctx context.Context, stream model.FileStreamer,
8684
return err
8785
}
8886

87+
fileSize := stream.GetSize()
88+
chunkSize := calPartSize(fileSize)
8989
partNum := (stream.GetSize() + chunkSize - 1) / chunkSize
9090
parts := make([]oss.UploadPart, partNum)
9191
offset := int64(0)
92+
ss, err := streamPkg.NewStreamSectionReader(stream, int(chunkSize))
93+
if err != nil {
94+
return err
95+
}
9296
for i := int64(1); i <= partNum; i++ {
9397
if utils.IsCanceled(ctx) {
9498
return ctx.Err()
@@ -98,10 +102,13 @@ func (d *Open115) multpartUpload(ctx context.Context, stream model.FileStreamer,
98102
if i == partNum {
99103
partSize = fileSize - (i-1)*chunkSize
100104
}
101-
rd := utils.NewMultiReadable(io.LimitReader(stream, partSize))
105+
rd, err := ss.GetSectionReader(offset, partSize)
106+
if err != nil {
107+
return err
108+
}
109+
rateLimitedRd := driver.NewLimitedUploadStream(ctx, rd)
102110
err = retry.Do(func() error {
103-
_ = rd.Reset()
104-
rateLimitedRd := driver.NewLimitedUploadStream(ctx, rd)
111+
rd.Seek(0, io.SeekStart)
105112
part, err := bucket.UploadPart(imur, rateLimitedRd, partSize, int(i))
106113
if err != nil {
107114
return err
@@ -112,6 +119,7 @@ func (d *Open115) multpartUpload(ctx context.Context, stream model.FileStreamer,
112119
retry.Attempts(3),
113120
retry.DelayType(retry.BackOffDelay),
114121
retry.Delay(time.Second))
122+
ss.RecycleSectionReader(rd)
115123
if err != nil {
116124
return err
117125
}

‎drivers/123/meta.go‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,8 @@ type Addition struct {
1111
driver.RootID
1212
//OrderBy string `json:"order_by" type:"select" options:"file_id,file_name,size,update_at" default:"file_name"`
1313
//OrderDirection string `json:"order_direction" type:"select" options:"asc,desc" default:"asc"`
14-
AccessToken string
14+
AccessToken string
15+
UploadThread int `json:"UploadThread" type:"number" default:"3" help:"the threads of upload"`
1516
}
1617

1718
var config = driver.Config{
@@ -22,6 +23,11 @@ var config = driver.Config{
2223

2324
func init() {
2425
op.RegisterDriver(func() driver.Driver {
25-
return &Pan123{}
26+
// 新增默认选项 要在RegisterDriver初始化设置 才会对正在使用的用户生效
27+
return &Pan123{
28+
Addition: Addition{
29+
UploadThread: 3,
30+
},
31+
}
2632
})
2733
}

‎drivers/123/upload.go‎

Lines changed: 94 additions & 62 deletions
Original file line numberDiff line numberDiff line change
@@ -6,11 +6,16 @@ import (
66
"io"
77
"net/http"
88
"strconv"
9+
"time"
910

1011
"github.com/OpenListTeam/OpenList/v4/drivers/base"
1112
"github.com/OpenListTeam/OpenList/v4/internal/driver"
1213
"github.com/OpenListTeam/OpenList/v4/internal/model"
14+
"github.com/OpenListTeam/OpenList/v4/internal/stream"
15+
"github.com/OpenListTeam/OpenList/v4/pkg/errgroup"
16+
"github.com/OpenListTeam/OpenList/v4/pkg/singleflight"
1317
"github.com/OpenListTeam/OpenList/v4/pkg/utils"
18+
"github.com/avast/retry-go"
1419
"github.com/go-resty/resty/v2"
1520
)
1621

@@ -69,18 +74,15 @@ func (d *Pan123) completeS3(ctx context.Context, upReq *UploadResp, file model.F
6974
}
7075

7176
func (d *Pan123) newUpload(ctx context.Context, upReq *UploadResp, file model.FileStreamer, up driver.UpdateProgress) error {
72-
tmpF, err := file.CacheFullInTempFile()
73-
if err != nil {
74-
return err
75-
}
7677
// fetch s3 pre signed urls
7778
size := file.GetSize()
78-
chunkSize := min(size, 16*utils.MB)
79-
chunkCount := int(size / chunkSize)
79+
chunkSize := int64(16 * utils.MB)
80+
chunkCount := 1
81+
if size > chunkSize {
82+
chunkCount = int((size + chunkSize - 1) / chunkSize)
83+
}
8084
lastChunkSize := size % chunkSize
81-
if lastChunkSize > 0 {
82-
chunkCount++
83-
} else {
85+
if lastChunkSize == 0 {
8486
lastChunkSize = chunkSize
8587
}
8688
// only 1 batch is allowed
@@ -90,73 +92,103 @@ func (d *Pan123) newUpload(ctx context.Context, upReq *UploadResp, file model.Fi
9092
batchSize = 10
9193
getS3UploadUrl = d.getS3PreSignedUrls
9294
}
95+
ss, err := stream.NewStreamSectionReader(file, int(chunkSize))
96+
if err != nil {
97+
return err
98+
}
99+
100+
thread := min(int(chunkCount), d.UploadThread)
101+
threadG, uploadCtx := errgroup.NewOrderedGroupWithContext(ctx, thread,
102+
retry.Attempts(3),
103+
retry.Delay(time.Second),
104+
retry.DelayType(retry.BackOffDelay))
93105
for i := 1; i <= chunkCount; i += batchSize {
94-
if utils.IsCanceled(ctx) {
95-
return ctx.Err()
106+
if utils.IsCanceled(uploadCtx) {
107+
break
96108
}
97109
start := i
98110
end := min(i+batchSize, chunkCount+1)
99-
s3PreSignedUrls, err := getS3UploadUrl(ctx, upReq, start, end)
111+
s3PreSignedUrls, err := getS3UploadUrl(uploadCtx, upReq, start, end)
100112
if err != nil {
101113
return err
102114
}
103115
// upload each chunk
104-
for j := start; j < end; j++ {
105-
if utils.IsCanceled(ctx) {
106-
return ctx.Err()
116+
for cur := start; cur < end; cur++ {
117+
if utils.IsCanceled(uploadCtx) {
118+
break
107119
}
120+
offset := int64(cur-1) * chunkSize
108121
curSize := chunkSize
109-
if j == chunkCount {
122+
if cur == chunkCount {
110123
curSize = lastChunkSize
111124
}
112-
err = d.uploadS3Chunk(ctx, upReq, s3PreSignedUrls, j, end, io.NewSectionReader(tmpF, chunkSize*int64(j-1), curSize), curSize, false, getS3UploadUrl)
113-
if err != nil {
114-
return err
115-
}
116-
up(float64(j) * 100 / float64(chunkCount))
125+
var reader *stream.SectionReader
126+
var rateLimitedRd io.Reader
127+
threadG.GoWithLifecycle(errgroup.Lifecycle{
128+
Before: func(ctx context.Context) error {
129+
if reader == nil {
130+
var err error
131+
reader, err = ss.GetSectionReader(offset, curSize)
132+
if err != nil {
133+
return err
134+
}
135+
rateLimitedRd = driver.NewLimitedUploadStream(ctx, reader)
136+
}
137+
return nil
138+
},
139+
Do: func(ctx context.Context) error {
140+
reader.Seek(0, io.SeekStart)
141+
uploadUrl := s3PreSignedUrls.Data.PreSignedUrls[strconv.Itoa(cur)]
142+
if uploadUrl == "" {
143+
return fmt.Errorf("upload url is empty, s3PreSignedUrls: %+v", s3PreSignedUrls)
144+
}
145+
reader.Seek(0, io.SeekStart)
146+
req, err := http.NewRequestWithContext(ctx, http.MethodPut, uploadUrl, rateLimitedRd)
147+
if err != nil {
148+
return err
149+
}
150+
req.ContentLength = curSize
151+
//req.Header.Set("Content-Length", strconv.FormatInt(curSize, 10))
152+
res, err := base.HttpClient.Do(req)
153+
if err != nil {
154+
return err
155+
}
156+
defer res.Body.Close()
157+
if res.StatusCode == http.StatusForbidden {
158+
singleflight.AnyGroup.Do(fmt.Sprintf("Pan123.newUpload_%p", threadG), func() (any, error) {
159+
newS3PreSignedUrls, err := getS3UploadUrl(ctx, upReq, cur, end)
160+
if err != nil {
161+
return nil, err
162+
}
163+
s3PreSignedUrls.Data.PreSignedUrls = newS3PreSignedUrls.Data.PreSignedUrls
164+
return nil, nil
165+
})
166+
if err != nil {
167+
return err
168+
}
169+
return fmt.Errorf("upload s3 chunk %d failed, status code: %d", cur, res.StatusCode)
170+
}
171+
if res.StatusCode != http.StatusOK {
172+
body, err := io.ReadAll(res.Body)
173+
if err != nil {
174+
return err
175+
}
176+
return fmt.Errorf("upload s3 chunk %d failed, status code: %d, body: %s", cur, res.StatusCode, body)
177+
}
178+
progress := 10.0 + 85.0*float64(threadG.Success())/float64(chunkCount)
179+
up(progress)
180+
return nil
181+
},
182+
After: func(err error) {
183+
ss.RecycleSectionReader(reader)
184+
},
185+
})
117186
}
118187
}
119-
// complete s3 upload
120-
return d.completeS3(ctx, upReq, file, chunkCount > 1)
121-
}
122-
123-
func (d *Pan123) uploadS3Chunk(ctx context.Context, upReq *UploadResp, s3PreSignedUrls *S3PreSignedURLs, cur, end int, reader *io.SectionReader, curSize int64, retry bool, getS3UploadUrl func(ctx context.Context, upReq *UploadResp, start int, end int) (*S3PreSignedURLs, error)) error {
124-
uploadUrl := s3PreSignedUrls.Data.PreSignedUrls[strconv.Itoa(cur)]
125-
if uploadUrl == "" {
126-
return fmt.Errorf("upload url is empty, s3PreSignedUrls: %+v", s3PreSignedUrls)
127-
}
128-
req, err := http.NewRequest("PUT", uploadUrl, driver.NewLimitedUploadStream(ctx, reader))
129-
if err != nil {
188+
if err := threadG.Wait(); err != nil {
130189
return err
131190
}
132-
req = req.WithContext(ctx)
133-
req.ContentLength = curSize
134-
//req.Header.Set("Content-Length", strconv.FormatInt(curSize, 10))
135-
res, err := base.HttpClient.Do(req)
136-
if err != nil {
137-
return err
138-
}
139-
defer res.Body.Close()
140-
if res.StatusCode == http.StatusForbidden {
141-
if retry {
142-
return fmt.Errorf("upload s3 chunk %d failed, status code: %d", cur, res.StatusCode)
143-
}
144-
// refresh s3 pre signed urls
145-
newS3PreSignedUrls, err := getS3UploadUrl(ctx, upReq, cur, end)
146-
if err != nil {
147-
return err
148-
}
149-
s3PreSignedUrls.Data.PreSignedUrls = newS3PreSignedUrls.Data.PreSignedUrls
150-
// retry
151-
reader.Seek(0, io.SeekStart)
152-
return d.uploadS3Chunk(ctx, upReq, s3PreSignedUrls, cur, end, reader, curSize, true, getS3UploadUrl)
153-
}
154-
if res.StatusCode != http.StatusOK {
155-
body, err := io.ReadAll(res.Body)
156-
if err != nil {
157-
return err
158-
}
159-
return fmt.Errorf("upload s3 chunk %d failed, status code: %d, body: %s", cur, res.StatusCode, body)
160-
}
161-
return nil
191+
defer up(100)
192+
// complete s3 upload
193+
return d.completeS3(ctx, upReq, file, chunkCount > 1)
162194
}

‎drivers/123_open/upload.go‎

Lines changed: 46 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -2,15 +2,16 @@ package _123_open
22

33
import (
44
"context"
5+
"io"
56
"net/http"
67
"strings"
78
"time"
89

910
"github.com/OpenListTeam/OpenList/v4/drivers/base"
1011
"github.com/OpenListTeam/OpenList/v4/internal/driver"
1112
"github.com/OpenListTeam/OpenList/v4/internal/model"
13+
"github.com/OpenListTeam/OpenList/v4/internal/stream"
1214
"github.com/OpenListTeam/OpenList/v4/pkg/errgroup"
13-
"github.com/OpenListTeam/OpenList/v4/pkg/http_range"
1415
"github.com/OpenListTeam/OpenList/v4/pkg/utils"
1516
"github.com/avast/retry-go"
1617
"github.com/go-resty/resty/v2"
@@ -79,49 +80,64 @@ func (d *Open123) Upload(ctx context.Context, file model.FileStreamer, createRes
7980
size := file.GetSize()
8081
chunkSize := createResp.Data.SliceSize
8182
uploadNums := (size + chunkSize - 1) / chunkSize
82-
threadG, uploadCtx := errgroup.NewGroupWithContext(ctx, d.UploadThread,
83+
thread := min(int(uploadNums), d.UploadThread)
84+
threadG, uploadCtx := errgroup.NewOrderedGroupWithContext(ctx, thread,
8385
retry.Attempts(3),
8486
retry.Delay(time.Second),
8587
retry.DelayType(retry.BackOffDelay))
8688

89+
ss, err := stream.NewStreamSectionReader(file, int(chunkSize))
90+
if err != nil {
91+
return err
92+
}
8793
for partIndex := int64(0); partIndex < uploadNums; partIndex++ {
8894
if utils.IsCanceled(uploadCtx) {
89-
return ctx.Err()
95+
break
9096
}
9197
partIndex := partIndex
9298
partNumber := partIndex + 1 // 分片号从1开始
9399
offset := partIndex * chunkSize
94100
size := min(chunkSize, size-offset)
95-
limitedReader, err := file.RangeRead(http_range.Range{
96-
Start: offset,
97-
Length: size})
98-
if err != nil {
99-
return err
100-
}
101-
limitedReader = driver.NewLimitedUploadStream(ctx, limitedReader)
102-
103-
threadG.Go(func(ctx context.Context) error {
104-
uploadPartUrl, err := d.url(createResp.Data.PreuploadID, partNumber)
105-
if err != nil {
106-
return err
107-
}
101+
var reader *stream.SectionReader
102+
var rateLimitedRd io.Reader
103+
threadG.GoWithLifecycle(errgroup.Lifecycle{
104+
Before: func(ctx context.Context) error {
105+
if reader == nil {
106+
var err error
107+
reader, err = ss.GetSectionReader(offset, size)
108+
if err != nil {
109+
return err
110+
}
111+
rateLimitedRd = driver.NewLimitedUploadStream(ctx, reader)
112+
}
113+
return nil
114+
},
115+
Do: func(ctx context.Context) error {
116+
reader.Seek(0, io.SeekStart)
117+
uploadPartUrl, err := d.url(createResp.Data.PreuploadID, partNumber)
118+
if err != nil {
119+
return err
120+
}
108121

109-
req, err := http.NewRequestWithContext(ctx, "PUT", uploadPartUrl, limitedReader)
110-
if err != nil {
111-
return err
112-
}
113-
req = req.WithContext(ctx)
114-
req.ContentLength = size
122+
req, err := http.NewRequestWithContext(ctx, http.MethodPut, uploadPartUrl, rateLimitedRd)
123+
if err != nil {
124+
return err
125+
}
126+
req.ContentLength = size
115127

116-
res, err := base.HttpClient.Do(req)
117-
if err != nil {
118-
return err
119-
}
120-
_ = res.Body.Close()
128+
res, err := base.HttpClient.Do(req)
129+
if err != nil {
130+
return err
131+
}
132+
_ = res.Body.Close()
121133

122-
progress := 10.0 + 85.0*float64(threadG.Success())/float64(uploadNums)
123-
up(progress)
124-
return nil
134+
progress := 10.0 + 85.0*float64(threadG.Success())/float64(uploadNums)
135+
up(progress)
136+
return nil
137+
},
138+
After: func(err error) {
139+
ss.RecycleSectionReader(reader)
140+
},
125141
})
126142
}
127143

0 commit comments

Comments
 (0)