refactor: improve upload handling

drivers/115_open/upload.go

@@ -107,16 +107,16 @@ func (d *Open115) multpartUpload(ctx context.Context, stream model.FileStreamer,
 		if err != nil {
 			return err
 		}
-		rateLimitedRd := driver.NewLimitedUploadStream(ctx, rd)
 		err = retry.Do(func() error {
 			rd.Seek(0, io.SeekStart)
-			part, err := bucket.UploadPart(imur, rateLimitedRd, partSize, int(i))
+			part, err := bucket.UploadPart(imur, driver.NewLimitedUploadStream(ctx, rd), partSize, int(i))
 			if err != nil {
 				return err
 			}
 			parts[i-1] = part
 			return nil
 		},
+			retry.Context(ctx),
 			retry.Attempts(3),
 			retry.DelayType(retry.BackOffDelay),
 			retry.Delay(time.Second))

drivers/123/upload.go

@@ -125,27 +125,18 @@ func (d *Pan123) newUpload(ctx context.Context, upReq *UploadResp, file model.Fi
 				curSize = lastChunkSize
 			}
 			var reader io.ReadSeeker
-			var rateLimitedRd io.Reader
 			threadG.GoWithLifecycle(errgroup.Lifecycle{
-				Before: func(ctx context.Context) error {
-					if reader == nil {
-						var err error
-						reader, err = ss.GetSectionReader(offset, curSize)
-						if err != nil {
-							return err
-						}
-						rateLimitedRd = driver.NewLimitedUploadStream(ctx, reader)
-					}
-					return nil
+				Before: func(ctx context.Context) (err error) {
+					reader, err = ss.GetSectionReader(offset, curSize)
+					return
 				},
-				Do: func(ctx context.Context) error {
+				Do: func(ctx context.Context) (err error) {
 					reader.Seek(0, io.SeekStart)
 					uploadUrl := s3PreSignedUrls.Data.PreSignedUrls[strconv.Itoa(cur)]
 					if uploadUrl == "" {
 						return fmt.Errorf("upload url is empty, s3PreSignedUrls: %+v", s3PreSignedUrls)
 					}
-					reader.Seek(0, io.SeekStart)
-					req, err := http.NewRequestWithContext(ctx, http.MethodPut, uploadUrl, rateLimitedRd)
+					req, err := http.NewRequestWithContext(ctx, http.MethodPut, uploadUrl, driver.NewLimitedUploadStream(ctx, reader))
 					if err != nil {
 						return err
 					}
@@ -157,7 +148,7 @@ func (d *Pan123) newUpload(ctx context.Context, upReq *UploadResp, file model.Fi
 					}
 					defer res.Body.Close()
 					if res.StatusCode == http.StatusForbidden {
-						singleflight.AnyGroup.Do(fmt.Sprintf("Pan123.newUpload_%p", threadG), func() (any, error) {
+						_, err, _ = singleflight.AnyGroup.Do(fmt.Sprintf("Pan123.newUpload_%p", threadG), func() (any, error) {
 							newS3PreSignedUrls, err := getS3UploadUrl(ctx, upReq, cur, end)
 							if err != nil {
 								return nil, err
@@ -177,7 +168,7 @@ func (d *Pan123) newUpload(ctx context.Context, upReq *UploadResp, file model.Fi
 						}
 						return fmt.Errorf("upload s3 chunk %d failed, status code: %d, body: %s", cur, res.StatusCode, body)
 					}
-					progress := 10.0 + 85.0*float64(threadG.Success())/float64(chunkCount)
+					progress := 100 * float64(threadG.Success()+1) / float64(chunkCount+1)
 					up(progress)
 					return nil
 				},

drivers/123_open/upload.go

@@ -73,25 +73,20 @@ func (d *Open123) Upload(ctx context.Context, file model.FileStreamer, createRes
 		// 表单
 		b := bytes.NewBuffer(make([]byte, 0, 2048))
 		threadG.GoWithLifecycle(errgroup.Lifecycle{
-			Before: func(ctx context.Context) error {
-				if reader == nil {
-					var err error
-					// 每个分片一个reader
-					reader, err = ss.GetSectionReader(offset, size)
-					if err != nil {
-						return err
-					}
-					// 计算当前分片的MD5
+			Before: func(ctx context.Context) (err error) {
+				reader, err = ss.GetSectionReader(offset, size)
+				return
+			},
+			Do: func(ctx context.Context) (err error) {
+				reader.Seek(0, io.SeekStart)
+				if sliceMD5 == "" {
+					// 把耗时的计算放在这里，避免阻塞其他协程
 					sliceMD5, err = utils.HashReader(utils.MD5, reader)
 					if err != nil {
 						return err
 					}
+					reader.Seek(0, io.SeekStart)
 				}
-				return nil
-			},
-			Do: func(ctx context.Context) error {
-				// 重置分片reader位置，因为HashReader、上一次失败已经读取到分片EOF
-				reader.Seek(0, io.SeekStart)
 
 				b.Reset()
 				w := multipart.NewWriter(b)
@@ -140,20 +135,21 @@ func (d *Open123) Upload(ctx context.Context, file model.FileStreamer, createRes
 				if res.StatusCode != 200 {
 					return fmt.Errorf("slice %d upload failed, status code: %d", partNumber, res.StatusCode)
 				}
-				var resp BaseResp
-				respBody, err := io.ReadAll(res.Body)
+				b.Reset()
+				_, err = b.ReadFrom(res.Body)
 				if err != nil {
 					return err
 				}
-				err = json.Unmarshal(respBody, &resp)
+				var resp BaseResp
+				err = json.Unmarshal(b.Bytes(), &resp)
 				if err != nil {
 					return err
 				}
 				if resp.Code != 0 {
 					return fmt.Errorf("slice %d upload failed: %s", partNumber, resp.Message)
 				}
 
-				progress := 10.0 + 85.0*float64(threadG.Success())/float64(uploadNums)
+				progress := 100 * float64(threadG.Success()+1) / float64(uploadNums+1)
 				up(progress)
 				return nil
 			},

drivers/189pc/utils.go

@@ -756,30 +756,24 @@ func (y *Cloud189PC) StreamUpload(ctx context.Context, dstDir model.Obj, file mo
 		}
 		partInfo := ""
 		var reader io.ReadSeeker
-		var rateLimitedRd io.Reader
 		threadG.GoWithLifecycle(errgroup.Lifecycle{
-			Before: func(ctx context.Context) error {
-				if reader == nil {
-					var err error
-					reader, err = ss.GetSectionReader(offset, partSize)
-					if err != nil {
-						return err
-					}
-					silceMd5.Reset()
-					w, err := utils.CopyWithBuffer(writers, reader)
-					if w != partSize {
-						return fmt.Errorf("failed to read all data: (expect =%d, actual =%d) %w", partSize, w, err)
-					}
-					// 计算块md5并进行hex和base64编码
-					md5Bytes := silceMd5.Sum(nil)
-					silceMd5Hexs = append(silceMd5Hexs, strings.ToUpper(hex.EncodeToString(md5Bytes)))
-					partInfo = fmt.Sprintf("%d-%s", i, base64.StdEncoding.EncodeToString(md5Bytes))
-
-					rateLimitedRd = driver.NewLimitedUploadStream(ctx, reader)
+			Before: func(ctx context.Context) (err error) {
+				reader, err = ss.GetSectionReader(offset, partSize)
+				if err != nil {
+					return err
 				}
+				silceMd5.Reset()
+				w, err := utils.CopyWithBuffer(writers, reader)
+				if w != partSize {
+					return fmt.Errorf("failed to read all data: (expect =%d, actual =%d) %w", partSize, w, err)
+				}
+				// 计算块md5并进行hex和base64编码
+				md5Bytes := silceMd5.Sum(nil)
+				silceMd5Hexs = append(silceMd5Hexs, strings.ToUpper(hex.EncodeToString(md5Bytes)))
+				partInfo = fmt.Sprintf("%d-%s", i, base64.StdEncoding.EncodeToString(md5Bytes))
 				return nil
 			},
-			Do: func(ctx context.Context) error {
+			Do: func(ctx context.Context) (err error) {
 				reader.Seek(0, io.SeekStart)
 				uploadUrls, err := y.GetMultiUploadUrls(ctx, isFamily, initMultiUpload.Data.UploadFileID, partInfo)
 				if err != nil {
@@ -788,11 +782,11 @@ func (y *Cloud189PC) StreamUpload(ctx context.Context, dstDir model.Obj, file mo
 
 				// step.4 上传切片
 				uploadUrl := uploadUrls[0]
-				_, err = y.put(ctx, uploadUrl.RequestURL, uploadUrl.Headers, false, rateLimitedRd, isFamily)
+				_, err = y.put(ctx, uploadUrl.RequestURL, uploadUrl.Headers, false, driver.NewLimitedUploadStream(ctx, reader), isFamily)
 				if err != nil {
 					return err
 				}
-				up(float64(threadG.Success()) * 100 / float64(count))
+				up(float64(threadG.Success()+1) * 100 / float64(count+1))
 				return nil
 			},
 			After: func(err error) {
@@ -804,6 +798,7 @@ func (y *Cloud189PC) StreamUpload(ctx context.Context, dstDir model.Obj, file mo
 	if err = threadG.Wait(); err != nil {
 		return nil, err
 	}
+	defer up(100)
 
 	if fileMd5 != nil {
 		fileMd5Hex = strings.ToUpper(hex.EncodeToString(fileMd5.Sum(nil)))
@@ -995,7 +990,7 @@ func (y *Cloud189PC) FastUpload(ctx context.Context, dstDir model.Obj, file mode
 					return err
 				}
 
-				up(float64(threadG.Success()) * 100 / float64(len(uploadUrls)))
+				up(float64(threadG.Success()+1) * 100 / float64(len(uploadUrls)+1))
 				uploadProgress.UploadParts[i] = ""
 				return nil
 			})
@@ -1007,6 +1002,7 @@ func (y *Cloud189PC) FastUpload(ctx context.Context, dstDir model.Obj, file mode
 			}
 			return nil, err
 		}
+		defer up(100)
 	}
 
 	// step.5 提交

drivers/aliyundrive_open/upload.go

@@ -242,11 +242,11 @@ func (d *AliyundriveOpen) upload(ctx context.Context, dstDir model.Obj, stream m
 			if err != nil {
 				return nil, err
 			}
-			rateLimitedRd := driver.NewLimitedUploadStream(ctx, rd)
 			err = retry.Do(func() error {
 				rd.Seek(0, io.SeekStart)
-				return d.uploadPart(ctx, rateLimitedRd, createResp.PartInfoList[i])
+				return d.uploadPart(ctx, driver.NewLimitedUploadStream(ctx, rd), createResp.PartInfoList[i])
 			},
+				retry.Context(ctx),
 				retry.Attempts(3),
 				retry.DelayType(retry.BackOffDelay),
 				retry.Delay(time.Second))