perf: optimize IO read/write usage

drivers/115/driver.go

@@ -8,6 +8,7 @@ import (
 	driver115 "github.com/SheltonZhu/115driver/pkg/driver"
 	"github.com/alist-org/alist/v3/internal/driver"
 	"github.com/alist-org/alist/v3/internal/model"
+	streamPkg "github.com/alist-org/alist/v3/internal/stream"
 	"github.com/alist-org/alist/v3/pkg/http_range"
 	"github.com/alist-org/alist/v3/pkg/utils"
 	"github.com/pkg/errors"
@@ -184,12 +185,8 @@ func (d *Pan115) Put(ctx context.Context, dstDir model.Obj, stream model.FileStr
 	}
 	preHash = strings.ToUpper(preHash)
 	fullHash := stream.GetHash().GetHash(utils.SHA1)
-	if len(fullHash) <= 0 {
-		tmpF, err := stream.CacheFullInTempFile()
-		if err != nil {
-			return nil, err
-		}
-		fullHash, err = utils.HashFile(utils.SHA1, tmpF)
+	if len(fullHash) != utils.SHA1.Width {
+		_, fullHash, err = streamPkg.CacheFullInTempFileAndHash(stream, utils.SHA1)
 		if err != nil {
 			return nil, err
 		}

drivers/115_open/driver.go

@@ -15,6 +15,8 @@ import (
 	"github.com/alist-org/alist/v3/internal/driver"
 	"github.com/alist-org/alist/v3/internal/model"
 	"github.com/alist-org/alist/v3/internal/op"
+	"github.com/alist-org/alist/v3/internal/stream"
+	"github.com/alist-org/alist/v3/pkg/http_range"
 	"github.com/alist-org/alist/v3/pkg/utils"
 	"github.com/aliyun/aliyun-oss-go-sdk/oss"
 	sdk "github.com/xhofe/115-sdk-go"
@@ -184,25 +186,29 @@ func (d *Open115) Remove(ctx context.Context, obj model.Obj) error {
 }
 
 func (d *Open115) Put(ctx context.Context, dstDir model.Obj, file model.FileStreamer, up driver.UpdateProgress) error {
-	tempF, err := file.CacheFullInTempFile()
-	if err != nil {
-		return err
-	}
 	// cal full sha1
-	sha1, err := utils.HashReader(utils.SHA1, tempF)
-	if err != nil {
-		return err
-	}
-	_, err = tempF.Seek(0, io.SeekStart)
-	if err != nil {
-		return err
+	sha1 := file.GetHash().GetHash(utils.SHA1)
+	var tempF model.File
+	var err error
+	if len(sha1) != utils.SHA1.Width {
+		tempF, sha1, err = stream.CacheFullInTempFileAndHash(file, utils.SHA1)
+		if err != nil {
+			return err
+		}
+	} else {
+		tempF, err = file.CacheFullInTempFile()
+		if err != nil {
+			return err
+		}
 	}
 	// pre 128k sha1
-	sha1128k, err := utils.HashReader(utils.SHA1, io.LimitReader(tempF, 128*1024))
+	const PreHashSize int64 = 128 * utils.KB
+	hashSize := min(file.GetSize(), PreHashSize)
+	reader, err := file.RangeRead(http_range.Range{Start: 0, Length: hashSize})
 	if err != nil {
 		return err
 	}
-	_, err = tempF.Seek(0, io.SeekStart)
+	sha1128k, err := utils.HashReader(utils.SHA1, reader)
 	if err != nil {
 		return err
 	}

drivers/123/driver.go

@@ -2,11 +2,8 @@ package _123
 
 import (
 	"context"
-	"crypto/md5"
 	"encoding/base64"
-	"encoding/hex"
 	"fmt"
-	"io"
 	"net/http"
 	"net/url"
 	"sync"
@@ -18,6 +15,7 @@ import (
 	"github.com/alist-org/alist/v3/internal/driver"
 	"github.com/alist-org/alist/v3/internal/errs"
 	"github.com/alist-org/alist/v3/internal/model"
+	"github.com/alist-org/alist/v3/internal/stream"
 	"github.com/alist-org/alist/v3/pkg/utils"
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/credentials"
@@ -187,25 +185,12 @@ func (d *Pan123) Remove(ctx context.Context, obj model.Obj) error {
 
 func (d *Pan123) Put(ctx context.Context, dstDir model.Obj, file model.FileStreamer, up driver.UpdateProgress) error {
 	etag := file.GetHash().GetHash(utils.MD5)
-	if len(etag) < utils.MD5.Width {
-		// const DEFAULT int64 = 10485760
-		h := md5.New()
-		// need to calculate md5 of the full content
-		tempFile, err := file.CacheFullInTempFile()
+	var err error
+	if len(etag) != utils.MD5.Width {
+		_, etag, err = stream.CacheFullInTempFileAndHash(file, utils.MD5)
 		if err != nil {
 			return err
 		}
-		defer func() {
-			_ = tempFile.Close()
-		}()
-		if _, err = utils.CopyWithBuffer(h, tempFile); err != nil {
-			return err
-		}
-		_, err = tempFile.Seek(0, io.SeekStart)
-		if err != nil {
-			return err
-		}
-		etag = hex.EncodeToString(h.Sum(nil))
 	}
 	data := base.Json{
 		"driveId":      0,

drivers/123/upload.go

@@ -1,10 +1,10 @@
 package _123
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"io"
-	"math"
 	"net/http"
 	"strconv"
 
@@ -70,27 +70,29 @@ func (d *Pan123) completeS3(ctx context.Context, upReq *UploadResp, file model.F
 }
 
 func (d *Pan123) newUpload(ctx context.Context, upReq *UploadResp, file model.FileStreamer, up driver.UpdateProgress) error {
-	chunkSize := int64(1024 * 1024 * 16)
 	// fetch s3 pre signed urls
-	chunkCount := int(math.Ceil(float64(file.GetSize()) / float64(chunkSize)))
+	size := file.GetSize()
+	chunkSize := min(size, 16*utils.MB)
+	chunkCount := int(size / chunkSize)
+	lastChunkSize := size % chunkSize
+	if lastChunkSize > 0 {
+		chunkCount++
+	}
 	// only 1 batch is allowed
-	isMultipart := chunkCount > 1
 	batchSize := 1
 	getS3UploadUrl := d.getS3Auth
-	if isMultipart {
+	if chunkCount > 1 {
 		batchSize = 10
 		getS3UploadUrl = d.getS3PreSignedUrls
 	}
 	limited := driver.NewLimitedUploadStream(ctx, file)
+	buf := make([]byte, chunkSize)
 	for i := 1; i <= chunkCount; i += batchSize {
 		if utils.IsCanceled(ctx) {
 			return ctx.Err()
 		}
 		start := i
-		end := i + batchSize
-		if end > chunkCount+1 {
-			end = chunkCount + 1
-		}
+		end := min(i+batchSize, chunkCount+1)
 		s3PreSignedUrls, err := getS3UploadUrl(ctx, upReq, start, end)
 		if err != nil {
 			return err
@@ -100,11 +102,17 @@ func (d *Pan123) newUpload(ctx context.Context, upReq *UploadResp, file model.Fi
 			if utils.IsCanceled(ctx) {
 				return ctx.Err()
 			}
-			curSize := chunkSize
-			if j == chunkCount {
-				curSize = file.GetSize() - (int64(chunkCount)-1)*chunkSize
+			if j == chunkCount && lastChunkSize > 0 {
+				buf = buf[:lastChunkSize]
+			}
+			n, err := io.ReadFull(limited, buf)
+			if err == io.ErrUnexpectedEOF {
+				return fmt.Errorf("upload s3 chunk %d failed, can't read data, expected=%d, got=%d", j, len(buf), n)
+			}
+			if err != nil {
+				return err
 			}
-			err = d.uploadS3Chunk(ctx, upReq, s3PreSignedUrls, j, end, io.LimitReader(limited, chunkSize), curSize, false, getS3UploadUrl)
+			err = d.uploadS3Chunk(ctx, upReq, s3PreSignedUrls, j, end, bytes.NewReader(buf), int64(n), false, getS3UploadUrl)
 			if err != nil {
 				return err
 			}
@@ -115,7 +123,7 @@ func (d *Pan123) newUpload(ctx context.Context, upReq *UploadResp, file model.Fi
 	return d.completeS3(ctx, upReq, file, chunkCount > 1)
 }
 
-func (d *Pan123) uploadS3Chunk(ctx context.Context, upReq *UploadResp, s3PreSignedUrls *S3PreSignedURLs, cur, end int, reader io.Reader, curSize int64, retry bool, getS3UploadUrl func(ctx context.Context, upReq *UploadResp, start int, end int) (*S3PreSignedURLs, error)) error {
+func (d *Pan123) uploadS3Chunk(ctx context.Context, upReq *UploadResp, s3PreSignedUrls *S3PreSignedURLs, cur, end int, reader *bytes.Reader, curSize int64, retry bool, getS3UploadUrl func(ctx context.Context, upReq *UploadResp, start int, end int) (*S3PreSignedURLs, error)) error {
 	uploadUrl := s3PreSignedUrls.Data.PreSignedUrls[strconv.Itoa(cur)]
 	if uploadUrl == "" {
 		return fmt.Errorf("upload url is empty, s3PreSignedUrls: %+v", s3PreSignedUrls)
@@ -143,6 +151,7 @@ func (d *Pan123) uploadS3Chunk(ctx context.Context, upReq *UploadResp, s3PreSign
 		}
 		s3PreSignedUrls.Data.PreSignedUrls = newS3PreSignedUrls.Data.PreSignedUrls
 		// retry
+		reader.Seek(0, io.SeekStart)
 		return d.uploadS3Chunk(ctx, upReq, s3PreSignedUrls, cur, end, reader, curSize, true, getS3UploadUrl)
 	}
 	if res.StatusCode != http.StatusOK {

drivers/139/driver.go

@@ -2,19 +2,18 @@ package _139
 
 import (
 	"context"
-	"encoding/base64"
 	"fmt"
 	"io"
 	"net/http"
 	"path"
 	"strconv"
-	"strings"
 	"time"
 
 	"github.com/alist-org/alist/v3/drivers/base"
 	"github.com/alist-org/alist/v3/internal/driver"
 	"github.com/alist-org/alist/v3/internal/errs"
 	"github.com/alist-org/alist/v3/internal/model"
+	streamPkg "github.com/alist-org/alist/v3/internal/stream"
 	"github.com/alist-org/alist/v3/pkg/cron"
 	"github.com/alist-org/alist/v3/pkg/utils"
 	"github.com/alist-org/alist/v3/pkg/utils/random"
@@ -71,28 +70,29 @@ func (d *Yun139) Init(ctx context.Context) error {
 	default:
 		return errs.NotImplement
 	}
-	if d.ref != nil {
-		return nil
-	}
-	decode, err := base64.StdEncoding.DecodeString(d.Authorization)
-	if err != nil {
-		return err
-	}
-	decodeStr := string(decode)
-	splits := strings.Split(decodeStr, ":")
-	if len(splits) < 2 {
-		return fmt.Errorf("authorization is invalid, splits < 2")
-	}
-	d.Account = splits[1]
-	_, err = d.post("/orchestration/personalCloud/user/v1.0/qryUserExternInfo", base.Json{
-		"qryUserExternInfoReq": base.Json{
-			"commonAccountInfo": base.Json{
-				"account":     d.getAccount(),
-				"accountType": 1,
-			},
-		},
-	}, nil)
-	return err
+	// if d.ref != nil {
+	// 	return nil
+	// }
+	// decode, err := base64.StdEncoding.DecodeString(d.Authorization)
+	// if err != nil {
+	// 	return err
+	// }
+	// decodeStr := string(decode)
+	// splits := strings.Split(decodeStr, ":")
+	// if len(splits) < 2 {
+	// 	return fmt.Errorf("authorization is invalid, splits < 2")
+	// }
+	// d.Account = splits[1]
+	// _, err = d.post("/orchestration/personalCloud/user/v1.0/qryUserExternInfo", base.Json{
+	// 	"qryUserExternInfoReq": base.Json{
+	// 		"commonAccountInfo": base.Json{
+	// 			"account":     d.getAccount(),
+	// 			"accountType": 1,
+	// 		},
+	// 	},
+	// }, nil)
+	// return err
+	return nil
 }
 
 func (d *Yun139) InitReference(storage driver.Driver) error {
@@ -526,29 +526,26 @@ func (d *Yun139) Put(ctx context.Context, dstDir model.Obj, stream model.FileStr
 	case MetaPersonalNew:
 		var err error
 		fullHash := stream.GetHash().GetHash(utils.SHA256)
-		if len(fullHash) <= 0 {
-			tmpF, err := stream.CacheFullInTempFile()
-			if err != nil {
-				return err
-			}
-			fullHash, err = utils.HashFile(utils.SHA256, tmpF)
+		if len(fullHash) != utils.SHA256.Width {
+			_, fullHash, err = streamPkg.CacheFullInTempFileAndHash(stream, utils.SHA256)
 			if err != nil {
 				return err
 			}
 		}
 
-		partInfos := []PartInfo{}
-		var partSize = d.getPartSize(stream.GetSize())
-		part := (stream.GetSize() + partSize - 1) / partSize
-		if part == 0 {
-			part = 1
+		size := stream.GetSize()
+		var partSize = d.getPartSize(size)
+		part := size / partSize
+		if size%partSize > 0 {
+			part++
 		}
-		for i := int64(0); i < part; i++ {
+		partInfos := make([]PartInfo, 0, part)
+		for i := range part {
 			if utils.IsCanceled(ctx) {
 				return ctx.Err()
 			}
 			start := i * partSize
-			byteSize := stream.GetSize() - start
+			byteSize := size - start
 			if byteSize > partSize {
 				byteSize = partSize
 			}
@@ -576,7 +573,7 @@ func (d *Yun139) Put(ctx context.Context, dstDir model.Obj, stream model.FileStr
 			"contentType":          "application/octet-stream",
 			"parallelUpload":       false,
 			"partInfos":            firstPartInfos,
-			"size":                 stream.GetSize(),
+			"size":                 size,
 			"parentFileId":         dstDir.GetID(),
 			"name":                 stream.GetName(),
 			"type":                 "file",
@@ -629,7 +626,7 @@ func (d *Yun139) Put(ctx context.Context, dstDir model.Obj, stream model.FileStr
 			}
 
 			// Progress
-			p := driver.NewProgress(stream.GetSize(), up)
+			p := driver.NewProgress(size, up)
 
 			rateLimited := driver.NewLimitedUploadStream(ctx, stream)
 			// 上传所有分片
@@ -780,22 +777,22 @@ func (d *Yun139) Put(ctx context.Context, dstDir model.Obj, stream model.FileStr
 			return err
 		}
 
-		// Progress
-		p := driver.NewProgress(stream.GetSize(), up)
-
-		var partSize = d.getPartSize(stream.GetSize())
-		part := (stream.GetSize() + partSize - 1) / partSize
-		if part == 0 {
-			part = 1
+		size := stream.GetSize()
+		var partSize = d.getPartSize(size)
+		part := size / partSize
+		if size%partSize > 0 {
+			part++
 		}
+		// Progress
+		p := driver.NewProgress(size, up)
 		rateLimited := driver.NewLimitedUploadStream(ctx, stream)
-		for i := int64(0); i < part; i++ {
+		for i := range part {
 			if utils.IsCanceled(ctx) {
 				return ctx.Err()
 			}
 
 			start := i * partSize
-			byteSize := stream.GetSize() - start
+			byteSize := size - start
 			if byteSize > partSize {
 				byteSize = partSize
 			}
@@ -810,7 +807,7 @@ func (d *Yun139) Put(ctx context.Context, dstDir model.Obj, stream model.FileStr
 
 			req = req.WithContext(ctx)
 			req.Header.Set("Content-Type", "text/plain;name="+unicode(stream.GetName()))
-			req.Header.Set("contentSize", strconv.FormatInt(stream.GetSize(), 10))
+			req.Header.Set("contentSize", strconv.FormatInt(size, 10))
 			req.Header.Set("range", fmt.Sprintf("bytes=%d-%d", start, start+byteSize-1))
 			req.Header.Set("uploadtaskID", resp.Data.UploadResult.UploadTaskID)
 			req.Header.Set("rangeType", "0")

drivers/139/util.go

@@ -75,6 +75,7 @@ func (d *Yun139) refreshToken() error {
 	if err != nil {
 		return fmt.Errorf("authorization is invalid")
 	}
+	d.Account = splits[1]
 	expiration -= time.Now().UnixMilli()
 	if expiration > 1000*60*60*24*15 {
 		// Authorization有效期大于15天无需刷新