From b95df1d7457c6a72cf7e8242061e8a52fc9a3095 Mon Sep 17 00:00:00 2001 From: Mmx Date: Thu, 25 Apr 2024 20:11:15 +0800 Subject: [PATCH] perf: use io copy with buffer pool (#6389) * feat: add io methods with buffer * chore: move io.Copy calls to utils.CopyWithBuffer --- drivers/123/driver.go | 2 +- drivers/189pc/utils.go | 2 +- drivers/aliyundrive/driver.go | 2 +- drivers/aliyundrive_open/upload.go | 2 +- drivers/baidu_netdisk/driver.go | 2 +- drivers/baidu_photo/driver.go | 2 +- drivers/chaoxing/driver.go | 2 +- drivers/ilanzou/driver.go | 2 +- drivers/mediatrack/driver.go | 2 +- drivers/pikpak/util.go | 3 ++- drivers/quark_uc/driver.go | 4 ++-- drivers/smb/util.go | 4 ++-- drivers/thunder/util.go | 2 +- internal/net/request.go | 3 ++- internal/net/serve.go | 4 ++-- internal/net/util.go | 3 ++- internal/stream/stream.go | 2 +- pkg/gowebdav/client.go | 3 ++- pkg/utils/file.go | 4 ++-- pkg/utils/hash.go | 2 +- pkg/utils/hash_test.go | 3 +-- pkg/utils/io.go | 31 +++++++++++++++++++++++++++++- 22 files changed, 59 insertions(+), 27 deletions(-) diff --git a/drivers/123/driver.go b/drivers/123/driver.go index f5d981ef636..240027405d5 100644 --- a/drivers/123/driver.go +++ b/drivers/123/driver.go @@ -194,7 +194,7 @@ func (d *Pan123) Put(ctx context.Context, dstDir model.Obj, stream model.FileStr defer func() { _ = tempFile.Close() }() - if _, err = io.Copy(h, tempFile); err != nil { + if _, err = utils.CopyWithBuffer(h, tempFile); err != nil { return err } _, err = tempFile.Seek(0, io.SeekStart) diff --git a/drivers/189pc/utils.go b/drivers/189pc/utils.go index ee96af3e160..a000a84e005 100644 --- a/drivers/189pc/utils.go +++ b/drivers/189pc/utils.go @@ -595,7 +595,7 @@ func (y *Cloud189PC) FastUpload(ctx context.Context, dstDir model.Obj, file mode } silceMd5.Reset() - if _, err := io.CopyN(io.MultiWriter(fileMd5, silceMd5), tempFile, byteSize); err != nil && err != io.EOF { + if _, err := utils.CopyWithBufferN(io.MultiWriter(fileMd5, silceMd5), tempFile, byteSize); err != nil && err != io.EOF { return nil, err } md5Byte := silceMd5.Sum(nil) diff --git a/drivers/aliyundrive/driver.go b/drivers/aliyundrive/driver.go index eab38f58e1c..2a977aa35e5 100644 --- a/drivers/aliyundrive/driver.go +++ b/drivers/aliyundrive/driver.go @@ -194,7 +194,7 @@ func (d *AliDrive) Put(ctx context.Context, dstDir model.Obj, streamer model.Fil } if d.RapidUpload { buf := bytes.NewBuffer(make([]byte, 0, 1024)) - io.CopyN(buf, file, 1024) + utils.CopyWithBufferN(buf, file, 1024) reqBody["pre_hash"] = utils.HashData(utils.SHA1, buf.Bytes()) if localFile != nil { if _, err := localFile.Seek(0, io.SeekStart); err != nil { diff --git a/drivers/aliyundrive_open/upload.go b/drivers/aliyundrive_open/upload.go index 5f57e8b5620..d152836c075 100644 --- a/drivers/aliyundrive_open/upload.go +++ b/drivers/aliyundrive_open/upload.go @@ -136,7 +136,7 @@ func (d *AliyundriveOpen) calProofCode(stream model.FileStreamer) (string, error if err != nil { return "", err } - _, err = io.CopyN(buf, reader, length) + _, err = utils.CopyWithBufferN(buf, reader, length) if err != nil { return "", err } diff --git a/drivers/baidu_netdisk/driver.go b/drivers/baidu_netdisk/driver.go index 43da834a143..ad52a4b5438 100644 --- a/drivers/baidu_netdisk/driver.go +++ b/drivers/baidu_netdisk/driver.go @@ -211,7 +211,7 @@ func (d *BaiduNetdisk) Put(ctx context.Context, dstDir model.Obj, stream model.F if i == count { byteSize = lastBlockSize } - _, err := io.CopyN(io.MultiWriter(fileMd5H, sliceMd5H, slicemd5H2Write), tempFile, byteSize) + _, err := utils.CopyWithBufferN(io.MultiWriter(fileMd5H, sliceMd5H, slicemd5H2Write), tempFile, byteSize) if err != nil && err != io.EOF { return nil, err } diff --git a/drivers/baidu_photo/driver.go b/drivers/baidu_photo/driver.go index c29bc110095..7477a8eb527 100644 --- a/drivers/baidu_photo/driver.go +++ b/drivers/baidu_photo/driver.go @@ -261,7 +261,7 @@ func (d *BaiduPhoto) Put(ctx context.Context, dstDir model.Obj, stream model.Fil if i == count { byteSize = lastBlockSize } - _, err := io.CopyN(io.MultiWriter(fileMd5H, sliceMd5H, slicemd5H2Write), tempFile, byteSize) + _, err := utils.CopyWithBufferN(io.MultiWriter(fileMd5H, sliceMd5H, slicemd5H2Write), tempFile, byteSize) if err != nil && err != io.EOF { return nil, err } diff --git a/drivers/chaoxing/driver.go b/drivers/chaoxing/driver.go index 143235fa481..de122c36c4d 100644 --- a/drivers/chaoxing/driver.go +++ b/drivers/chaoxing/driver.go @@ -229,7 +229,7 @@ func (d *ChaoXing) Put(ctx context.Context, dstDir model.Obj, stream model.FileS if err != nil { return err } - _, err = io.Copy(filePart, stream) + _, err = utils.CopyWithBuffer(filePart, stream) if err != nil { return err } diff --git a/drivers/ilanzou/driver.go b/drivers/ilanzou/driver.go index 341136da1cd..1d8e5d36b09 100644 --- a/drivers/ilanzou/driver.go +++ b/drivers/ilanzou/driver.go @@ -271,7 +271,7 @@ func (d *ILanZou) Put(ctx context.Context, dstDir model.Obj, stream model.FileSt defer func() { _ = tempFile.Close() }() - if _, err = io.Copy(h, tempFile); err != nil { + if _, err = utils.CopyWithBuffer(h, tempFile); err != nil { return nil, err } _, err = tempFile.Seek(0, io.SeekStart) diff --git a/drivers/mediatrack/driver.go b/drivers/mediatrack/driver.go index ef571832eb7..f0f1ded0087 100644 --- a/drivers/mediatrack/driver.go +++ b/drivers/mediatrack/driver.go @@ -206,7 +206,7 @@ func (d *MediaTrack) Put(ctx context.Context, dstDir model.Obj, stream model.Fil return err } h := md5.New() - _, err = io.Copy(h, tempFile) + _, err = utils.CopyWithBuffer(h, tempFile) if err != nil { return err } diff --git a/drivers/pikpak/util.go b/drivers/pikpak/util.go index 02b988bcd64..71ad1dca8a3 100644 --- a/drivers/pikpak/util.go +++ b/drivers/pikpak/util.go @@ -4,6 +4,7 @@ import ( "crypto/sha1" "encoding/hex" "errors" + "github.com/alist-org/alist/v3/pkg/utils" "io" "net/http" @@ -141,7 +142,7 @@ func getGcid(r io.Reader, size int64) (string, error) { readSize := calcBlockSize(size) for { hash2.Reset() - if n, err := io.CopyN(hash2, r, readSize); err != nil && n == 0 { + if n, err := utils.CopyWithBufferN(hash2, r, readSize); err != nil && n == 0 { if err != io.EOF { return "", err } diff --git a/drivers/quark_uc/driver.go b/drivers/quark_uc/driver.go index 291189ce088..8674fbab26f 100644 --- a/drivers/quark_uc/driver.go +++ b/drivers/quark_uc/driver.go @@ -143,7 +143,7 @@ func (d *QuarkOrUC) Put(ctx context.Context, dstDir model.Obj, stream model.File _ = tempFile.Close() }() m := md5.New() - _, err = io.Copy(m, tempFile) + _, err = utils.CopyWithBuffer(m, tempFile) if err != nil { return err } @@ -153,7 +153,7 @@ func (d *QuarkOrUC) Put(ctx context.Context, dstDir model.Obj, stream model.File } md5Str := hex.EncodeToString(m.Sum(nil)) s := sha1.New() - _, err = io.Copy(s, tempFile) + _, err = utils.CopyWithBuffer(s, tempFile) if err != nil { return err } diff --git a/drivers/smb/util.go b/drivers/smb/util.go index f4605536da7..d9fbf6c5a5a 100644 --- a/drivers/smb/util.go +++ b/drivers/smb/util.go @@ -1,7 +1,7 @@ package smb import ( - "io" + "github.com/alist-org/alist/v3/pkg/utils" "io/fs" "net" "os" @@ -74,7 +74,7 @@ func (d *SMB) CopyFile(src, dst string) error { } defer dstfd.Close() - if _, err = io.Copy(dstfd, srcfd); err != nil { + if _, err = utils.CopyWithBuffer(dstfd, srcfd); err != nil { return err } if srcinfo, err = d.fs.Stat(src); err != nil { diff --git a/drivers/thunder/util.go b/drivers/thunder/util.go index f6dec3260cf..3ec8db58ffe 100644 --- a/drivers/thunder/util.go +++ b/drivers/thunder/util.go @@ -190,7 +190,7 @@ func getGcid(r io.Reader, size int64) (string, error) { readSize := calcBlockSize(size) for { hash2.Reset() - if n, err := io.CopyN(hash2, r, readSize); err != nil && n == 0 { + if n, err := utils.CopyWithBufferN(hash2, r, readSize); err != nil && n == 0 { if err != io.EOF { return "", err } diff --git a/internal/net/request.go b/internal/net/request.go index 71f45aa7afc..088ff66ab4f 100644 --- a/internal/net/request.go +++ b/internal/net/request.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "fmt" + "github.com/alist-org/alist/v3/pkg/utils" "io" "math" "net/http" @@ -271,7 +272,7 @@ func (d *downloader) tryDownloadChunk(params *HttpRequestParams, ch *chunk) (int } } - n, err := io.Copy(ch.buf, resp.Body) + n, err := utils.CopyWithBuffer(ch.buf, resp.Body) if err != nil { return n, &errReadingBody{err: err} diff --git a/internal/net/serve.go b/internal/net/serve.go index a0566780759..adee75ae1d6 100644 --- a/internal/net/serve.go +++ b/internal/net/serve.go @@ -162,7 +162,7 @@ func ServeHTTP(w http.ResponseWriter, r *http.Request, name string, modTime time pw.CloseWithError(err) return } - if _, err := io.CopyN(part, reader, ra.Length); err != nil { + if _, err := utils.CopyWithBufferN(part, reader, ra.Length); err != nil { pw.CloseWithError(err) return } @@ -182,7 +182,7 @@ func ServeHTTP(w http.ResponseWriter, r *http.Request, name string, modTime time w.WriteHeader(code) if r.Method != "HEAD" { - written, err := io.CopyN(w, sendContent, sendSize) + written, err := utils.CopyWithBufferN(w, sendContent, sendSize) if err != nil { log.Warnf("ServeHttp error. err: %s ", err) if written != sendSize { diff --git a/internal/net/util.go b/internal/net/util.go index 4347e2c404d..44201859487 100644 --- a/internal/net/util.go +++ b/internal/net/util.go @@ -2,6 +2,7 @@ package net import ( "fmt" + "github.com/alist-org/alist/v3/pkg/utils" "io" "math" "mime/multipart" @@ -330,7 +331,7 @@ func GetRangedHttpReader(readCloser io.ReadCloser, offset, length int64) (io.Rea log.Warnf("offset is more than 100MB, if loading data from internet, high-latency and wasting of bandwidth is expected") } - if _, err := io.Copy(io.Discard, io.LimitReader(readCloser, offset)); err != nil { + if _, err := utils.CopyWithBuffer(io.Discard, io.LimitReader(readCloser, offset)); err != nil { return nil, err } diff --git a/internal/stream/stream.go b/internal/stream/stream.go index 4b882c519e0..40482f45a36 100644 --- a/internal/stream/stream.go +++ b/internal/stream/stream.go @@ -104,7 +104,7 @@ func (f *FileStream) RangeRead(httpRange http_range.Range) (io.Reader, error) { if httpRange.Start == 0 && httpRange.Length <= InMemoryBufMaxSizeBytes && f.peekBuff == nil { bufSize := utils.Min(httpRange.Length, f.GetSize()) newBuf := bytes.NewBuffer(make([]byte, 0, bufSize)) - n, err := io.CopyN(newBuf, f.Reader, bufSize) + n, err := utils.CopyWithBufferN(newBuf, f.Reader, bufSize) if err != nil { return nil, err } diff --git a/pkg/gowebdav/client.go b/pkg/gowebdav/client.go index 2fca0b7f43d..cef501b9a15 100644 --- a/pkg/gowebdav/client.go +++ b/pkg/gowebdav/client.go @@ -4,6 +4,7 @@ import ( "bytes" "encoding/xml" "fmt" + "github.com/alist-org/alist/v3/pkg/utils" "io" "net/http" "net/url" @@ -419,7 +420,7 @@ func (c *Client) ReadStreamRange(path string, offset, length int64) (io.ReadClos // stream in rs.Body if rs.StatusCode == 200 { // discard first 'offset' bytes. - if _, err := io.Copy(io.Discard, io.LimitReader(rs.Body, offset)); err != nil { + if _, err := utils.CopyWithBuffer(io.Discard, io.LimitReader(rs.Body, offset)); err != nil { return nil, newPathErrorErr("ReadStreamRange", path, err) } diff --git a/pkg/utils/file.go b/pkg/utils/file.go index 7ae07158998..54247636dcb 100644 --- a/pkg/utils/file.go +++ b/pkg/utils/file.go @@ -32,7 +32,7 @@ func CopyFile(src, dst string) error { } defer dstfd.Close() - if _, err = io.Copy(dstfd, srcfd); err != nil { + if _, err = CopyWithBuffer(dstfd, srcfd); err != nil { return err } if srcinfo, err = os.Stat(src); err != nil { @@ -121,7 +121,7 @@ func CreateTempFile(r io.Reader, size int64) (*os.File, error) { if err != nil { return nil, err } - readBytes, err := io.Copy(f, r) + readBytes, err := CopyWithBuffer(f, r) if err != nil { _ = os.Remove(f.Name()) return nil, errs.NewErr(err, "CreateTempFile failed") diff --git a/pkg/utils/hash.go b/pkg/utils/hash.go index 8f8aaa26781..fa06bcc24c2 100644 --- a/pkg/utils/hash.go +++ b/pkg/utils/hash.go @@ -96,7 +96,7 @@ func HashData(hashType *HashType, data []byte, params ...any) string { // HashReader get hash of one hashType from a reader func HashReader(hashType *HashType, reader io.Reader, params ...any) (string, error) { h := hashType.NewFunc(params...) - _, err := io.Copy(h, reader) + _, err := CopyWithBuffer(h, reader) if err != nil { return "", errs.NewErr(err, "HashReader error") } diff --git a/pkg/utils/hash_test.go b/pkg/utils/hash_test.go index 55713c1afb3..0f5a2a3b14e 100644 --- a/pkg/utils/hash_test.go +++ b/pkg/utils/hash_test.go @@ -4,7 +4,6 @@ import ( "bytes" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - "io" "testing" ) @@ -36,7 +35,7 @@ var hashTestSet = []hashTest{ func TestMultiHasher(t *testing.T) { for _, test := range hashTestSet { mh := NewMultiHasher([]*HashType{MD5, SHA1, SHA256}) - n, err := io.Copy(mh, bytes.NewBuffer(test.input)) + n, err := CopyWithBuffer(mh, bytes.NewBuffer(test.input)) require.NoError(t, err) assert.Len(t, test.input, int(n)) hashInfo := mh.GetHashInfo() diff --git a/pkg/utils/io.go b/pkg/utils/io.go index 6852e28a83d..7be989c3fd7 100644 --- a/pkg/utils/io.go +++ b/pkg/utils/io.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "io" + "sync" "time" "golang.org/x/exp/constraints" @@ -29,7 +30,7 @@ func CopyWithCtx(ctx context.Context, out io.Writer, in io.Reader, size int64, p // possible in the call process. var finish int64 = 0 s := size / 100 - _, err := io.Copy(out, readerFunc(func(p []byte) (int, error) { + _, err := CopyWithBuffer(out, readerFunc(func(p []byte) (int, error) { // golang non-blocking channel: https://gobyexample.com/non-blocking-channel-operations select { // if context has been canceled @@ -204,3 +205,31 @@ func Max[T constraints.Ordered](a, b T) T { } return a } + +var IoBuffPool = &sync.Pool{ + New: func() interface{} { + return make([]byte, 32*1024*2) // Two times of size in io package + }, +} + +func CopyWithBuffer(dst io.Writer, src io.Reader) (written int64, err error) { + buff := IoBuffPool.Get().([]byte) + defer IoBuffPool.Put(buff) + written, err = io.CopyBuffer(dst, src, buff) + if err != nil { + return + } + return written, nil +} + +func CopyWithBufferN(dst io.Writer, src io.Reader, n int64) (written int64, err error) { + written, err = CopyWithBuffer(dst, io.LimitReader(src, n)) + if written == n { + return n, nil + } + if written < n && err == nil { + // src stopped early; must have been EOF. + err = io.EOF + } + return +}