func (e *Entry) shouldUpdateStagedMetadatasWithLock(sm metadata.StagedMetadata) bool {
	if e.cutoverNanos > sm.CutoverNanos {
		e.metrics.untimed.staleMetadata.Inc(1)
		return false
	}
	if e.cutoverNanos < sm.CutoverNanos {
		return true
	}
	bs := bitset.New(uint(len(e.aggregations)))
	for _, pipeline := range sm.Pipelines {
		storagePolicies := e.storagePolicies(pipeline.StoragePolicies)
		for _, storagePolicy := range storagePolicies {
			key := aggregationKey{
				aggregationID:      pipeline.AggregationID,
				storagePolicy:      storagePolicy,
				pipeline:           pipeline.Pipeline,
				idPrefixSuffixType: WithPrefixWithSuffix,
			}
			idx := e.aggregations.index(key)
			if idx < 0 {
				return true
			}
			bs.Set(uint(idx))
		}
	}
	return !bs.All(uint(len(e.aggregations)))
}
NB: The metadata passed in is guaranteed to have cut over based on the current time.
package aggregator
import (
	"container/list"
	"errors"
	"sync"
	"sync/atomic"
	"time"
	"github.com/m3db/m3aggregator/bitset"
	"github.com/m3db/m3aggregator/rate"
	"github.com/m3db/m3aggregator/runtime"
	"github.com/m3db/m3metrics/aggregation"
	"github.com/m3db/m3metrics/metadata"
	"github.com/m3db/m3metrics/metric"
	"github.com/m3db/m3metrics/metric/aggregated"
	metricid "github.com/m3db/m3metrics/metric/id"
	"github.com/m3db/m3metrics/metric/unaggregated"
	"github.com/m3db/m3metrics/policy"
	xerrors "github.com/m3db/m3x/errors"
	"github.com/uber-go/tally"
)
const (
	initialAggregationCapacity = 2
)
var (
	errEntryClosed                 = errors.New("entry is closed")
	errWriteValueRateLimitExceeded = errors.New("write value rate limit is exceeded")
	errEmptyMetadatas              = errors.New("empty metadata list")
	errNoApplicableMetadata        = errors.New("no applicable metadata")
	errNoPipelinesInMetadata       = errors.New("no pipelines in metadata")
	errTooFarInTheFuture           = errors.New("too far in the future")
	errTooFarInThePast             = errors.New("too far in the past")
	errArrivedTooLate              = errors.New("arrived too late")
)
type rateLimitEntryMetrics struct {
	valueRateLimitExceeded tally.Counter
	droppedValues          tally.Counter
}
func newRateLimitEntryMetrics(scope tally.Scope) rateLimitEntryMetrics {
	return rateLimitEntryMetrics{
		valueRateLimitExceeded: scope.Counter("value-rate-limit-exceeded"),
		droppedValues:          scope.Counter("dropped-values"),
	}
}
type untimedEntryMetrics struct {
	rateLimit               rateLimitEntryMetrics
	emptyMetadatas          tally.Counter
	noApplicableMetadata    tally.Counter
	noPipelinesInMetadata   tally.Counter
	emptyPipeline           tally.Counter
	noAggregationInPipeline tally.Counter
	staleMetadata           tally.Counter
	tombstonedMetadata      tally.Counter
	metadatasUpdates        tally.Counter
}
func newUntimedEntryMetrics(scope tally.Scope) untimedEntryMetrics {
	return untimedEntryMetrics{
		rateLimit:               newRateLimitEntryMetrics(scope),
		emptyMetadatas:          scope.Counter("empty-metadatas"),
		noApplicableMetadata:    scope.Counter("no-applicable-metadata"),
		noPipelinesInMetadata:   scope.Counter("no-pipelines-in-metadata"),
		emptyPipeline:           scope.Counter("empty-pipeline"),
		noAggregationInPipeline: scope.Counter("no-aggregation-in-pipeline"),
		staleMetadata:           scope.Counter("stale-metadata"),
		tombstonedMetadata:      scope.Counter("tombstoned-metadata"),
		metadatasUpdates:        scope.Counter("metadatas-updates"),
	}
}
type timedEntryMetrics struct {
	rateLimit         rateLimitEntryMetrics
	tooFarInTheFuture tally.Counter
	tooFarInThePast   tally.Counter
	metadataUpdates   tally.Counter
}
func newTimedEntryMetrics(scope tally.Scope) timedEntryMetrics {
	return timedEntryMetrics{
		rateLimit:         newRateLimitEntryMetrics(scope),
		tooFarInTheFuture: scope.Counter("too-far-in-the-future"),
		tooFarInThePast:   scope.Counter("too-far-in-the-past"),
		metadataUpdates:   scope.Counter("metadata-updates"),
	}
}
type forwardedEntryMetrics struct {
	rateLimit        rateLimitEntryMetrics
	arrivedTooLate   tally.Counter
	duplicateSources tally.Counter
	metadataUpdates  tally.Counter
}
func newForwardedEntryMetrics(scope tally.Scope) forwardedEntryMetrics {
	return forwardedEntryMetrics{
		rateLimit:        newRateLimitEntryMetrics(scope),
		arrivedTooLate:   scope.Counter("arrived-too-late"),
		duplicateSources: scope.Counter("duplicate-sources"),
		metadataUpdates:  scope.Counter("metadata-updates"),
	}
}
type entryMetrics struct {
	untimed   untimedEntryMetrics
	timed     timedEntryMetrics
	forwarded forwardedEntryMetrics
}
func newEntryMetrics(scope tally.Scope) entryMetrics {
	untimedEntryScope := scope.Tagged(map[string]string{"entry-type": "untimed"})
	timedEntryScope := scope.Tagged(map[string]string{"entry-type": "timed"})
	forwardedEntryScope := scope.Tagged(map[string]string{"entry-type": "forwarded"})
	return entryMetrics{
		untimed:   newUntimedEntryMetrics(untimedEntryScope),
		timed:     newTimedEntryMetrics(timedEntryScope),
		forwarded: newForwardedEntryMetrics(forwardedEntryScope),
	}
}
type Entry struct {
	sync.RWMutex
	closed              bool
	opts                Options
	rateLimiter         *rate.Limiter
	hasDefaultMetadatas bool
	cutoverNanos        int64
	lists               *metricLists
	numWriters          int32
	lastAccessNanos     int64
	aggregations        aggregationValues
	metrics             entryMetrics
	decompressor aggregation.IDDecompressor
}
func NewEntry(lists *metricLists, runtimeOpts runtime.Options, opts Options) *Entry {
	scope := opts.InstrumentOptions().MetricsScope().SubScope("entry")
	e := &Entry{
		aggregations: make(aggregationValues, 0, initialAggregationCapacity),
		metrics:      newEntryMetrics(scope),
		decompressor: aggregation.NewPooledIDDecompressor(opts.AggregationTypesOptions().TypesPool()),
	}
	e.ResetSetData(lists, runtimeOpts, opts)
	return e
}
func (e *Entry) IncWriter() { atomic.AddInt32(&e.numWriters, 1) }
func (e *Entry) DecWriter() { atomic.AddInt32(&e.numWriters, -1) }
func (e *Entry) ResetSetData(lists *metricLists, runtimeOpts runtime.Options, opts Options) {
	e.Lock()
	e.closed = false
	e.opts = opts
	e.resetRateLimiterWithLock(runtimeOpts)
	e.hasDefaultMetadatas = false
	e.cutoverNanos = uninitializedCutoverNanos
	e.lists = lists
	e.numWriters = 0
	e.recordLastAccessed(e.opts.ClockOptions().NowFn()())
	e.Unlock()
}
func (e *Entry) SetRuntimeOptions(opts runtime.Options) {
	e.Lock()
	if e.closed {
		e.Unlock()
		return
	}
	e.resetRateLimiterWithLock(opts)
	e.Unlock()
}
func (e *Entry) AddUntimed(
	metricUnion unaggregated.MetricUnion,
	metadatas metadata.StagedMetadatas,
) error {
	switch metricUnion.Type {
	case metric.TimerType:
		var err error
		if err = e.applyValueRateLimit(
			int64(len(metricUnion.BatchTimerVal)),
			e.metrics.untimed.rateLimit,
		); err == nil {
			err = e.writeBatchTimerWithMetadatas(metricUnion, metadatas)
		}
		if metricUnion.BatchTimerVal != nil && metricUnion.TimerValPool != nil {
			metricUnion.TimerValPool.Put(metricUnion.BatchTimerVal)
		}
		return err
	default:
		if err := e.applyValueRateLimit(1, e.metrics.untimed.rateLimit); err != nil {
			return err
		}
		return e.addUntimed(metricUnion, metadatas)
	}
}
func (e *Entry) AddTimed(
	metric aggregated.Metric,
	metadata metadata.TimedMetadata,
) error {
	if err := e.applyValueRateLimit(1, e.metrics.timed.rateLimit); err != nil {
		return err
	}
	return e.addTimed(metric, metadata)
}
func (e *Entry) AddForwarded(
	metric aggregated.ForwardedMetric,
	metadata metadata.ForwardMetadata,
) error {
	if err := e.applyValueRateLimit(1, e.metrics.forwarded.rateLimit); err != nil {
		return err
	}
	return e.addForwarded(metric, metadata)
}
func (e *Entry) ShouldExpire(now time.Time) bool {
	e.RLock()
	if e.closed {
		e.RUnlock()
		return false
	}
	e.RUnlock()
	return e.shouldExpire(now)
}
func (e *Entry) TryExpire(now time.Time) bool {
	e.Lock()
	if e.closed {
		e.Unlock()
		return false
	}
	if !e.shouldExpire(now) {
		e.Unlock()
		return false
	}
	e.closed = true
	for i := range e.aggregations {
		e.aggregations[i].elem.Value.(metricElem).MarkAsTombstoned()
		e.aggregations[i] = aggregationValue{}
	}
	e.aggregations = e.aggregations[:0]
	e.lists = nil
	pool := e.opts.EntryPool()
	e.Unlock()
	pool.Put(e)
	return true
}
func (e *Entry) writeBatchTimerWithMetadatas(
	metric unaggregated.MetricUnion,
	metadatas metadata.StagedMetadatas,
) error {
	maxTimerBatchSizePerWrite := e.opts.MaxTimerBatchSizePerWrite()
	if maxTimerBatchSizePerWrite == 0 {
		return e.addUntimed(metric, metadatas)
	}
	var (
		timerValues    = metric.BatchTimerVal
		numTimerValues = len(timerValues)
		start, end     int
	)
	for start = 0; start < numTimerValues; start = end {
		end = start + maxTimerBatchSizePerWrite
		if end > numTimerValues {
			end = numTimerValues
		}
		splitTimer := metric
		splitTimer.BatchTimerVal = timerValues[start:end]
		if err := e.addUntimed(splitTimer, metadatas); err != nil {
			return err
		}
	}
	return nil
}
func (e *Entry) addUntimed(
	metric unaggregated.MetricUnion,
	metadatas metadata.StagedMetadatas,
) error {
	timeLock := e.opts.TimeLock()
	timeLock.RLock()
	currTime := e.opts.ClockOptions().NowFn()()
	e.recordLastAccessed(currTime)
	e.RLock()
	if e.closed {
		e.RUnlock()
		timeLock.RUnlock()
		return errEntryClosed
	}
	hasDefaultMetadatas := metadatas.IsDefault()
	if e.hasDefaultMetadatas && hasDefaultMetadatas {
		err := e.addUntimedWithLock(currTime, metric)
		e.RUnlock()
		timeLock.RUnlock()
		return err
	}
	sm, err := e.activeStagedMetadataWithLock(currTime, metadatas)
	if err != nil {
		e.RUnlock()
		timeLock.RUnlock()
		return err
	}
	if sm.Tombstoned {
		e.RUnlock()
		timeLock.RUnlock()
		e.metrics.untimed.tombstonedMetadata.Inc(1)
		return nil
	}
	if len(sm.Pipelines) == 0 {
		e.RUnlock()
		timeLock.RUnlock()
		e.metrics.untimed.noPipelinesInMetadata.Inc(1)
		return errNoPipelinesInMetadata
	}
	if !e.shouldUpdateStagedMetadatasWithLock(sm) {
		err = e.addUntimedWithLock(currTime, metric)
		e.RUnlock()
		timeLock.RUnlock()
		return err
	}
	e.RUnlock()
	e.Lock()
	if e.closed {
		e.Unlock()
		timeLock.RUnlock()
		return errEntryClosed
	}
	if e.shouldUpdateStagedMetadatasWithLock(sm) {
		if err = e.updateStagedMetadatasWithLock(metric, hasDefaultMetadatas, sm); err != nil {
			e.Unlock()
			timeLock.RUnlock()
			return err
		}
	}
	err = e.addUntimedWithLock(currTime, metric)
	e.Unlock()
	timeLock.RUnlock()
	return err
}
func (e *Entry) activeStagedMetadataWithLock(
	t time.Time,
	metadatas metadata.StagedMetadatas,
) (metadata.StagedMetadata, error) {
	if len(metadatas) == 0 {
		e.metrics.untimed.emptyMetadatas.Inc(1)
		return metadata.DefaultStagedMetadata, errEmptyMetadatas
	}
	timeNanos := t.UnixNano()
	for idx := len(metadatas) - 1; idx >= 0; idx-- {
		if metadatas[idx].CutoverNanos <= timeNanos {
			return metadatas[idx], nil
		}
	}
	e.metrics.untimed.noApplicableMetadata.Inc(1)
	return metadata.DefaultStagedMetadata, errNoApplicableMetadata
}