func (e *GaugeElem) indexOfWithLock(alignedStart int64) (int, bool) {
	numValues := len(e.values)
	if numValues > 0 && e.values[numValues-1].startAtNanos == alignedStart {
		return numValues - 1, true
	}
	left, right := 0, numValues
	for left < right {
		mid := left + (right-left)/2  
		if e.values[mid].startAtNanos < alignedStart {
			left = mid + 1
		} else {
			right = mid
		}
	}
	if left < numValues && e.values[left].startAtNanos == alignedStart {
		return left, true
	}
	return left, false
}
indexOfWithLock finds the smallest element index whose timestamp
is no smaller than the start time passed in, and true if it's an
exact match, false otherwise.
package aggregator
import (
	"fmt"
	"math"
	"sync"
	"time"
	maggregation "github.com/m3db/m3metrics/aggregation"
	"github.com/m3db/m3metrics/metric/id"
	"github.com/m3db/m3metrics/metric/unaggregated"
	"github.com/m3db/m3metrics/pipeline/applied"
	"github.com/m3db/m3metrics/policy"
	"github.com/m3db/m3metrics/transformation"
	"github.com/willf/bitset"
)
type lockedGaugeAggregation struct {
	sync.Mutex
	closed      bool
	sourcesSeen *bitset.BitSet
	aggregation gaugeAggregation
}
type timedGauge struct {
	startAtNanos int64  
	lockedAgg    *lockedGaugeAggregation
}
func (ta *timedGauge) Reset() {
	ta.startAtNanos = 0
	ta.lockedAgg = nil
}
type GaugeElem struct {
	elemBase
	gaugeElemBase
	values              []timedGauge  
	toConsume           []timedGauge  
	lastConsumedAtNanos int64         
	lastConsumedValues  []float64     
}
func NewGaugeElem(
	id id.RawID,
	sp policy.StoragePolicy,
	aggTypes maggregation.Types,
	pipeline applied.Pipeline,
	numForwardedTimes int,
	idPrefixSuffixType IDPrefixSuffixType,
	opts Options,
) (*GaugeElem, error) {
	e := &GaugeElem{
		elemBase: newElemBase(opts),
		values:   make([]timedGauge, 0, defaultNumAggregations),  
	}
	if err := e.ResetSetData(id, sp, aggTypes, pipeline, numForwardedTimes, idPrefixSuffixType); err != nil {
		return nil, err
	}
	return e, nil
}
func MustNewGaugeElem(
	id id.RawID,
	sp policy.StoragePolicy,
	aggTypes maggregation.Types,
	pipeline applied.Pipeline,
	numForwardedTimes int,
	idPrefixSuffixType IDPrefixSuffixType,
	opts Options,
) *GaugeElem {
	elem, err := NewGaugeElem(id, sp, aggTypes, pipeline, numForwardedTimes, idPrefixSuffixType, opts)
	if err != nil {
		panic(fmt.Errorf("unable to create element: %v", err))
	}
	return elem
}
func (e *GaugeElem) ResetSetData(
	id id.RawID,
	sp policy.StoragePolicy,
	aggTypes maggregation.Types,
	pipeline applied.Pipeline,
	numForwardedTimes int,
	idPrefixSuffixType IDPrefixSuffixType,
) error {
	useDefaultAggregation := aggTypes.IsDefault()
	if useDefaultAggregation {
		aggTypes = e.DefaultAggregationTypes(e.aggTypesOpts)
	}
	if err := e.elemBase.resetSetData(id, sp, aggTypes, useDefaultAggregation, pipeline, numForwardedTimes, idPrefixSuffixType); err != nil {
		return err
	}
	if err := e.gaugeElemBase.ResetSetData(e.aggTypesOpts, aggTypes, useDefaultAggregation); err != nil {
		return err
	}
	if !e.parsedPipeline.HasDerivativeTransform {
		return nil
	}
	numAggTypes := len(e.aggTypes)
	if cap(e.lastConsumedValues) < numAggTypes {
		e.lastConsumedValues = make([]float64, numAggTypes)
	}
	e.lastConsumedValues = e.lastConsumedValues[:numAggTypes]
	for i := 0; i < len(e.lastConsumedValues); i++ {
		e.lastConsumedValues[i] = nan
	}
	return nil
}
func (e *GaugeElem) AddUnion(timestamp time.Time, mu unaggregated.MetricUnion) error {
	alignedStart := timestamp.Truncate(e.sp.Resolution().Window).UnixNano()
	lockedAgg, err := e.findOrCreate(alignedStart, createAggregationOptions{})
	if err != nil {
		return err
	}
	lockedAgg.Lock()
	if lockedAgg.closed {
		lockedAgg.Unlock()
		return errAggregationClosed
	}
	lockedAgg.aggregation.AddUnion(mu)
	lockedAgg.Unlock()
	return nil
}
func (e *GaugeElem) AddValue(timestamp time.Time, value float64) error {
	alignedStart := timestamp.Truncate(e.sp.Resolution().Window).UnixNano()
	lockedAgg, err := e.findOrCreate(alignedStart, createAggregationOptions{})
	if err != nil {
		return err
	}
	lockedAgg.Lock()
	if lockedAgg.closed {
		lockedAgg.Unlock()
		return errAggregationClosed
	}
	lockedAgg.aggregation.Add(value)
	lockedAgg.Unlock()
	return nil
}
func (e *GaugeElem) AddUnique(timestamp time.Time, values []float64, sourceID uint32) error {
	alignedStart := timestamp.Truncate(e.sp.Resolution().Window).UnixNano()
	lockedAgg, err := e.findOrCreate(alignedStart, createAggregationOptions{initSourceSet: true})
	if err != nil {
		return err
	}
	lockedAgg.Lock()
	if lockedAgg.closed {
		lockedAgg.Unlock()
		return errAggregationClosed
	}
	source := uint(sourceID)
	if lockedAgg.sourcesSeen.Test(source) {
		lockedAgg.Unlock()
		return errDuplicateForwardingSource
	}
	lockedAgg.sourcesSeen.Set(source)
	for _, v := range values {
		lockedAgg.aggregation.Add(v)
	}
	lockedAgg.Unlock()
	return nil
}
func (e *GaugeElem) Consume(
	targetNanos int64,
	isEarlierThanFn isEarlierThanFn,
	timestampNanosFn timestampNanosFn,
	flushLocalFn flushLocalMetricFn,
	flushForwardedFn flushForwardedMetricFn,
	onForwardedFlushedFn onForwardingElemFlushedFn,
) bool {
	resolution := e.sp.Resolution().Window
	e.Lock()
	if e.closed {
		e.Unlock()
		return false
	}
	idx := 0
	for range e.values {
		if !isEarlierThanFn(e.values[idx].startAtNanos, resolution, targetNanos) {
			break
		}
		idx++
	}
	e.toConsume = e.toConsume[:0]
	if idx > 0 {
		e.toConsume = append(e.toConsume, e.values[:idx]...)
		n := copy(e.values[0:], e.values[idx:])
		for i := n; i < len(e.values); i++ {
			e.values[i].Reset()
		}
		e.values = e.values[:n]
	}
	canCollect := len(e.values) == 0 && e.tombstoned
	e.Unlock()
	for i := range e.toConsume {
		timeNanos := timestampNanosFn(e.toConsume[i].startAtNanos, resolution)
		e.toConsume[i].lockedAgg.Lock()
		e.processValueWithAggregationLock(timeNanos, e.toConsume[i].lockedAgg, flushLocalFn, flushForwardedFn)
		e.toConsume[i].lockedAgg.closed = true
		e.toConsume[i].lockedAgg.aggregation.Close()
		if e.toConsume[i].lockedAgg.sourcesSeen != nil {
			e.cachedSourceSetsLock.Lock()
			if len(e.cachedSourceSets) < e.opts.MaxNumCachedSourceSets() {
				e.cachedSourceSets = append(e.cachedSourceSets, e.toConsume[i].lockedAgg.sourcesSeen)
			}
			e.cachedSourceSetsLock.Unlock()
			e.toConsume[i].lockedAgg.sourcesSeen = nil
		}
		e.toConsume[i].lockedAgg.Unlock()
		e.toConsume[i].Reset()
	}
	if e.parsedPipeline.HasRollup {
		forwardedAggregationKey, _ := e.ForwardedAggregationKey()
		onForwardedFlushedFn(e.onForwardedAggregationWrittenFn, forwardedAggregationKey)
	}
	return canCollect
}
func (e *GaugeElem) Close() {
	e.Lock()
	if e.closed {
		e.Unlock()
		return
	}
	e.closed = true
	e.id = nil
	e.parsedPipeline = parsedPipeline{}
	e.writeForwardedMetricFn = nil
	e.onForwardedAggregationWrittenFn = nil
	for idx := range e.cachedSourceSets {
		e.cachedSourceSets[idx] = nil
	}
	e.cachedSourceSets = nil
	for idx := range e.values {
		e.values[idx].lockedAgg.sourcesSeen = nil
		e.values[idx].lockedAgg.aggregation.Close()
		e.values[idx].Reset()
	}
	e.values = e.values[:0]
	e.toConsume = e.toConsume[:0]
	e.lastConsumedValues = e.lastConsumedValues[:0]
	e.gaugeElemBase.Close()
	aggTypesPool := e.aggTypesOpts.TypesPool()
	pool := e.ElemPool(e.opts)
	e.Unlock()
	if !e.useDefaultAggregation {
		aggTypesPool.Put(e.aggTypes)
	}
	pool.Put(e)
}
func (e *GaugeElem) findOrCreate(
	alignedStart int64,
	createOpts createAggregationOptions,
) (*lockedGaugeAggregation, error) {
	e.RLock()
	if e.closed {
		e.RUnlock()
		return nil, errElemClosed
	}
	idx, found := e.indexOfWithLock(alignedStart)
	if found {
		agg := e.values[idx].lockedAgg
		e.RUnlock()
		return agg, nil
	}
	e.RUnlock()
	e.Lock()
	if e.closed {
		e.Unlock()
		return nil, errElemClosed
	}
	idx, found = e.indexOfWithLock(alignedStart)
	if found {
		agg := e.values[idx].lockedAgg
		e.Unlock()
		return agg, nil
	}
	numValues := len(e.values)
	e.values = append(e.values, timedGauge{})
	copy(e.values[idx+1:numValues+1], e.values[idx:numValues])
	var sourcesSeen *bitset.BitSet
	if createOpts.initSourceSet {
		e.cachedSourceSetsLock.Lock()
		if numCachedSourceSets := len(e.cachedSourceSets); numCachedSourceSets > 0 {
			sourcesSeen = e.cachedSourceSets[numCachedSourceSets-1]
			e.cachedSourceSets[numCachedSourceSets-1] = nil
			e.cachedSourceSets = e.cachedSourceSets[:numCachedSourceSets-1]
			sourcesSeen.ClearAll()
		} else {
			sourcesSeen = bitset.New(defaultNumSources)
		}
		e.cachedSourceSetsLock.Unlock()
	}
	e.values[idx] = timedGauge{
		startAtNanos: alignedStart,
		lockedAgg: &lockedGaugeAggregation{
			sourcesSeen: sourcesSeen,
			aggregation: e.NewAggregation(e.opts, e.aggOpts),
		},
	}
	agg := e.values[idx].lockedAgg
	e.Unlock()
	return agg, nil
}