[pool] Add support for dynamic, sync.Pool backed, object pools (#3334)

src/cmd/services/m3dbnode/config/pooling.go

@@ -20,7 +20,11 @@
 
 package config
 
-import "fmt"
+import (
+	"fmt"
+
+	"github.com/m3db/m3/src/x/pool"
+)
 
 // PoolingType is a type of pooling, using runtime or mmap'd bytes pooling.
 type PoolingType string
@@ -39,7 +43,7 @@ const (
 )
 
 type poolPolicyDefault struct {
-	size                int
+	size                pool.Size
 	refillLowWaterMark  float64
 	refillHighWaterMark float64
 
@@ -214,55 +218,55 @@ var (
 				{
 					Capacity: intPtr(16),
 					PoolPolicy: PoolPolicy{
-						Size:                intPtr(524288),
+						Size:                poolSizePtr(524288),
 						RefillLowWaterMark:  &defaultRefillLowWaterMark,
 						RefillHighWaterMark: &defaultRefillHighWaterMark,
 					},
 				},
 				{
 					Capacity: intPtr(32),
 					PoolPolicy: PoolPolicy{
-						Size:                intPtr(262144),
+						Size:                poolSizePtr(262144),
 						RefillLowWaterMark:  &defaultRefillLowWaterMark,
 						RefillHighWaterMark: &defaultRefillHighWaterMark,
 					},
 				},
 				{
 					Capacity: intPtr(64),
 					PoolPolicy: PoolPolicy{
-						Size:                intPtr(131072),
+						Size:                poolSizePtr(131072),
 						RefillLowWaterMark:  &defaultRefillLowWaterMark,
 						RefillHighWaterMark: &defaultRefillHighWaterMark,
 					},
 				},
 				{
 					Capacity: intPtr(128),
 					PoolPolicy: PoolPolicy{
-						Size:                intPtr(65536),
+						Size:                poolSizePtr(65536),
 						RefillLowWaterMark:  &defaultRefillLowWaterMark,
 						RefillHighWaterMark: &defaultRefillHighWaterMark,
 					},
 				},
 				{
 					Capacity: intPtr(256),
 					PoolPolicy: PoolPolicy{
-						Size:                intPtr(65536),
+						Size:                poolSizePtr(65536),
 						RefillLowWaterMark:  &defaultRefillLowWaterMark,
 						RefillHighWaterMark: &defaultRefillHighWaterMark,
 					},
 				},
 				{
 					Capacity: intPtr(1440),
 					PoolPolicy: PoolPolicy{
-						Size:                intPtr(16384),
+						Size:                poolSizePtr(16384),
 						RefillLowWaterMark:  &defaultRefillLowWaterMark,
 						RefillHighWaterMark: &defaultRefillHighWaterMark,
 					},
 				},
 				{
 					Capacity: intPtr(4096),
 					PoolPolicy: PoolPolicy{
-						Size:                intPtr(8192),
+						Size:                poolSizePtr(8192),
 						RefillLowWaterMark:  &defaultRefillLowWaterMark,
 						RefillHighWaterMark: &defaultRefillHighWaterMark,
 					},
@@ -486,7 +490,7 @@ func (p *PoolingPolicy) TypeOrDefault() PoolingType {
 // PoolPolicy specifies a single pool policy.
 type PoolPolicy struct {
 	// The size of the pool.
-	Size *int `yaml:"size"`
+	Size *pool.Size `yaml:"size"`
 
 	// The low watermark to start refilling the pool, if zero none.
 	RefillLowWaterMark *float64 `yaml:"lowWatermark"`
@@ -525,7 +529,7 @@ func (p *PoolPolicy) initDefaultsAndValidate(poolName string) error {
 }
 
 // SizeOrDefault returns the configured size if present, or a default value otherwise.
-func (p *PoolPolicy) SizeOrDefault() int {
+func (p *PoolPolicy) SizeOrDefault() pool.Size {
 	return *p.Size
 }
 
@@ -668,3 +672,8 @@ type WriteBatchPoolPolicy struct {
 func intPtr(x int) *int {
 	return &x
 }
+
+func poolSizePtr(x int) *pool.Size {
+	sz := pool.Size(x)
+	return &sz
+}

src/dbnode/client/session.go

@@ -939,7 +939,7 @@ func (s *session) setTopologyWithLock(topoMap topology.Map, queues []hostQueue,
 		s.pools.multiReaderIteratorArray = encoding.NewMultiReaderIteratorArrayPool([]pool.Bucket{
 			pool.Bucket{
 				Capacity: replicas,
-				Count:    s.opts.SeriesIteratorPoolSize(),
+				Count:    pool.Size(s.opts.SeriesIteratorPoolSize()),
 			},
 		})
 		s.pools.multiReaderIteratorArray.Init()

src/dbnode/encoding/multi_reader_iterator_array_pool.go

@@ -66,7 +66,7 @@ func (p *multiReaderIteratorArrayPool) Init() {
 	for i := range p.sizesAsc {
 		buckets[i].capacity = p.sizesAsc[i].Capacity
 		buckets[i].values = make(chan []MultiReaderIterator, p.sizesAsc[i].Count)
-		for j := 0; j < p.sizesAsc[i].Count; j++ {
+		for j := 0; pool.Size(j) < p.sizesAsc[i].Count; j++ {
 			buckets[i].values <- p.alloc(p.sizesAsc[i].Capacity)
 		}
 	}

src/dbnode/encoding/mutable_series_iterators_pool.go

@@ -59,7 +59,7 @@ func (p *seriesIteratorsPool) Init() {
 	for i := range p.sizesAsc {
 		buckets[i].capacity = p.sizesAsc[i].Capacity
 		buckets[i].values = make(chan MutableSeriesIterators, p.sizesAsc[i].Count)
-		for j := 0; j < p.sizesAsc[i].Count; j++ {
+		for j := 0; pool.Size(j) < p.sizesAsc[i].Count; j++ {
 			buckets[i].values <- p.alloc(p.sizesAsc[i].Capacity)
 		}
 	}

src/dbnode/server/server.go

@@ -1365,7 +1365,7 @@ func withEncodingAndPoolingOptions(
 
 		logger.Info("bytes pool configured",
 			zap.Int("capacity", bucket.CapacityOrDefault()),
-			zap.Int("size", bucket.SizeOrDefault()),
+			zap.Int("size", int(bucket.SizeOrDefault())),
 			zap.Float64("refillLowWaterMark", bucket.RefillLowWaterMarkOrDefault()),
 			zap.Float64("refillHighWaterMark", bucket.RefillHighWaterMarkOrDefault()))
 	}
@@ -1752,7 +1752,7 @@ func poolOptions(
 	)
 
 	if size > 0 {
-		opts = opts.SetSize(size)
+		opts = opts.SetSize(int(size))
 		if refillLowWaterMark > 0 &&
 			refillHighWaterMark > 0 &&
 			refillHighWaterMark > refillLowWaterMark {
@@ -1761,6 +1761,8 @@ func poolOptions(
 				SetRefillHighWatermark(refillHighWaterMark)
 		}
 	}
+	opts = opts.SetDynamic(size.IsDynamic())
+
 	if scope != nil {
 		opts = opts.SetInstrumentOptions(opts.InstrumentOptions().
 			SetMetricsScope(scope))
@@ -1780,14 +1782,16 @@ func capacityPoolOptions(
 	)
 
 	if size > 0 {
-		opts = opts.SetSize(size)
+		opts = opts.SetSize(int(size))
 		if refillLowWaterMark > 0 &&
 			refillHighWaterMark > 0 &&
 			refillHighWaterMark > refillLowWaterMark {
 			opts = opts.SetRefillLowWatermark(refillLowWaterMark)
 			opts = opts.SetRefillHighWatermark(refillHighWaterMark)
 		}
 	}
+	opts = opts.SetDynamic(size.IsDynamic())
+
 	if scope != nil {
 		opts = opts.SetInstrumentOptions(opts.InstrumentOptions().
 			SetMetricsScope(scope))
@@ -1807,14 +1811,16 @@ func maxCapacityPoolOptions(
 	)
 
 	if size > 0 {
-		opts = opts.SetSize(size)
+		opts = opts.SetSize(int(size))
 		if refillLowWaterMark > 0 &&
 			refillHighWaterMark > 0 &&
 			refillHighWaterMark > refillLowWaterMark {
 			opts = opts.SetRefillLowWatermark(refillLowWaterMark)
 			opts = opts.SetRefillHighWatermark(refillHighWaterMark)
 		}
 	}
+	opts = opts.SetDynamic(size.IsDynamic())
+
 	if scope != nil {
 		opts = opts.SetInstrumentOptions(opts.InstrumentOptions().
 			SetMetricsScope(scope))

src/msg/consumer/config.go

@@ -44,7 +44,7 @@ type Configuration struct {
 // options, which extends the default object pool configuration.
 type MessagePoolConfiguration struct {
 	// Size is the size of the pool.
-	Size int `yaml:"size"`
+	Size pool.Size `yaml:"size"`
 
 	// Watermark is the object pool watermark configuration.
 	Watermark pool.WatermarkConfiguration `yaml:"watermark"`

src/msg/protocol/proto/roundtrip_test.go

@@ -197,7 +197,7 @@ func getBytesPool(bucketSizes int, bucketCaps []int) pool.BytesPool {
 	buckets := make([]pool.Bucket, len(bucketCaps))
 	for i, cap := range bucketCaps {
 		buckets[i] = pool.Bucket{
-			Count:    bucketSizes,
+			Count:    pool.Size(bucketSizes),
 			Capacity: cap,
 		}
 	}

src/query/ts/m3db/options.go

@@ -74,7 +74,7 @@ type encodedBlockOptions struct {
 func NewOptions() Options {
 	bytesPool := pool.NewCheckedBytesPool([]pool.Bucket{{
 		Capacity: defaultCapacity,
-		Count:    defaultCount,
+		Count:    pool.Size(defaultCount),
 	}}, nil, func(s []pool.Bucket) pool.BytesPool {
 		return pool.NewBytesPool(s, nil)
 	})

src/x/pool/bucketized.go

@@ -21,8 +21,8 @@
 package pool
 
 import (
-	"fmt"
 	"sort"
+	"strconv"
 
 	"github.com/uber-go/tally"
 )
@@ -77,13 +77,21 @@ func (p *bucketizedObjectPool) Init(alloc BucketizedAllocator) {
 			opts = perBucketOpts
 		}
 
-		opts = opts.SetSize(size)
+		if size > 0 {
+			opts = opts.SetSize(int(size))
+		}
+		opts = opts.SetDynamic(size.IsDynamic())
+
 		iopts := opts.InstrumentOptions()
 
 		if iopts.MetricsScope() != nil {
+			sz := strconv.Itoa(capacity)
+			if capacity <= 0 {
+				sz = "dynamic"
+			}
 			opts = opts.SetInstrumentOptions(iopts.SetMetricsScope(
 				iopts.MetricsScope().Tagged(map[string]string{
-					"bucket-capacity": fmt.Sprintf("%d", capacity),
+					"bucket-capacity": sz,
 				})))
 		}
 

src/x/pool/bytes_test.go

@@ -89,7 +89,7 @@ func getBytesPool(bucketSizes int, bucketCaps []int) *bytesPool {
 	buckets := make([]Bucket, len(bucketCaps))
 	for i, cap := range bucketCaps {
 		buckets[i] = Bucket{
-			Count:    bucketSizes,
+			Count:    Size(bucketSizes),
 			Capacity: cap,
 		}
 	}

src/x/pool/checked_bytes_test.go

@@ -109,7 +109,7 @@ func getCheckedBytesPool(
 	buckets := make([]Bucket, len(bucketCaps))
 	for i, cap := range bucketCaps {
 		buckets[i] = Bucket{
-			Count:    bucketSizes,
+			Count:    Size(bucketSizes),
 			Capacity: cap,
 		}
 	}

src/x/pool/config.go

@@ -20,12 +20,17 @@
 
 package pool
 
-import "github.com/m3db/m3/src/x/instrument"
+import (
+	"errors"
+	"strconv"
+
+	"github.com/m3db/m3/src/x/instrument"
+)
 
 // ObjectPoolConfiguration contains configuration for object pools.
 type ObjectPoolConfiguration struct {
 	// The size of the pool.
-	Size int `yaml:"size"`
+	Size Size `yaml:"size"`
 
 	// The watermark configuration.
 	Watermark WatermarkConfiguration `yaml:"watermark"`
@@ -35,13 +40,15 @@ type ObjectPoolConfiguration struct {
 func (c *ObjectPoolConfiguration) NewObjectPoolOptions(
 	instrumentOpts instrument.Options,
 ) ObjectPoolOptions {
-	size := defaultSize
-	if c.Size != 0 {
-		size = c.Size
+	size := _defaultSize
+	if c.Size > 0 {
+		size = int(c.Size)
 	}
+
 	return NewObjectPoolOptions().
 		SetInstrumentOptions(instrumentOpts).
 		SetSize(size).
+		SetDynamic(c.Size.IsDynamic()).
 		SetRefillLowWatermark(c.Watermark.RefillLowWatermark).
 		SetRefillHighWatermark(c.Watermark.RefillHighWatermark)
 }
@@ -78,7 +85,7 @@ func (c *BucketizedPoolConfiguration) NewBuckets() []Bucket {
 // BucketConfiguration contains configuration for a pool bucket.
 type BucketConfiguration struct {
 	// The count of the items in the bucket.
-	Count int `yaml:"count"`
+	Count Size `yaml:"count"`
 
 	// The capacity of each item in the bucket.
 	Capacity int `yaml:"capacity"`
@@ -100,3 +107,32 @@ type WatermarkConfiguration struct {
 	// The high watermark to stop refilling the pool, if zero none.
 	RefillHighWatermark float64 `yaml:"high" validate:"min=0.0,max=1.0"`
 }
+
+// Size stores pool capacity for pools that can be either dynamic or pre-allocated
+type Size int
+
+// UnmarshalText unmarshals Size.
+func (s *Size) UnmarshalText(b []byte) error {
+	if string(b) == "dynamic" {
+		*s = _dynamicPoolSize
+		return nil
+	}
+
+	i, err := strconv.ParseInt(string(b), 10, 64)
+	if err != nil {
+		return err
+	}
+
+	if i < 0 {
+		return errors.New("pool capacity cannot be negative")
+	}
+
+	*s = Size(i)
+
+	return nil
+}
+
+// IsDynamic returns whether the pool should be fixed size or not.
+func (s Size) IsDynamic() bool {
+	return s == _dynamicPoolSize
+}