vendor: update all dependencies

2019-11-14 21:51:34 +00:00
parent 3dcf1e61cf
commit 479c803fd9
446 changed files with 25052 additions and 8062 deletions
@@ -0,0 +1,81 @@
+package s3manager
+
+import (
+	"io"
+
+	"github.com/aws/aws-sdk-go/internal/sdkio"
+)
+
+// BufferedReadSeeker is buffered io.ReadSeeker
+type BufferedReadSeeker struct {
+	r                 io.ReadSeeker
+	buffer            []byte
+	readIdx, writeIdx int
+}
+
+// NewBufferedReadSeeker returns a new BufferedReadSeeker
+// if len(b) == 0 then the buffer will be initialized to 64 KiB.
+func NewBufferedReadSeeker(r io.ReadSeeker, b []byte) *BufferedReadSeeker {
+	if len(b) == 0 {
+		b = make([]byte, 64*1024)
+	}
+	return &BufferedReadSeeker{r: r, buffer: b}
+}
+
+func (b *BufferedReadSeeker) reset(r io.ReadSeeker) {
+	b.r = r
+	b.readIdx, b.writeIdx = 0, 0
+}
+
+// Read will read up len(p) bytes into p and will return
+// the number of bytes read and any error that occurred.
+// If the len(p) > the buffer size then a single read request
+// will be issued to the underlying io.ReadSeeker for len(p) bytes.
+// A Read request will at most perform a single Read to the underlying
+// io.ReadSeeker, and may return < len(p) if serviced from the buffer.
+func (b *BufferedReadSeeker) Read(p []byte) (n int, err error) {
+	if len(p) == 0 {
+		return n, err
+	}
+
+	if b.readIdx == b.writeIdx {
+		if len(p) >= len(b.buffer) {
+			n, err = b.r.Read(p)
+			return n, err
+		}
+		b.readIdx, b.writeIdx = 0, 0
+
+		n, err = b.r.Read(b.buffer)
+		if n == 0 {
+			return n, err
+		}
+
+		b.writeIdx += n
+	}
+
+	n = copy(p, b.buffer[b.readIdx:b.writeIdx])
+	b.readIdx += n
+
+	return n, err
+}
+
+// Seek will position then underlying io.ReadSeeker to the given offset
+// and will clear the buffer.
+func (b *BufferedReadSeeker) Seek(offset int64, whence int) (int64, error) {
+	n, err := b.r.Seek(offset, whence)
+
+	b.reset(b.r)
+
+	return n, err
+}
+
+// ReadAt will read up to len(p) bytes at the given file offset.
+// This will result in the buffer being cleared.
+func (b *BufferedReadSeeker) ReadAt(p []byte, off int64) (int, error) {
+	_, err := b.Seek(off, sdkio.SeekStart)
+	if err != nil {
+		return 0, err
+	}
+
+	return b.Read(p)
+}
@@ -0,0 +1,7 @@
+// +build !windows
+
+package s3manager
+
+func defaultUploadBufferProvider() ReadSeekerWriteToProvider {
+	return nil
+}
@@ -0,0 +1,5 @@
+package s3manager
+
+func defaultUploadBufferProvider() ReadSeekerWriteToProvider {
+	return NewBufferedReadSeekerWriteToPool(1024 * 1024)
+}
@@ -0,0 +1,7 @@
+// +build !windows
+
+package s3manager
+
+func defaultDownloadBufferProvider() WriterReadFromProvider {
+	return nil
+}
@@ -0,0 +1,5 @@
+package s3manager
+
+func defaultDownloadBufferProvider() WriterReadFromProvider {
+	return NewPooledBufferedWriterReadFromProvider(1024 * 1024)
+}
@@ -25,13 +25,25 @@ const DefaultDownloadPartSize = 1024 * 1024 * 5
 // when using Download().
 const DefaultDownloadConcurrency = 5

