Fix #730 by using a guaranteed reuse buffer pool.

Always return buffers to the pool.
minio · Aug 8, 2017 · 5991548 · 5991548
1 parent 1a09415
commit 5991548
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 18 deletions.
diff --git a/api-put-object-multipart.go b/api-put-object-multipart.go
@@ -27,11 +27,21 @@ import (
 	"sort"
 	"strconv"
 	"strings"
-	"sync"
 
 	"github.com/minio/minio-go/pkg/s3utils"
+	"github.com/oxtoacart/bpool"
 )
 
+// Pool to manage re-usable memory for upload objects
+// with streams with unknown size.
+var bufPool *bpool.BytePool
+
+func init() {
+	_, partSize, _, _ := optimalPartInfo(-1)
+	maxBuffers := 5 // *** ideally this would be 75% of (available_system_memory / partSize), but this is reasonable for 4GB machines?
+	bufPool = bpool.NewBytePool(maxBuffers, int(partSize))
+}
+
 func (c Client) putObjectMultipart(bucketName, objectName string, reader io.Reader, size int64,
 	metadata map[string][]string, progress io.Reader) (n int64, err error) {
 	n, err = c.putObjectMultipartNoStream(bucketName, objectName, reader, metadata, progress)
@@ -51,16 +61,6 @@ func (c Client) putObjectMultipart(bucketName, objectName string, reader io.Read
 	return n, err
 }
 
-// Pool to manage re-usable memory for upload objects
-// with streams with unknown size.
-var bufPool = sync.Pool{
-	New: func() interface{} {
-		_, partSize, _, _ := optimalPartInfo(-1)
-		b := make([]byte, partSize)
-		return &b
-	},
-}
-
 func (c Client) putObjectMultipartNoStream(bucketName, objectName string, reader io.Reader, metadata map[string][]string, progress io.Reader) (n int64, err error) {
 	// Input validation.
 	if err = s3utils.CheckValidBucketName(bucketName); err != nil {
@@ -107,9 +107,10 @@ func (c Client) putObjectMultipartNoStream(bucketName, objectName string, reader
 		// HTTPS connection.
 		hashAlgos, hashSums := c.hashMaterials()
 
-		bufp := bufPool.Get().(*[]byte)
-		length, rErr := io.ReadFull(reader, *bufp)
+		bufp := bufPool.Get()
+		length, rErr := io.ReadFull(reader, bufp)
 		if rErr == io.EOF {
+			bufPool.Put(bufp)
 			break
 		}
 		if rErr != nil && rErr != io.ErrUnexpectedEOF {
@@ -119,13 +120,13 @@ func (c Client) putObjectMultipartNoStream(bucketName, objectName string, reader
 
 		// Calculates hash sums while copying partSize bytes into cw.
 		for k, v := range hashAlgos {
-			v.Write((*bufp)[:length])
+			v.Write(bufp[:length])
 			hashSums[k] = v.Sum(nil)
 		}
 
 		// Update progress reader appropriately to the latest offset
 		// as we read from the source.
-		rd := newHook(bytes.NewReader((*bufp)[:length]), progress)
+		rd := newHook(bytes.NewReader(bufp[:length]), progress)
 
 		// Proceed to upload the part.
 		var objPart ObjectPart

diff --git a/api-put-object.go b/api-put-object.go
@@ -257,9 +257,10 @@ func (c Client) putObjectMultipartStreamNoLength(bucketName, objectName string,
 	partsInfo := make(map[int]ObjectPart)
 
 	for partNumber <= totalPartsCount {
-		bufp := bufPool.Get().(*[]byte)
-		length, rErr := io.ReadFull(reader, *bufp)
+		bufp := bufPool.Get()
+		length, rErr := io.ReadFull(reader, bufp)
 		if rErr == io.EOF {
+			bufPool.Put(bufp)
 			break
 		}
 		if rErr != nil && rErr != io.ErrUnexpectedEOF {
@@ -269,7 +270,7 @@ func (c Client) putObjectMultipartStreamNoLength(bucketName, objectName string,
 
 		// Update progress reader appropriately to the latest offset
 		// as we read from the source.
-		rd := newHook(bytes.NewReader((*bufp)[:length]), progress)
+		rd := newHook(bytes.NewReader(bufp[:length]), progress)
 
 		// Proceed to upload the part.
 		var objPart ObjectPart