replace int fields manual lock with atomic lock and remove extra clone

Author: wty-Bryant

feature/s3/transfermanager/api_op_DownloadDirectory.go

@@ -10,6 +10,7 @@ import (
 	"path/filepath"
 	"strings"
 	"sync"
+	"sync/atomic"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
@@ -87,12 +88,12 @@ type DownloadDirectoryOutput struct {
 	// Total number of objects successfully downloaded
 	// this value might not be the real number of success if user passed a customized
 	// failure policy in input
-	ObjectsDownloaded int
+	ObjectsDownloaded int64
 
 	// Total number of objects failed to download
 	// this value might not be the real number of failure if user passed a customized
 	// failure policy in input
-	ObjectsFailed int
+	ObjectsFailed int64
 }
 
 type objectEntry struct {
@@ -132,8 +133,8 @@ type directoryDownloader struct {
 	in            *DownloadDirectoryInput
 	failurePolicy DownloadDirectoryFailurePolicy
 
-	objectsDownloaded int
-	objectsFailed     int
+	objectsDownloaded int64
+	objectsFailed     int64
 
 	err error
 
@@ -272,7 +273,7 @@ func (d *directoryDownloader) downloadObject(ctx context.Context, ch chan object
 			if err != nil {
 				d.setErr(fmt.Errorf("error when heading info of object %s: %v", data.key, err))
 			} else {
-				d.incrObjectsFailed(1)
+				atomic.AddInt64(&d.objectsFailed, 1)
 			}
 			continue
 		}
@@ -298,31 +299,17 @@ func (d *directoryDownloader) downloadObject(ctx context.Context, ch chan object
 			if err != nil {
 				d.setErr(fmt.Errorf("error when getting object and writing to local file %s: %v", data.path, err))
 			} else {
-				d.incrObjectsFailed(1)
+				atomic.AddInt64(&d.objectsFailed, 1)
 			}
 			os.Remove(data.path)
 			continue
 		}
 
-		d.incrObjectsDownloaded(1)
+		atomic.AddInt64(&d.objectsDownloaded, 1)
 		d.emitter.ObjectsTransferred(ctx, n)
 	}
 }
 
-func (d *directoryDownloader) incrObjectsDownloaded(n int) {
-	d.mu.Lock()
-	defer d.mu.Unlock()
-
-	d.objectsDownloaded += n
-}
-
-func (d *directoryDownloader) incrObjectsFailed(n int) {
-	d.mu.Lock()
-	defer d.mu.Unlock()
-
-	d.objectsFailed += n
-}
-
 func (d *directoryDownloader) setErr(err error) {
 	d.mu.Lock()
 	defer d.mu.Unlock()

feature/s3/transfermanager/api_op_DownloadObject.go

@@ -10,6 +10,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
@@ -518,7 +519,7 @@ func (o *DownloadObjectOutput) mapFromGetObjectOutput(out *s3.GetObjectOutput, c
 	o.TagCount = out.TagCount
 	o.VersionID = out.VersionId
 	o.WebsiteRedirectLocation = out.WebsiteRedirectLocation
-	o.ResultMetadata = out.ResultMetadata.Clone()
+	o.ResultMetadata = out.ResultMetadata
 }
 
 // DownloadObject downloads an object from S3, intelligently splitting large
@@ -793,7 +794,7 @@ func (d *downloader) tryDownloadChunk(ctx context.Context, params *s3.GetObjectI
 		return nil, &errReadingBody{err: err}
 	}
 
-	d.incrWritten(n)
+	atomic.AddInt64(&d.written, n)
 	d.emitter.BytesTransferred(ctx, n)
 	return out, nil
 }

feature/s3/transfermanager/api_op_GetObject.go

@@ -6,6 +6,7 @@ import (
 	"io"
 	"strconv"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
@@ -531,7 +532,7 @@ func (o *GetObjectOutput) mapFromGetObjectOutput(out *s3.GetObjectOutput, checks
 	o.TagCount = out.TagCount
 	o.VersionID = out.VersionId
 	o.WebsiteRedirectLocation = out.WebsiteRedirectLocation
-	o.ResultMetadata = out.ResultMetadata.Clone()
+	o.ResultMetadata = out.ResultMetadata
 }
 
 func (o *GetObjectOutput) mapFromHeadObjectOutput(out *s3.HeadObjectOutput, checksumMode types.ChecksumMode, enableChecksumValidation bool, body *concurrentReader) {
@@ -579,7 +580,7 @@ func (o *GetObjectOutput) mapFromHeadObjectOutput(out *s3.HeadObjectOutput, chec
 	o.TagCount = out.TagCount
 	o.VersionID = out.VersionId
 	o.WebsiteRedirectLocation = out.WebsiteRedirectLocation
-	o.ResultMetadata = out.ResultMetadata.Clone()
+	o.ResultMetadata = out.ResultMetadata
 }
 
 // GetObject downloads an object from S3, intelligently splitting large
@@ -648,7 +649,7 @@ func (g *getter) get(ctx context.Context) (out *GetObjectOutput, err error) {
 		capacity := sectionParts
 		r.sectionParts = sectionParts
 		r.partSize = partSize
-		r.setCapacity(min(capacity, partsCount))
+		atomic.StoreInt32(&r.capacity, min(capacity, partsCount))
 		r.partsCount = partsCount
 	} else {
 		out, err := g.options.S3.HeadObject(ctx, &s3.HeadObjectInput{
@@ -672,7 +673,7 @@ func (g *getter) get(ctx context.Context) (out *GetObjectOutput, err error) {
 		sectionParts := int32(max(1, g.options.GetBufferSize/g.options.PartSizeBytes))
 		capacity := min(sectionParts, partsCount)
 		r.partSize = g.options.PartSizeBytes
-		r.setCapacity(capacity)
+		atomic.StoreInt32(&r.capacity, capacity)
 		r.partsCount = partsCount
 		r.sectionParts = sectionParts
 		r.totalBytes = total

feature/s3/transfermanager/api_op_UploadDirectory.go

@@ -6,6 +6,7 @@ import (
 	"os"
 	"path/filepath"
 	"sync"
+	"sync/atomic"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
 )
@@ -89,12 +90,12 @@ type UploadDirectoryOutput struct {
 	// Total number of objects successfully uploaded
 	// this value might not be the real number of success if user passed a customized
 	// failure policy in input
-	ObjectsUploaded int
+	ObjectsUploaded int64
 
 	// Total number of objects failed to upload
 	// this value might not be the real number of failure is user passed a customized
 	// failure policy in input
-	ObjectsFailed int
+	ObjectsFailed int64
 }
 
 // UploadDirectory traverses a local directory recursively/non-recursively and intelligently
@@ -127,8 +128,8 @@ type directoryUploader struct {
 	in            *UploadDirectoryInput
 	failurePolicy UploadDirectoryFailurePolicy
 
-	filesUploaded int
-	filesFailed   int
+	filesUploaded int64
+	filesFailed   int64
 	traversed     map[string]interface{}
 
 	err error
@@ -378,33 +379,19 @@ func (u *directoryUploader) uploadFile(ctx context.Context, ch chan fileEntry) {
 				u.setErr(fmt.Errorf("error when uploading file %s: %v", data.path, err))
 			} else {
 				// this failed object is ignored, just increase the failure count
-				u.incrFilesFailed(1)
+				atomic.AddInt64(&u.filesFailed, 1)
 			}
 			continue
 		}
 
 		u.progressOnce.Do(func() {
 			u.emitter.Start(ctx, u.in)
 		})
-		u.incrFilesUploaded(1)
+		atomic.AddInt64(&u.filesUploaded, 1)
 		u.emitter.ObjectsTransferred(ctx, aws.ToInt64(out.ContentLength))
 	}
 }
 
-func (u *directoryUploader) incrFilesUploaded(n int) {
-	u.mu.Lock()
-	defer u.mu.Unlock()
-
-	u.filesUploaded += n
-}
-
-func (u *directoryUploader) incrFilesFailed(n int) {
-	u.mu.Lock()
-	defer u.mu.Unlock()
-
-	u.filesFailed += n
-}
-
 func (u *directoryUploader) setErr(err error) {
 	u.mu.Lock()
 	defer u.mu.Unlock()

feature/s3/transfermanager/api_op_UploadObject.go

@@ -738,7 +738,7 @@ func (o *UploadObjectOutput) mapFromPutObjectOutput(out *s3.PutObjectOutput, buc
 	o.SSEKMSKeyID = out.SSEKMSKeyId
 	o.ServerSideEncryption = types.ServerSideEncryption(out.ServerSideEncryption)
 	o.VersionID = out.VersionId
-	o.ResultMetadata = out.ResultMetadata.Clone()
+	o.ResultMetadata = out.ResultMetadata
 }
 
 func (o *UploadObjectOutput) mapFromCompleteMultipartUploadOutput(out *s3.CompleteMultipartUploadOutput, bucket, uploadID *string, contentLength int64, completedParts completedParts) {
@@ -761,7 +761,7 @@ func (o *UploadObjectOutput) mapFromCompleteMultipartUploadOutput(out *s3.Comple
 	o.SSEKMSKeyID = out.SSEKMSKeyId
 	o.ServerSideEncryption = types.ServerSideEncryption(out.ServerSideEncryption)
 	o.VersionID = out.VersionId
-	o.ResultMetadata = out.ResultMetadata.Clone()
+	o.ResultMetadata = out.ResultMetadata
 }
 
 // UploadObject uploads an object to S3, intelligently buffering large

feature/s3/transfermanager/concurrent_reader.go

@@ -10,6 +10,7 @@ import (
 	"github.com/aws/aws-sdk-go-v2/service/s3"
 	"io"
 	"sync"
+	"sync/atomic"
 )
 
 // concurrentReader receives object parts from working goroutines, composes those chunks in order and read
@@ -81,7 +82,7 @@ func (r *concurrentReader) Read(p []byte) (int, error) {
 			break
 		}
 
-		if r.index == r.getCapacity() {
+		if r.index == atomic.LoadInt32(&r.capacity) {
 			continue
 		}
 
@@ -202,7 +203,7 @@ func (r *concurrentReader) read(p []byte) (int, error) {
 
 	partSize := r.partSize
 	minIndex := int32(r.written / partSize)
-	maxIndex := min(int32((r.written+int64(cap(p))-1)/partSize), r.getCapacity()-1)
+	maxIndex := min(int32((r.written+int64(cap(p))-1)/partSize), atomic.LoadInt32(&r.capacity)-1)
 	for i := minIndex; i <= maxIndex; i++ {
 		if e := r.getErr(); e != nil && e != io.EOF {
 			r.clean()
@@ -223,9 +224,9 @@ func (r *concurrentReader) read(p []byte) (int, error) {
 			if c.cur >= c.length {
 				r.readCount++
 				delete(r.buf, i)
-				if r.readCount == r.getCapacity() {
-					capacity := min(r.getCapacity()+r.sectionParts, r.partsCount)
-					r.setCapacity(capacity)
+				if r.readCount == atomic.LoadInt32(&r.capacity) {
+					capacity := min(atomic.LoadInt32(&r.capacity)+r.sectionParts, r.partsCount)
+					atomic.StoreInt32(&r.capacity, capacity)
 				}
 				if r.readCount >= r.partsCount {
 					r.setErr(io.EOF)
@@ -234,7 +235,7 @@ func (r *concurrentReader) read(p []byte) (int, error) {
 		}
 	}
 
-	for r.receiveCount < r.getCapacity() {
+	for r.receiveCount < atomic.LoadInt32(&r.capacity) {
 		if e := r.getErr(); e != nil && e != io.EOF {
 			r.clean()
 			return written, e
@@ -263,9 +264,9 @@ func (r *concurrentReader) read(p []byte) (int, error) {
 			r.buf[oc.index] = &oc
 		} else {
 			r.readCount++
-			if r.readCount == r.getCapacity() {
-				capacity := min(r.getCapacity()+r.sectionParts, r.partsCount)
-				r.setCapacity(capacity)
+			if r.readCount == atomic.LoadInt32(&r.capacity) {
+				capacity := min(atomic.LoadInt32(&r.capacity)+r.sectionParts, r.partsCount)
+				atomic.StoreInt32(&r.capacity, capacity)
 			}
 			if r.readCount >= r.partsCount {
 				r.setErr(io.EOF)
@@ -276,20 +277,6 @@ func (r *concurrentReader) read(p []byte) (int, error) {
 	return written, r.getErr()
 }
 
-func (r *concurrentReader) setCapacity(n int32) {
-	r.m.Lock()
-	defer r.m.Unlock()
-
-	r.capacity = n
-}
-
-func (r *concurrentReader) getCapacity() int32 {
-	r.m.Lock()
-	defer r.m.Unlock()
-
-	return r.capacity
-}
-
 func (r *concurrentReader) setDone(done bool) {
 	r.m.Lock()
 	defer r.m.Unlock()

feature/s3/transfermanager/download_directory_test.go

@@ -59,8 +59,8 @@ func TestDownloadDirectory(t *testing.T) {
 		expectKeys              []string
 		expectFiles             []string
 		expectErr               string
-		expectObjectsDownloaded int
-		expectObjectsFailed     int
+		expectObjectsDownloaded int64
+		expectObjectsFailed     int64
 		listenerValidationFn    func(*testing.T, *mockDirectoryListener, any, any, error)
 	}{
 		"single object": {

feature/s3/transfermanager/mapping_reference_test.go

@@ -689,7 +689,6 @@ func mockValue(v reflect.Value, field string, legacyTypes map[string]interface{}
 			if !ok {
 				panic(fmt.Sprintf("need to handle %v for field %s", v.Type().Elem().Kind(), field))
 			}
-			//indirV := reflect.Indirect(reflect.ValueOf(&vv))
 			v.Set(reflect.ValueOf(vv))
 		default:
 			panic(fmt.Sprintf("need to handle %v", v.Type().Elem().Kind()))

feature/s3/transfermanager/setup_integ_test.go

@@ -389,7 +389,7 @@ type uploadDirectoryTestData struct {
 	Source              string
 	Recursive           bool
 	KeyPrefix           string
-	ExpectFilesUploaded int
+	ExpectFilesUploaded int64
 	ExpectKeys          []string
 	ExpectError         string
 }
@@ -478,7 +478,7 @@ func testUploadDirectory(t *testing.T, bucket string, testData uploadDirectoryTe
 type downloadDirectoryTestData struct {
 	ObjectsSize             map[string]int64
 	KeyPrefix               string
-	ExpectObjectsDownloaded int
+	ExpectObjectsDownloaded int64
 	ExpectFiles             []string
 	ExpectError             string
 }

feature/s3/transfermanager/upload_directory_test.go

@@ -49,15 +49,14 @@ func TestUploadDirectory(t *testing.T) {
 		recursive            bool
 		keyPrefix            string
 		filter               FileFilter
-		s3Delimiter          string
 		callback             PutRequestCallback
 		failurePolicy        UploadDirectoryFailurePolicy
 		putobjectFunc        func(*s3testing.TransferManagerLoggingClient, *s3.PutObjectInput) (*s3.PutObjectOutput, error)
 		preprocessFunc       func(string) (func() error, error)
 		expectKeys           []string
 		expectErr            string
-		expectFilesUploaded  int
-		expectFilesFailed    int
+		expectFilesUploaded  int64
+		expectFilesFailed    int64
 		listenerValidationFn func(*testing.T, *mockDirectoryListener, any, any, error)
 	}{
 		"single file recursively": {