service/s3/s3manager: Move part buffer pool upwards to allow reuse.

Author: skmcgrail

service/s3/s3manager/upload.go

@@ -165,6 +165,9 @@ type Uploader struct {
 
 	// Defines the buffer strategy used when uploading a part
 	BufferProvider ReadSeekerWriteToProvider
+
+	// partPool allows for the re-usage of streaming payload part buffers between upload calls
+	partPool *partPool
 }
 
 // NewUploader creates a new Uploader instance to upload objects to S3. Pass In
@@ -201,6 +204,8 @@ func newUploader(client s3iface.S3API, options ...func(*Uploader)) *Uploader {
 		option(u)
 	}
 
+	u.partPool = newPartPool(u.PartSize)
+
 	return u
 }
 
@@ -283,6 +288,13 @@ func (u Uploader) UploadWithContext(ctx aws.Context, input *UploadInput, opts ..
 	for _, opt := range opts {
 		opt(&i.cfg)
 	}
+
+	// If PartSize was changed or partPool was never setup then we need to allocated a new pool
+	// so that we return []byte slices of the correct size
+	if i.cfg.partPool == nil || i.cfg.partPool.partSize != i.cfg.PartSize {
+		i.cfg.partPool = newPartPool(i.cfg.PartSize)
+	}
+
 	i.cfg.RequestOptions = append(i.cfg.RequestOptions, request.WithAppendUserAgent("S3Manager"))
 
 	return i.upload()
@@ -352,8 +364,6 @@ type uploader struct {
 
 	readerPos int64 // current reader position
 	totalSize int64 // set to -1 if the size is not known
-
-	bufferPool sync.Pool
 }
 
 // internal logic for deciding whether to upload a single part or use a
@@ -393,10 +403,6 @@ func (u *uploader) init() error {
 		u.cfg.MaxUploadParts = MaxUploadParts
 	}
 
-	u.bufferPool = sync.Pool{
-		New: func() interface{} { return make([]byte, u.cfg.PartSize) },
-	}
-
 	// Try to get the total size for some optimizations
 	return u.initSize()
 }
@@ -466,12 +472,12 @@ func (u *uploader) nextReader() (io.ReadSeeker, int, func(), error) {
 		return reader, int(n), cleanup, err
 
 	default:
-		part := u.bufferPool.Get().([]byte)
+		part := u.cfg.partPool.Get().([]byte)
 		n, err := readFillBuf(r, part)
 		u.readerPos += int64(n)
 
 		cleanup := func() {
-			u.bufferPool.Put(part)
+			u.cfg.partPool.Put(part)
 		}
 
 		return bytes.NewReader(part[0:n]), n, cleanup, err
@@ -751,3 +757,18 @@ func (u *multiuploader) complete() *s3.CompleteMultipartUploadOutput {
 
 	return resp
 }
+
+type partPool struct {
+	partSize int64
+	sync.Pool
+}
+
+func newPartPool(partSize int64) *partPool {
+	p := &partPool{partSize: partSize}
+
+	p.New = func() interface{} {
+		return make([]byte, p.partSize)
+	}
+
+	return p
+}