fix sca again

Author: gouhongshen

pkg/fileservice/parallel_sdk_test.go

@@ -17,7 +17,6 @@ package fileservice
 import (
 	"bytes"
 	"context"
-	"encoding/xml"
 	"fmt"
 	"io"
 	"net/http"
@@ -295,11 +294,6 @@ func TestAwsMultipartCreateFail(t *testing.T) {
 	}
 }
 
-type qcloudResult struct {
-	XMLName  xml.Name `xml:"InitiateMultipartUploadResult"`
-	UploadID string   `xml:"UploadId"`
-}
-
 func newMockCOSServer(t *testing.T, failPart int) (*httptest.Server, *cosServerState) {
 	t.Helper()
 	state := &cosServerState{

pkg/fileservice/parallel_upload_test.go

@@ -69,7 +69,6 @@ func TestNormalizeParallelOption(t *testing.T) {
 	}
 
 	for _, c := range cases {
-		c := c
 		t.Run(c.name, func(t *testing.T) {
 			res := normalizeParallelOption(c.opt)
 			if res.PartSize != c.expectSize {

pkg/fileservice/qcloud_sdk.go

@@ -312,29 +312,31 @@ func (a *QCloudSDK) WriteMultipartParallel(
 
 	bufPool := sync.Pool{
 		New: func() any {
-			return make([]byte, options.PartSize)
+			buf := make([]byte, options.PartSize)
+			return &buf
 		},
 	}
 
-	readChunk := func() (buf []byte, n int, err error) {
-		raw := bufPool.Get().([]byte)
+	readChunk := func() (bufPtr *[]byte, buf []byte, n int, err error) {
+		bufPtr = bufPool.Get().(*[]byte)
+		raw := *bufPtr
 		n, err = io.ReadFull(r, raw)
 		switch {
 		case errors.Is(err, io.EOF):
-			bufPool.Put(raw)
-			return nil, 0, io.EOF
+			bufPool.Put(bufPtr)
+			return nil, nil, 0, io.EOF
 		case errors.Is(err, io.ErrUnexpectedEOF):
 			err = io.EOF
-			return raw, n, err
+			return bufPtr, raw, n, err
 		case err != nil:
-			bufPool.Put(raw)
-			return nil, 0, err
+			bufPool.Put(bufPtr)
+			return nil, nil, 0, err
 		default:
-			return raw, n, nil
+			return bufPtr, raw, n, nil
 		}
 	}
 
-	firstBuf, firstN, err := readChunk()
+	firstBufPtr, firstBuf, firstN, err := readChunk()
 	if err != nil && !errors.Is(err, io.EOF) {
 		return err
 	}
@@ -344,7 +346,7 @@ func (a *QCloudSDK) WriteMultipartParallel(
 	if errors.Is(err, io.EOF) && int64(firstN) < minMultipartPartSize {
 		data := make([]byte, firstN)
 		copy(data, firstBuf[:firstN])
-		bufPool.Put(firstBuf)
+		bufPool.Put(firstBufPtr)
 		size := int64(firstN)
 		return a.Write(ctx, key, bytes.NewReader(data), &size, options.Expire)
 	}
@@ -364,7 +366,7 @@ func (a *QCloudSDK) WriteMultipartParallel(
 		return res, e
 	}, maxRetryAttemps, IsRetryableError)
 	if createErr != nil {
-		bufPool.Put(firstBuf)
+		bufPool.Put(firstBufPtr)
 		return createErr
 	}
 
@@ -375,9 +377,10 @@ func (a *QCloudSDK) WriteMultipartParallel(
 	}()
 
 	type partJob struct {
-		num int32
-		buf []byte
-		n   int
+		num    int32
+		buf    []byte
+		bufPtr *[]byte
+		n      int
 	}
 
 	var (
@@ -407,7 +410,9 @@ func (a *QCloudSDK) WriteMultipartParallel(
 			defer wg.Done()
 			for job := range jobCh {
 				if ctx.Err() != nil {
-					bufPool.Put(job.buf)
+					if job.bufPtr != nil {
+						bufPool.Put(job.bufPtr)
+					}
 					continue
 				}
 				uploadOpt := &cos.ObjectUploadPartOptions{
@@ -418,14 +423,18 @@ func (a *QCloudSDK) WriteMultipartParallel(
 				}, maxRetryAttemps, IsRetryableError)
 				if uploadErr != nil {
 					setErr(uploadErr)
-					bufPool.Put(job.buf)
+					if job.bufPtr != nil {
+						bufPool.Put(job.bufPtr)
+					}
 					continue
 				}
 				etag := ""
 				if resp != nil && resp.Header != nil {
 					etag = resp.Header.Get("ETag")
 				}
-				bufPool.Put(job.buf)
+				if job.bufPtr != nil {
+					bufPool.Put(job.bufPtr)
+				}
 				partsLock.Lock()
 				parts = append(parts, cos.Object{
 					PartNumber: int(job.num),
@@ -443,29 +452,34 @@ func (a *QCloudSDK) WriteMultipartParallel(
 		}
 	}
 
-	sendJob := func(buf []byte, n int) bool {
+	sendJob := func(bufPtr *[]byte, buf []byte, n int) bool {
 		partNum++
 		if partNum > maxMultipartParts {
 			setErr(moerr.NewInternalErrorNoCtxf("too many parts for multipart upload: %d", partNum))
-			bufPool.Put(buf)
+			if bufPtr != nil {
+				bufPool.Put(bufPtr)
+			}
 			return false
 		}
 		job := partJob{
-			num: partNum,
-			buf: buf,
-			n:   n,
+			num:    partNum,
+			buf:    buf,
+			bufPtr: bufPtr,
+			n:      n,
 		}
 		select {
 		case jobCh <- job:
 			return true
 		case <-ctx.Done():
-			bufPool.Put(buf)
+			if bufPtr != nil {
+				bufPool.Put(bufPtr)
+			}
 			setErr(ctx.Err())
 			return false
 		}
 	}
 
-	if !sendJob(firstBuf, firstN) {
+	if !sendJob(firstBufPtr, firstBuf, firstN) {
 		close(jobCh)
 		wg.Wait()
 		if firstErr != nil {
@@ -475,24 +489,24 @@ func (a *QCloudSDK) WriteMultipartParallel(
 	}
 
 	for {
-		nextBuf, nextN, readErr := readChunk()
+		nextBufPtr, nextBuf, nextN, readErr := readChunk()
 		if errors.Is(readErr, io.EOF) && nextN == 0 {
 			break
 		}
 		if readErr != nil && !errors.Is(readErr, io.EOF) {
 			setErr(readErr)
-			if nextBuf != nil {
-				bufPool.Put(nextBuf)
+			if nextBufPtr != nil {
+				bufPool.Put(nextBufPtr)
 			}
 			break
 		}
 		if nextN == 0 {
-			if nextBuf != nil {
-				bufPool.Put(nextBuf)
+			if nextBufPtr != nil {
+				bufPool.Put(nextBufPtr)
 			}
 			break
 		}
-		if !sendJob(nextBuf, nextN) {
+		if !sendJob(nextBufPtr, nextBuf, nextN) {
 			break
 		}
 		if readErr != nil && errors.Is(readErr, io.EOF) {