Clear string pointers before returning them to the memory pool. (#131)

* Clear string pointers before returning them to the memory pool.

index/field/docs_field.go

@@ -9,6 +9,7 @@ import (
 	"github.com/xichen2020/eventdb/index"
 	"github.com/xichen2020/eventdb/values/impl"
 	"github.com/xichen2020/eventdb/x/pool"
+	"github.com/xichen2020/eventdb/x/strings"
 
 	"github.com/pilosa/pilosa/roaring"
 )
@@ -854,7 +855,7 @@ func (b *docsFieldBuilder) addDouble(docID int32, v float64) error {
 func (b *docsFieldBuilder) addString(docID int32, v string) error {
 	if b.sfb == nil {
 		docIDsBuilder := b.newDocIDSetBuilder()
-		stringValuesBuilder := impl.NewArrayBasedStringValues(b.opts.StringArrayPool())
+		stringValuesBuilder := impl.NewArrayBasedStringValues(b.opts.StringArrayPool(), b.opts.StringArrayResetFn())
 		b.sfb = newStringFieldBuilder(docIDsBuilder, stringValuesBuilder)
 	}
 	return b.sfb.Add(docID, v)
@@ -871,11 +872,12 @@ func (b *docsFieldBuilder) addTime(docID int32, v int64) error {
 
 // DocsFieldBuilderOptions provide a set of options for the field builder.
 type DocsFieldBuilderOptions struct {
-	boolArrayPool   *pool.BucketizedBoolArrayPool
-	intArrayPool    *pool.BucketizedIntArrayPool
-	doubleArrayPool *pool.BucketizedFloat64ArrayPool
-	stringArrayPool *pool.BucketizedStringArrayPool
-	int64ArrayPool  *pool.BucketizedInt64ArrayPool
+	boolArrayPool      *pool.BucketizedBoolArrayPool
+	intArrayPool       *pool.BucketizedIntArrayPool
+	doubleArrayPool    *pool.BucketizedFloat64ArrayPool
+	stringArrayPool    *pool.BucketizedStringArrayPool
+	int64ArrayPool     *pool.BucketizedInt64ArrayPool
+	stringArrayResetFn strings.ArrayFn
 }
 
 // NewDocsFieldBuilderOptions creates a new set of field builder options.
@@ -963,3 +965,15 @@ func (o *DocsFieldBuilderOptions) SetInt64ArrayPool(v *pool.BucketizedInt64Array
 func (o *DocsFieldBuilderOptions) Int64ArrayPool() *pool.BucketizedInt64ArrayPool {
 	return o.int64ArrayPool
 }
+
+// SetStringArrayResetFn sets a value reset function for string values.
+func (o *DocsFieldBuilderOptions) SetStringArrayResetFn(fn strings.ArrayFn) *DocsFieldBuilderOptions {
+	opts := *o
+	opts.stringArrayResetFn = fn
+	return &opts
+}
+
+// StringArrayResetFn resets string array values before returning a string array back to the memory pool.
+func (o *DocsFieldBuilderOptions) StringArrayResetFn() strings.ArrayFn {
+	return o.stringArrayResetFn
+}

storage/mutable_segment.go

@@ -11,6 +11,7 @@ import (
 	"github.com/xichen2020/eventdb/persist"
 	"github.com/xichen2020/eventdb/query"
 	"github.com/xichen2020/eventdb/x/hash"
+	"github.com/xichen2020/eventdb/x/strings"
 	"github.com/xichen2020/eventdb/x/unsafe"
 
 	"github.com/m3db/m3x/context"
@@ -100,7 +101,10 @@ func newMutableSegment(
 		SetIntArrayPool(opts.IntArrayPool()).
 		SetDoubleArrayPool(opts.DoubleArrayPool()).
 		SetStringArrayPool(opts.StringArrayPool()).
-		SetInt64ArrayPool(opts.Int64ArrayPool())
+		SetInt64ArrayPool(opts.Int64ArrayPool()).
+		// Reset string array to avoid holding onto documents after we've returned the referencing
+		// array to the memory pool.
+		SetStringArrayResetFn(strings.ResetArray)
 
 	fieldHashFn := opts.FieldHashFn()
 	timestampFieldPath := opts.TimestampFieldPath()

values/impl/array_based_bool_values.go

@@ -27,7 +27,7 @@ type ArrayBasedBoolValues struct {
 // NewArrayBasedBoolValues create a new array based bool values.
 func NewArrayBasedBoolValues(p *pool.BucketizedBoolArrayPool) *ArrayBasedBoolValues {
 	rawArr := p.Get(defaultInitialFieldValuesCapacity)
-	refCountedArr := pool.NewRefCountedPooledBoolArray(rawArr, p)
+	refCountedArr := pool.NewRefCountedPooledBoolArray(rawArr, p, nil)
 	return &ArrayBasedBoolValues{
 		vals: refCountedArr,
 	}

values/impl/array_based_double_values.go

@@ -28,7 +28,7 @@ type ArrayBasedDoubleValues struct {
 // NewArrayBasedDoubleValues create a new array based double values.
 func NewArrayBasedDoubleValues(p *pool.BucketizedFloat64ArrayPool) *ArrayBasedDoubleValues {
 	rawArr := p.Get(defaultInitialFieldValuesCapacity)
-	refCountedArr := pool.NewRefCountedPooledFloat64Array(rawArr, p)
+	refCountedArr := pool.NewRefCountedPooledFloat64Array(rawArr, p, nil)
 	return &ArrayBasedDoubleValues{
 		vals: refCountedArr,
 	}

values/impl/array_based_int_values.go

@@ -28,7 +28,7 @@ type ArrayBasedIntValues struct {
 // NewArrayBasedIntValues create a new array based int values.
 func NewArrayBasedIntValues(p *pool.BucketizedIntArrayPool) *ArrayBasedIntValues {
 	rawArr := p.Get(defaultInitialFieldValuesCapacity)
-	refCountedArr := pool.NewRefCountedPooledIntArray(rawArr, p)
+	refCountedArr := pool.NewRefCountedPooledIntArray(rawArr, p, nil)
 	return &ArrayBasedIntValues{
 		vals: refCountedArr,
 	}

values/impl/array_based_string_values.go

@@ -9,6 +9,7 @@ import (
 	"github.com/xichen2020/eventdb/values/iterator"
 	iterimpl "github.com/xichen2020/eventdb/values/iterator/impl"
 	"github.com/xichen2020/eventdb/x/pool"
+	"github.com/xichen2020/eventdb/x/strings"
 )
 
 var (
@@ -26,9 +27,12 @@ type ArrayBasedStringValues struct {
 }
 
 // NewArrayBasedStringValues create a new array based string values.
-func NewArrayBasedStringValues(p *pool.BucketizedStringArrayPool) *ArrayBasedStringValues {
+func NewArrayBasedStringValues(
+	p *pool.BucketizedStringArrayPool,
+	stringArrayResetFn strings.ArrayFn,
+) *ArrayBasedStringValues {
 	rawArr := p.Get(defaultInitialFieldValuesCapacity)
-	refCountedArr := pool.NewRefCountedPooledStringArray(rawArr, p)
+	refCountedArr := pool.NewRefCountedPooledStringArray(rawArr, p, stringArrayResetFn)
 	return &ArrayBasedStringValues{
 		vals: refCountedArr,
 	}

values/impl/array_based_time_values.go

@@ -28,7 +28,7 @@ type ArrayBasedTimeValues struct {
 // NewArrayBasedTimeValues create a new array based time values.
 func NewArrayBasedTimeValues(p *pool.BucketizedInt64ArrayPool) *ArrayBasedTimeValues {
 	rawArr := p.Get(defaultInitialFieldValuesCapacity)
-	refCountedArr := pool.NewRefCountedPooledInt64Array(rawArr, p)
+	refCountedArr := pool.NewRefCountedPooledInt64Array(rawArr, p, nil)
 	return &ArrayBasedTimeValues{
 		vals: refCountedArr,
 	}

x/pool/ref_counted_pooled_bool_array.gen.go

@@ -12,21 +12,24 @@ import (
 
 // RefCountedPooledBoolArray is a refcounted, pooled generic value array.
 type RefCountedPooledBoolArray struct {
-	closed bool
-	cnt    *refcnt.RefCounter
-	p      *BucketizedBoolArrayPool
-	vals   []bool
+	closed        bool
+	cnt           *refcnt.RefCounter
+	p             *BucketizedBoolArrayPool
+	vals          []bool
+	valuesResetFn func(values []bool)
 }
 
 // NewRefCountedPooledBoolArray creates a new refcounted, pooled generic value array.
 func NewRefCountedPooledBoolArray(
 	vals []bool,
 	p *BucketizedBoolArrayPool,
+	resetFn func(values []bool),
 ) *RefCountedPooledBoolArray {
 	return &RefCountedPooledBoolArray{
-		cnt:  refcnt.NewRefCounter(),
-		p:    p,
-		vals: vals,
+		cnt:           refcnt.NewRefCounter(),
+		p:             p,
+		vals:          vals,
+		valuesResetFn: resetFn,
 	}
 }
 
@@ -75,6 +78,9 @@ func (rv *RefCountedPooledBoolArray) tryRelease() {
 	if rv.cnt.DecRef() > 0 {
 		return
 	}
+	if rv.valuesResetFn != nil {
+		rv.valuesResetFn(rv.vals)
+	}
 	rv.vals = rv.vals[:0]
 	rv.p.Put(rv.vals, cap(rv.vals))
 	rv.vals = nil

x/pool/ref_counted_pooled_float64_array.gen.go

@@ -12,21 +12,24 @@ import (
 
 // RefCountedPooledFloat64Array is a refcounted, pooled generic value array.
 type RefCountedPooledFloat64Array struct {
-	closed bool
-	cnt    *refcnt.RefCounter
-	p      *BucketizedFloat64ArrayPool
-	vals   []float64
+	closed        bool
+	cnt           *refcnt.RefCounter
+	p             *BucketizedFloat64ArrayPool
+	vals          []float64
+	valuesResetFn func(values []float64)
 }
 
 // NewRefCountedPooledFloat64Array creates a new refcounted, pooled generic value array.
 func NewRefCountedPooledFloat64Array(
 	vals []float64,
 	p *BucketizedFloat64ArrayPool,
+	resetFn func(values []float64),
 ) *RefCountedPooledFloat64Array {
 	return &RefCountedPooledFloat64Array{
-		cnt:  refcnt.NewRefCounter(),
-		p:    p,
-		vals: vals,
+		cnt:           refcnt.NewRefCounter(),
+		p:             p,
+		vals:          vals,
+		valuesResetFn: resetFn,
 	}
 }
 
@@ -75,6 +78,9 @@ func (rv *RefCountedPooledFloat64Array) tryRelease() {
 	if rv.cnt.DecRef() > 0 {
 		return
 	}
+	if rv.valuesResetFn != nil {
+		rv.valuesResetFn(rv.vals)
+	}
 	rv.vals = rv.vals[:0]
 	rv.p.Put(rv.vals, cap(rv.vals))
 	rv.vals = nil

x/pool/ref_counted_pooled_int64_array.gen.go

@@ -12,21 +12,24 @@ import (
 
 // RefCountedPooledInt64Array is a refcounted, pooled generic value array.
 type RefCountedPooledInt64Array struct {
-	closed bool
-	cnt    *refcnt.RefCounter
-	p      *BucketizedInt64ArrayPool
-	vals   []int64
+	closed        bool
+	cnt           *refcnt.RefCounter
+	p             *BucketizedInt64ArrayPool
+	vals          []int64
+	valuesResetFn func(values []int64)
 }
 
 // NewRefCountedPooledInt64Array creates a new refcounted, pooled generic value array.
 func NewRefCountedPooledInt64Array(
 	vals []int64,
 	p *BucketizedInt64ArrayPool,
+	resetFn func(values []int64),
 ) *RefCountedPooledInt64Array {
 	return &RefCountedPooledInt64Array{
-		cnt:  refcnt.NewRefCounter(),
-		p:    p,
-		vals: vals,
+		cnt:           refcnt.NewRefCounter(),
+		p:             p,
+		vals:          vals,
+		valuesResetFn: resetFn,
 	}
 }
 
@@ -75,6 +78,9 @@ func (rv *RefCountedPooledInt64Array) tryRelease() {
 	if rv.cnt.DecRef() > 0 {
 		return
 	}
+	if rv.valuesResetFn != nil {
+		rv.valuesResetFn(rv.vals)
+	}
 	rv.vals = rv.vals[:0]
 	rv.p.Put(rv.vals, cap(rv.vals))
 	rv.vals = nil

x/pool/ref_counted_pooled_int_array.gen.go

@@ -12,21 +12,24 @@ import (
 
 // RefCountedPooledIntArray is a refcounted, pooled generic value array.
 type RefCountedPooledIntArray struct {
-	closed bool
-	cnt    *refcnt.RefCounter
-	p      *BucketizedIntArrayPool
-	vals   []int
+	closed        bool
+	cnt           *refcnt.RefCounter
+	p             *BucketizedIntArrayPool
+	vals          []int
+	valuesResetFn func(values []int)
 }
 
 // NewRefCountedPooledIntArray creates a new refcounted, pooled generic value array.
 func NewRefCountedPooledIntArray(
 	vals []int,
 	p *BucketizedIntArrayPool,
+	resetFn func(values []int),
 ) *RefCountedPooledIntArray {
 	return &RefCountedPooledIntArray{
-		cnt:  refcnt.NewRefCounter(),
-		p:    p,
-		vals: vals,
+		cnt:           refcnt.NewRefCounter(),
+		p:             p,
+		vals:          vals,
+		valuesResetFn: resetFn,
 	}
 }
 
@@ -75,6 +78,9 @@ func (rv *RefCountedPooledIntArray) tryRelease() {
 	if rv.cnt.DecRef() > 0 {
 		return
 	}
+	if rv.valuesResetFn != nil {
+		rv.valuesResetFn(rv.vals)
+	}
 	rv.vals = rv.vals[:0]
 	rv.p.Put(rv.vals, cap(rv.vals))
 	rv.vals = nil

x/pool/ref_counted_pooled_string_array.gen.go

@@ -12,21 +12,24 @@ import (
 
 // RefCountedPooledStringArray is a refcounted, pooled generic value array.
 type RefCountedPooledStringArray struct {
-	closed bool
-	cnt    *refcnt.RefCounter
-	p      *BucketizedStringArrayPool
-	vals   []string
+	closed        bool
+	cnt           *refcnt.RefCounter
+	p             *BucketizedStringArrayPool
+	vals          []string
+	valuesResetFn func(values []string)
 }
 
 // NewRefCountedPooledStringArray creates a new refcounted, pooled generic value array.
 func NewRefCountedPooledStringArray(
 	vals []string,
 	p *BucketizedStringArrayPool,
+	resetFn func(values []string),
 ) *RefCountedPooledStringArray {
 	return &RefCountedPooledStringArray{
-		cnt:  refcnt.NewRefCounter(),
-		p:    p,
-		vals: vals,
+		cnt:           refcnt.NewRefCounter(),
+		p:             p,
+		vals:          vals,
+		valuesResetFn: resetFn,
 	}
 }
 
@@ -75,6 +78,9 @@ func (rv *RefCountedPooledStringArray) tryRelease() {
 	if rv.cnt.DecRef() > 0 {
 		return
 	}
+	if rv.valuesResetFn != nil {
+		rv.valuesResetFn(rv.vals)
+	}
 	rv.vals = rv.vals[:0]
 	rv.p.Put(rv.vals, cap(rv.vals))
 	rv.vals = nil

x/pool/ref_counted_pooled_string_array_test.go

@@ -15,7 +15,7 @@ func TestRefCountedPooledStringArray(t *testing.T) {
 	pool.Init(func(capacity int) []string { return make([]string, 0, capacity) })
 
 	vals := pool.Get(4)
-	arr := NewRefCountedPooledStringArray(vals, pool)
+	arr := NewRefCountedPooledStringArray(vals, pool, nil)
 	require.Equal(t, int32(1), arr.cnt.RefCount())
 	require.Equal(t, 0, len(arr.Get()))
 	require.Equal(t, 4, cap(arr.Get()))

x/pool/template/ref_counted_pooled_array.go

@@ -16,21 +16,24 @@ type GenericBucketizedValueArrayPool interface {
 
 // RefCountedPooledGenericValueArray is a refcounted, pooled generic value array.
 type RefCountedPooledGenericValueArray struct {
-	closed bool
-	cnt    *refcnt.RefCounter
-	p      GenericBucketizedValueArrayPool
-	vals   []GenericValue
+	closed        bool
+	cnt           *refcnt.RefCounter
+	p             GenericBucketizedValueArrayPool
+	vals          []GenericValue
+	valuesResetFn func(values []GenericValue)
 }
 
 // NewRefCountedPooledGenericValueArray creates a new refcounted, pooled generic value array.
 func NewRefCountedPooledGenericValueArray(
 	vals []GenericValue,
 	p GenericBucketizedValueArrayPool,
+	resetFn func(values []GenericValue),
 ) *RefCountedPooledGenericValueArray {
 	return &RefCountedPooledGenericValueArray{
-		cnt:  refcnt.NewRefCounter(),
-		p:    p,
-		vals: vals,
+		cnt:           refcnt.NewRefCounter(),
+		p:             p,
+		vals:          vals,
+		valuesResetFn: resetFn,
 	}
 }
 
@@ -79,6 +82,9 @@ func (rv *RefCountedPooledGenericValueArray) tryRelease() {
 	if rv.cnt.DecRef() > 0 {
 		return
 	}
+	if rv.valuesResetFn != nil {
+		rv.valuesResetFn(rv.vals)
+	}
 	rv.vals = rv.vals[:0]
 	rv.p.Put(rv.vals, cap(rv.vals))
 	rv.vals = nil