Transfer Manager v2 downloader concurrency fix and version control (#3058)

* migrate parts GET to concurrent reader

Author: wty-Bryant

feature/s3/transfermanager/api_op_DownloadObject.go

@@ -541,19 +541,22 @@ type downloader struct {
 in *DownloadObjectInput
 out *DownloadObjectOutput
 
-wg sync.WaitGroup
-m sync.Mutex
+wg sync.WaitGroup
+m sync.Mutex
+etagOnce sync.Once
+totalBytesOnce sync.Once
 
 offset int64
 pos int64
 totalBytes int64
 written int64
+etag string
 
 err error
 }
 
 func (d *downloader) download(ctx context.Context) (*DownloadObjectOutput, error) {
-if err := d.init(ctx); err != nil {
+if err := d.init(); err != nil {
 return nil, fmt.Errorf("unable to initialize download: %w", err)
 }
 
@@ -600,12 +603,12 @@ func (d *downloader) download(ctx context.Context) (*DownloadObjectOutput, error
 d.wg.Wait()
 }
 } else {
-if d.in.Range == "" {
-output = d.getChunk(ctx, 0, d.byteRange(), clientOptions...)
-} else {
+if d.in.Range != "" {
 d.pos, d.totalBytes = d.getDownloadRange()
 d.offset = d.pos
 }
+
+d.getChunk(ctx, 0, d.byteRange(), clientOptions...)
 total := d.totalBytes
 
 ch := make(chan dlChunk, d.options.Concurrency)
@@ -639,7 +642,7 @@ func (d *downloader) download(ctx context.Context) (*DownloadObjectOutput, error
 return d.out, nil
 }
 
-func (d *downloader) init(ctx context.Context) error {
+func (d *downloader) init() error {
 if d.options.PartSizeBytes < minPartSizeBytes {
 return fmt.Errorf("part size must be at least %d bytes", minPartSizeBytes)
 }
@@ -655,7 +658,6 @@ func (d *downloader) init(ctx context.Context) error {
 
 func (d *downloader) singleDownload(ctx context.Context, clientOptions ...func(*s3.Options)) (*DownloadObjectOutput, error) {
 chunk := dlChunk{w: d.in.WriterAt}
-// d.in.PartNumber = 0
 output, err := d.downloadChunk(ctx, chunk, clientOptions...)
 if err != nil {
 return nil, err
@@ -708,6 +710,9 @@ func (d *downloader) downloadChunk(ctx context.Context, chunk dlChunk, clientOpt
 if chunk.withRange != "" {
 params.Range = aws.String(chunk.withRange)
 }
+if params.VersionId == nil && d.etag != "" {
+params.IfMatch = aws.String(d.etag)
+}
 
 var out *s3.GetObjectOutput
 var n int64
@@ -737,6 +742,9 @@ func (d *downloader) downloadChunk(ctx context.Context, chunk dlChunk, clientOpt
 if out != nil {
 output = &DownloadObjectOutput{}
 output.mapFromGetObjectOutput(out, params.ChecksumMode)
+d.etagOnce.Do(func() {
+d.etag = aws.ToString(out.ETag)
+})
 }
 return output, err
 }
@@ -747,7 +755,9 @@ func (d *downloader) tryDownloadChunk(ctx context.Context, params *s3.GetObjectI
 return nil, 0, err
 }
 
-d.setTotalBytes(out) // Set total if not yet set.
+d.totalBytesOnce.Do(func() {
+d.setTotalBytes(out)
+}) // Set total in first GET
 
 var n int64
 defer out.Body.Close()
@@ -780,9 +790,6 @@ func (d *downloader) getTotalBytes() int64 {
 // does not include a Content-Range. Meaning the object was not chunked. This
 // occurs when the full file fits within the PartSize directive.
 func (d *downloader) setTotalBytes(resp *s3.GetObjectOutput) {
-d.m.Lock()
-defer d.m.Unlock()
-
 if d.totalBytes >= 0 {
 return
 }

feature/s3/transfermanager/api_op_DownloadObject_integ_test.go

@@ -11,7 +11,7 @@ import (
 )
 
 func TestInteg_DownloadObject(t *testing.T) {
-cases := map[string]getObjectTestData{
+cases := map[string]downloadObjectTestData{
 "part get seekable body": {Body: strings.NewReader("hello world"), ExpectBody: []byte("hello world")},
 "part get empty string body": {Body: strings.NewReader(""), ExpectBody: []byte("")},
 "part get multipart body": {Body: bytes.NewReader(largeObjectBuf), ExpectBody: largeObjectBuf},