+type errReadingBody struct {
+	err error
+}
+
+func (e *errReadingBody) Error() string {
+	return fmt.Sprintf("failed to read part body: %v", e.err)
+}
+
+func (e *errReadingBody) Unwrap() error {
+	return e.err
+}
+
 // The Downloader structure that calls Download(). It is safe to call Download()
 // on this structure for multiple objects and across concurrent goroutines.
 // Mutating the Downloader's properties is not safe to be done concurrently.
 type Downloader struct {
-	// The buffer size (in bytes) to use when buffering data into chunks and
-	// sending them as parts to S3. The minimum allowed part size is 5MB, and
-	// if this value is set to zero, the DefaultDownloadPartSize value will be used.
+	// The size (in bytes) to request from S3 for each part.
+	// The minimum allowed part size is 5MB, and  if this value is set to zero,
+	// the DefaultDownloadPartSize value will be used.
 	//
 	// PartSize is ignored if the Range input parameter is provided.
 	PartSize int64
@@ -50,6 +62,14 @@ type Downloader struct {
 	// List of request options that will be passed down to individual API
 	// operation requests made by the downloader.
 	RequestOptions []request.Option
+
+	// Defines the buffer strategy used when downloading a part.
+	//
+	// If a WriterReadFromProvider is given the Download manager
+	// will pass the io.WriterAt of the Download request to the provider
+	// and will use the returned WriterReadFrom from the provider as the
+	// destination writer when copying from http response body.
+	BufferProvider WriterReadFromProvider
 }

 // WithDownloaderRequestOptions appends to the Downloader's API request options.
@@ -77,10 +97,15 @@ func WithDownloaderRequestOptions(opts ...request.Option) func(*Downloader) {
 //          d.PartSize = 64 * 1024 * 1024 // 64MB per part
 //     })
 func NewDownloader(c client.ConfigProvider, options ...func(*Downloader)) *Downloader {
+	return newDownloader(s3.New(c), options...)
+}
+
+func newDownloader(client s3iface.S3API, options ...func(*Downloader)) *Downloader {
 	d := &Downloader{
-		S3:          s3.New(c),
-		PartSize:    DefaultDownloadPartSize,
-		Concurrency: DefaultDownloadConcurrency,
+		S3:             client,
+		PartSize:       DefaultDownloadPartSize,
+		Concurrency:    DefaultDownloadConcurrency,
+		BufferProvider: defaultDownloadBufferProvider(),
 	}
 	for _, option := range options {
 		option(d)
@@ -109,16 +134,7 @@ func NewDownloader(c client.ConfigProvider, options ...func(*Downloader)) *Downl
 //          d.PartSize = 64 * 1024 * 1024 // 64MB per part
 //     })
 func NewDownloaderWithClient(svc s3iface.S3API, options ...func(*Downloader)) *Downloader {
-	d := &Downloader{
-		S3:          svc,
-		PartSize:    DefaultDownloadPartSize,
-		Concurrency: DefaultDownloadConcurrency,
-	}
-	for _, option := range options {
-		option(d)
-	}
-
-	return d
+	return newDownloader(svc, options...)
 }

 type maxRetrier interface {
@@ -405,18 +421,20 @@ func (d *downloader) downloadChunk(chunk dlchunk) error {
 	var n int64
 	var err error
 	for retry := 0; retry <= d.partBodyMaxRetries; retry++ {
-		var resp *s3.GetObjectOutput
-		resp, err = d.cfg.S3.GetObjectWithContext(d.ctx, in, d.cfg.RequestOptions...)
-		if err != nil {
-			return err
-		}
-		d.setTotalBytes(resp) // Set total if not yet set.
-
-		n, err = io.Copy(&chunk, resp.Body)
-		resp.Body.Close()
+		n, err = d.tryDownloadChunk(in, &chunk)
 		if err == nil {
 			break
 		}
+		// Check if the returned error is an errReadingBody.
+		// If err is errReadingBody this indicates that an error
+		// occurred while copying the http response body.
+		// If this occurs we unwrap the err to set the underlying error
+		// and attempt any remaining retries.
+		if bodyErr, ok := err.(*errReadingBody); ok {
+			err = bodyErr.Unwrap()
+		} else {
+			return err
+		}

 		chunk.cur = 0
 		logMessage(d.cfg.S3, aws.LogDebugWithRequestRetries,
@@ -429,6 +447,28 @@ func (d *downloader) downloadChunk(chunk dlchunk) error {
 	return err
 }

+func (d *downloader) tryDownloadChunk(in *s3.GetObjectInput, w io.Writer) (int64, error) {
+	cleanup := func() {}
+	if d.cfg.BufferProvider != nil {
+		w, cleanup = d.cfg.BufferProvider.GetReadFrom(w)
+	}
+	defer cleanup()
+
+	resp, err := d.cfg.S3.GetObjectWithContext(d.ctx, in, d.cfg.RequestOptions...)
+	if err != nil {
+		return 0, err
+	}
+	d.setTotalBytes(resp) // Set total if not yet set.
+
+	n, err := io.Copy(w, resp.Body)
+	resp.Body.Close()
+	if err != nil {
+		return n, &errReadingBody{err: err}
+	}
+
+	return n, nil
+}
+
 func logMessage(svc s3iface.S3API, level aws.LogLevelType, msg string) {
 	s, ok := svc.(*s3.S3)
 	if !ok {
@@ -0,0 +1,65 @@
+package s3manager
+
+import (
+	"io"
+	"sync"
+)
+
+// ReadSeekerWriteTo defines an interface implementing io.WriteTo and io.ReadSeeker
+type ReadSeekerWriteTo interface {
+	io.ReadSeeker
+	io.WriterTo
+}
+
+// BufferedReadSeekerWriteTo wraps a BufferedReadSeeker with an io.WriteAt
+// implementation.
+type BufferedReadSeekerWriteTo struct {
+	*BufferedReadSeeker
+}
+
+// WriteTo writes to the given io.Writer from BufferedReadSeeker until there's no more data to write or
+// an error occurs. Returns the number of bytes written and any error encountered during the write.
+func (b *BufferedReadSeekerWriteTo) WriteTo(writer io.Writer) (int64, error) {
+	return io.Copy(writer, b.BufferedReadSeeker)
+}
+
+// ReadSeekerWriteToProvider provides an implementation of io.WriteTo for an io.ReadSeeker
+type ReadSeekerWriteToProvider interface {
+	GetWriteTo(seeker io.ReadSeeker) (r ReadSeekerWriteTo, cleanup func())
+}
+
+// BufferedReadSeekerWriteToPool uses a sync.Pool to create and reuse
+// []byte slices for buffering parts in memory
+type BufferedReadSeekerWriteToPool struct {
+	pool sync.Pool
+}
+
+// NewBufferedReadSeekerWriteToPool will return a new BufferedReadSeekerWriteToPool that will create
+// a pool of reusable buffers . If size is less then < 64 KiB then the buffer
+// will default to 64 KiB. Reason: io.Copy from writers or readers that don't support io.WriteTo or io.ReadFrom
+// respectively will default to copying 32 KiB.
+func NewBufferedReadSeekerWriteToPool(size int) *BufferedReadSeekerWriteToPool {
+	if size < 65536 {
+		size = 65536
+	}
+
+	return &BufferedReadSeekerWriteToPool{
+		pool: sync.Pool{New: func() interface{} {
+			return make([]byte, size)
+		}},
+	}
+}
+
+// GetWriteTo will wrap the provided io.ReadSeeker with a BufferedReadSeekerWriteTo.
+// The provided cleanup must be called after operations have been completed on the
+// returned io.ReadSeekerWriteTo in order to signal the return of resources to the pool.
+func (p *BufferedReadSeekerWriteToPool) GetWriteTo(seeker io.ReadSeeker) (r ReadSeekerWriteTo, cleanup func()) {
+	buffer := p.pool.Get().([]byte)
+
+	r = &BufferedReadSeekerWriteTo{BufferedReadSeeker: NewBufferedReadSeeker(seeker, buffer)}
+	cleanup = func() {
+		p.pool.Put(buffer)
+	}
+
+	return r, cleanup
+}
@@ -162,6 +162,12 @@ type Uploader struct {
 	// List of request options that will be passed down to individual API
 	// operation requests made by the uploader.
 	RequestOptions []request.Option
+
+	// 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
@@ -181,18 +187,25 @@ type Uploader struct {
 //          u.PartSize = 64 * 1024 * 1024 // 64MB per part
 //     })
 func NewUploader(c client.ConfigProvider, options ...func(*Uploader)) *Uploader {
+	return newUploader(s3.New(c), options...)
+}
+
+func newUploader(client s3iface.S3API, options ...func(*Uploader)) *Uploader {
 	u := &Uploader{
-		S3:                s3.New(c),
+		S3:                client,
 		PartSize:          DefaultUploadPartSize,
 		Concurrency:       DefaultUploadConcurrency,
 		LeavePartsOnError: false,
 		MaxUploadParts:    MaxUploadParts,
+		BufferProvider:    defaultUploadBufferProvider(),
 	}

 	for _, option := range options {
 		option(u)
 	}

+	u.partPool = newPartPool(u.PartSize)
+
 	return u
 }

@@ -215,19 +228,7 @@ func NewUploader(c client.ConfigProvider, options ...func(*Uploader)) *Uploader
 //          u.PartSize = 64 * 1024 * 1024 // 64MB per part
 //     })
 func NewUploaderWithClient(svc s3iface.S3API, options ...func(*Uploader)) *Uploader {
-	u := &Uploader{
-		S3:                svc,
-		PartSize:          DefaultUploadPartSize,
-		Concurrency:       DefaultUploadConcurrency,
-		LeavePartsOnError: false,
-		MaxUploadParts:    MaxUploadParts,
-	}
-
-	for _, option := range options {
-		option(u)
-	}
-
-	return u
+	return newUploader(svc, options...)
 }

 // Upload uploads an object to S3, intelligently buffering large files into
@@ -287,6 +288,7 @@ func (u Uploader) UploadWithContext(ctx aws.Context, input *UploadInput, opts ..
 	for _, opt := range opts {
 		opt(&i.cfg)
 	}
+
 	i.cfg.RequestOptions = append(i.cfg.RequestOptions, request.WithAppendUserAgent("S3Manager"))

 	return i.upload()
@@ -356,8 +358,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
@@ -373,15 +373,16 @@ func (u *uploader) upload() (*UploadOutput, error) {
 	}

 	// Do one read to determine if we have more than one part
-	reader, _, part, err := u.nextReader()
+	reader, _, cleanup, err := u.nextReader()
 	if err == io.EOF { // single part
-		return u.singlePart(reader)
+		return u.singlePart(reader, cleanup)
 	} else if err != nil {
+		cleanup()
 		return nil, awserr.New("ReadRequestBody", "read upload data failed", err)
 	}

 	mu := multiuploader{uploader: u}
-	return mu.upload(reader, part)
+	return mu.upload(reader, cleanup)
 }

 // init will initialize all default options.
@@ -396,8 +397,10 @@ func (u *uploader) init() error {
 		u.cfg.MaxUploadParts = MaxUploadParts
 	}

-	u.bufferPool = sync.Pool{
-		New: func() interface{} { return make([]byte, u.cfg.PartSize) },
+	// 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 u.cfg.partPool == nil || u.cfg.partPool.partSize != u.cfg.PartSize {
+		u.cfg.partPool = newPartPool(u.cfg.PartSize)
 	}

 	// Try to get the total size for some optimizations
@@ -433,7 +436,7 @@ func (u *uploader) initSize() error {
 // This operation increases the shared u.readerPos counter, but note that it
 // does not need to be wrapped in a mutex because nextReader is only called
 // from the main thread.
-func (u *uploader) nextReader() (io.ReadSeeker, int, []byte, error) {
+func (u *uploader) nextReader() (io.ReadSeeker, int, func(), error) {
 	type readerAtSeeker interface {
 		io.ReaderAt
 		io.ReadSeeker
@@ -452,17 +455,32 @@ func (u *uploader) nextReader() (io.ReadSeeker, int, []byte, error) {
 			}
 		}

-		reader := io.NewSectionReader(r, u.readerPos, n)
+		var (
+			reader  io.ReadSeeker
+			cleanup func()
+		)
+
+		reader = io.NewSectionReader(r, u.readerPos, n)
+		if u.cfg.BufferProvider != nil {
+			reader, cleanup = u.cfg.BufferProvider.GetWriteTo(reader)
+		} else {
+			cleanup = func() {}
+		}
+
 		u.readerPos += n

-		return reader, int(n), nil, err
+		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)

-		return bytes.NewReader(part[0:n]), n, part, err
+		cleanup := func() {
+			u.cfg.partPool.Put(part)
+		}
+
+		return bytes.NewReader(part[0:n]), n, cleanup, err
 	}
 }

@@ -479,10 +497,12 @@ func readFillBuf(r io.Reader, b []byte) (offset int, err error) {
 // singlePart contains upload logic for uploading a single chunk via
 // a regular PutObject request. Multipart requests require at least two
 // parts, or at least 5MB of data.
-func (u *uploader) singlePart(buf io.ReadSeeker) (*UploadOutput, error) {
+func (u *uploader) singlePart(r io.ReadSeeker, cleanup func()) (*UploadOutput, error) {
+	defer cleanup()
+
 	params := &s3.PutObjectInput{}
 	awsutil.Copy(params, u.in)
-	params.Body = buf
+	params.Body = r

 	// Need to use request form because URL generated in request is
 	// used in return.
@@ -512,9 +532,9 @@ type multiuploader struct {

 // keeps track of a single chunk of data being sent to S3.
 type chunk struct {
-	buf  io.ReadSeeker
-	part []byte
-	num  int64
+	buf     io.ReadSeeker
+	num     int64
+	cleanup func()
 }

 // completedParts is a wrapper to make parts sortable by their part number,
@@ -527,7 +547,7 @@ func (a completedParts) Less(i, j int) bool { return *a[i].PartNumber < *a[j].Pa

 // upload will perform a multipart upload using the firstBuf buffer containing
 // the first chunk of data.
-func (u *multiuploader) upload(firstBuf io.ReadSeeker, firstPart []byte) (*UploadOutput, error) {
+func (u *multiuploader) upload(firstBuf io.ReadSeeker, cleanup func()) (*UploadOutput, error) {
 	params := &s3.CreateMultipartUploadInput{}
 	awsutil.Copy(params, u.in)

@@ -547,46 +567,29 @@ func (u *multiuploader) upload(firstBuf io.ReadSeeker, firstPart []byte) (*Uploa

 	// Send part 1 to the workers
 	var num int64 = 1
-	ch <- chunk{buf: firstBuf, part: firstPart, num: num}
+	ch <- chunk{buf: firstBuf, num: num, cleanup: cleanup}

 	// Read and queue the rest of the parts
 	for u.geterr() == nil && err == nil {
-		var reader io.ReadSeeker
-		var nextChunkLen int
-		var part []byte
-		reader, nextChunkLen, part, err = u.nextReader()
+		var (
+			reader       io.ReadSeeker
+			nextChunkLen int
+			ok           bool
+		)

-		if err != nil && err != io.EOF {
-			u.seterr(awserr.New(
-				"ReadRequestBody",
-				"read multipart upload data failed",
-				err))
-			break
-		}
-
-		if nextChunkLen == 0 {
-			// No need to upload empty part, if file was empty to start
-			// with empty single part would of been created and never
-			// started multipart upload.
+		reader, nextChunkLen, cleanup, err = u.nextReader()
+		ok, err = u.shouldContinue(num, nextChunkLen, err)
+		if !ok {
+			cleanup()
+			if err != nil {
+				u.seterr(err)
+			}
 			break
 		}

 		num++
-		// This upload exceeded maximum number of supported parts, error now.
-		if num > int64(u.cfg.MaxUploadParts) || num > int64(MaxUploadParts) {
-			var msg string
-			if num > int64(u.cfg.MaxUploadParts) {
-				msg = fmt.Sprintf("exceeded total allowed configured MaxUploadParts (%d). Adjust PartSize to fit in this limit",
-					u.cfg.MaxUploadParts)
-			} else {
-				msg = fmt.Sprintf("exceeded total allowed S3 limit MaxUploadParts (%d). Adjust PartSize to fit in this limit",
-					MaxUploadParts)
-			}
-			u.seterr(awserr.New("TotalPartsExceeded", msg, nil))
-			break
-		}

-		ch <- chunk{buf: reader, part: part, num: num}
+		ch <- chunk{buf: reader, num: num, cleanup: cleanup}
 	}

 	// Close the channel, wait for workers, and complete upload
@@ -620,6 +623,35 @@ func (u *multiuploader) upload(firstBuf io.ReadSeeker, firstPart []byte) (*Uploa
 	}, nil
 }

+func (u *multiuploader) shouldContinue(part int64, nextChunkLen int, err error) (bool, error) {
+	if err != nil && err != io.EOF {
+		return false, awserr.New("ReadRequestBody", "read multipart upload data failed", err)
+	}
+
+	if nextChunkLen == 0 {
+		// No need to upload empty part, if file was empty to start
+		// with empty single part would of been created and never
+		// started multipart upload.
+		return false, nil
+	}
+
+	part++
+	// This upload exceeded maximum number of supported parts, error now.
+	if part > int64(u.cfg.MaxUploadParts) || part > int64(MaxUploadParts) {
+		var msg string
+		if part > int64(u.cfg.MaxUploadParts) {
+			msg = fmt.Sprintf("exceeded total allowed configured MaxUploadParts (%d). Adjust PartSize to fit in this limit",
+				u.cfg.MaxUploadParts)
+		} else {
+			msg = fmt.Sprintf("exceeded total allowed S3 limit MaxUploadParts (%d). Adjust PartSize to fit in this limit",
+				MaxUploadParts)
+		}
+		return false, awserr.New("TotalPartsExceeded", msg, nil)
+	}
+
+	return true, err
+}
+
 // readChunk runs in worker goroutines to pull chunks off of the ch channel
 // and send() them as UploadPart requests.
 func (u *multiuploader) readChunk(ch chan chunk) {
@@ -651,9 +683,9 @@ func (u *multiuploader) send(c chunk) error {
 		SSECustomerKey:       u.in.SSECustomerKey,
 		PartNumber:           &c.num,
 	}
+
 	resp, err := u.cfg.S3.UploadPartWithContext(u.ctx, params, u.cfg.RequestOptions...)
-	// put the byte array back into the pool to conserve memory
-	u.bufferPool.Put(c.part)
+	c.cleanup()
 	if err != nil {
 		return err
 	}
@@ -725,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
+}
@@ -12,7 +12,7 @@ import (
 // package's PutObjectInput with the exception that the Body member is an
 // io.Reader instead of an io.ReadSeeker.
 type UploadInput struct {
-	_ struct{} `type:"structure" payload:"Body"`
+	_ struct{} `locationName:"PutObjectRequest" type:"structure" payload:"Body"`

 	// The canned ACL to apply to the object.
 	ACL *string `location:"header" locationName:"x-amz-acl" type:"string" enum:"ObjectCannedACL"`
@@ -0,0 +1,75 @@
+package s3manager
+
+import (
+	"bufio"
+	"io"
+	"sync"
+
+	"github.com/aws/aws-sdk-go/internal/sdkio"
+)
+
+// WriterReadFrom defines an interface implementing io.Writer and io.ReaderFrom
+type WriterReadFrom interface {
+	io.Writer
+	io.ReaderFrom
+}
+
+// WriterReadFromProvider provides an implementation of io.ReadFrom for the given io.Writer
+type WriterReadFromProvider interface {
+	GetReadFrom(writer io.Writer) (w WriterReadFrom, cleanup func())
+}
+
+type bufferedWriter interface {
+	WriterReadFrom
+	Flush() error
+	Reset(io.Writer)
+}
+
+type bufferedReadFrom struct {
+	bufferedWriter
+}
+
+func (b *bufferedReadFrom) ReadFrom(r io.Reader) (int64, error) {
+	n, err := b.bufferedWriter.ReadFrom(r)
+	if flushErr := b.Flush(); flushErr != nil && err == nil {
+		err = flushErr
+	}
+	return n, err
+}
+
+// PooledBufferedReadFromProvider is a WriterReadFromProvider that uses a sync.Pool
+// to manage allocation and reuse of *bufio.Writer structures.
+type PooledBufferedReadFromProvider struct {
+	pool sync.Pool
+}
+
+// NewPooledBufferedWriterReadFromProvider returns a new PooledBufferedReadFromProvider
+// Size is used to control the size of the underlying *bufio.Writer created for
+// calls to GetReadFrom.
+func NewPooledBufferedWriterReadFromProvider(size int) *PooledBufferedReadFromProvider {
+	if size < int(32*sdkio.KibiByte) {
+		size = int(64 * sdkio.KibiByte)
+	}
+
+	return &PooledBufferedReadFromProvider{
+		pool: sync.Pool{
+			New: func() interface{} {
+				return &bufferedReadFrom{bufferedWriter: bufio.NewWriterSize(nil, size)}
+			},
+		},
+	}
+}
+
+// GetReadFrom takes an io.Writer and wraps it with a type which satisfies the WriterReadFrom
+// interface/ Additionally a cleanup function is provided which must be called after usage of the WriterReadFrom
+// has been completed in order to allow the reuse of the *bufio.Writer
+func (p *PooledBufferedReadFromProvider) GetReadFrom(writer io.Writer) (r WriterReadFrom, cleanup func()) {
+	buffer := p.pool.Get().(*bufferedReadFrom)
+	buffer.Reset(writer)
+	r = buffer
+	cleanup = func() {
+		buffer.Reset(nil) // Reset to nil writer to release reference
+		p.pool.Put(buffer)
+	}
+	return r, cleanup
+}