func (agg *forwardedAggregation) add(key aggregationKey) {
	if idx := agg.index(key); idx >= 0 {
		agg.byKey[idx].totalRefCnt++
		return
	}
	aggregation := forwardedAggregationWithKey{
		key:         key,
		totalRefCnt: 1,
		currRefCnt:  0,
		buckets:     make(forwardedAggregationBuckets, 0, 2),
	}
	agg.byKey = append(agg.byKey, aggregation)
	agg.metrics.added.Inc(1)
}
add adds the aggregation key to the set of aggregations. If the aggregation
key already exists, its ref count is incremented. Otherwise, a new aggregation
bucket is created and added to the set of aggregtaions.
package aggregator
import (
	"errors"
	"fmt"
	"github.com/m3db/m3aggregator/client"
	"github.com/m3db/m3aggregator/hash"
	"github.com/m3db/m3metrics/metadata"
	"github.com/m3db/m3metrics/metric"
	"github.com/m3db/m3metrics/metric/aggregated"
	"github.com/m3db/m3metrics/metric/id"
	xerrors "github.com/m3db/m3x/errors"
	"github.com/uber-go/tally"
)
const (
	initialValueArrayCapacity = 2
)
var (
	errMetricNotFound         = errors.New("metric not found")
	errAggregationKeyNotFound = errors.New("aggregation key not found")
	errForwardedWriterClosed  = errors.New("forwarded metric writer is closed")
)
type writeForwardedMetricFn func(
	key aggregationKey,
	timeNanos int64,
	value float64,
)
type onForwardedAggregationDoneFn func(key aggregationKey) error
type forwardedMetricWriter interface {
	Len() int
	Register(
		metricType metric.Type,
		metricID id.RawID,
		aggKey aggregationKey,
	) (writeForwardedMetricFn, onForwardedAggregationDoneFn, error)
	Unregister(
		metricType metric.Type,
		metricID id.RawID,
		aggKey aggregationKey,
	) error
	Prepare()
	Flush() error
	Close() error
}
type forwardedWriterMetrics struct {
	registerSuccess               tally.Counter
	registerWriterClosed          tally.Counter
	unregisterSuccess             tally.Counter
	unregisterWriterClosed        tally.Counter
	unregisterMetricNotFound      tally.Counter
	unregisterAggregationNotFound tally.Counter
	prepare                       tally.Counter
	flushSuccess                  tally.Counter
	flushErrors                   tally.Counter
}
func newForwardedWriterMetrics(scope tally.Scope) forwardedWriterMetrics {
	registerScope := scope.Tagged(map[string]string{"action": "register"})
	unregisterScope := scope.Tagged(map[string]string{"action": "unregister"})
	prepareScope := scope.Tagged(map[string]string{"action": "prepare"})
	flushScope := scope.Tagged(map[string]string{"action": "flush"})
	return forwardedWriterMetrics{
		registerSuccess: registerScope.Counter("success"),
		registerWriterClosed: registerScope.Tagged(map[string]string{
			"reason": "writer-closed",
		}).Counter("errors"),
		unregisterSuccess: unregisterScope.Counter("success"),
		unregisterWriterClosed: unregisterScope.Tagged(map[string]string{
			"reason": "writer-closed",
		}).Counter("errors"),
		unregisterMetricNotFound: unregisterScope.Tagged(map[string]string{
			"reason": "metric-not-found",
		}).Counter("errors"),
		unregisterAggregationNotFound: unregisterScope.Tagged(map[string]string{
			"reason": "aggregation-not-found",
		}).Counter("errors"),
		prepare:      prepareScope.Counter("prepare"),
		flushSuccess: flushScope.Counter("success"),
		flushErrors:  flushScope.Counter("errors"),
	}
}
type forwardedWriter struct {
	shard  uint32
	client client.AdminClient
	closed             bool
	aggregations       map[idKey]*forwardedAggregation  
	metrics            forwardedWriterMetrics
	aggregationMetrics *forwardedAggregationMetrics
}
func newForwardedWriter(
	shard uint32,
	client client.AdminClient,
	scope tally.Scope,
) forwardedMetricWriter {
	return &forwardedWriter{
		shard:              shard,
		client:             client,
		aggregations:       make(map[idKey]*forwardedAggregation),
		metrics:            newForwardedWriterMetrics(scope),
		aggregationMetrics: newForwardedAggregationMetrics(scope.SubScope("aggregations")),
	}
}
func (w *forwardedWriter) Len() int { return len(w.aggregations) }
func (w *forwardedWriter) Register(
	metricType metric.Type,
	metricID id.RawID,
	aggKey aggregationKey,
) (writeForwardedMetricFn, onForwardedAggregationDoneFn, error) {
	if w.closed {
		w.metrics.registerWriterClosed.Inc(1)
		return nil, nil, errForwardedWriterClosed
	}
	key := newIDKey(metricType, metricID)
	fa, exists := w.aggregations[key]
	if !exists {
		fa = newForwardedAggregation(metricType, metricID, w.shard, w.client, w.aggregationMetrics)
		w.aggregations[key] = fa
	}
	fa.add(aggKey)
	w.metrics.registerSuccess.Inc(1)
	return fa.writeForwardedMetricFn(), fa.onAggregationKeyDoneFn(), nil
}
func (w *forwardedWriter) Unregister(
	metricType metric.Type,
	metricID id.RawID,
	aggKey aggregationKey,
) error {
	if w.closed {
		w.metrics.unregisterWriterClosed.Inc(1)
		return errForwardedWriterClosed
	}
	key := newIDKey(metricType, metricID)
	fa, exists := w.aggregations[key]
	if !exists {
		w.metrics.unregisterMetricNotFound.Inc(1)
		return errMetricNotFound
	}
	remaining, ok := fa.remove(aggKey)
	if !ok {
		w.metrics.unregisterAggregationNotFound.Inc(1)
		return errAggregationKeyNotFound
	}
	if remaining == 0 {
		fa.clear()
		delete(w.aggregations, key)
	}
	w.metrics.unregisterSuccess.Inc(1)
	return nil
}
func (w *forwardedWriter) Prepare() {
	for _, agg := range w.aggregations {
		agg.reset()
	}
	w.metrics.prepare.Inc(1)
}
func (w *forwardedWriter) Flush() error {
	if err := w.client.Flush(); err != nil {
		w.metrics.flushErrors.Inc(1)
		return err
	}
	w.metrics.flushSuccess.Inc(1)
	return nil
}
func (w *forwardedWriter) Close() error {
	if w.closed {
		return errForwardedWriterClosed
	}
	w.closed = true
	w.client = nil
	w.aggregations = nil
	return nil
}
type idKey struct {
	metricType metric.Type
	idHash     hash.Hash128
}
func newIDKey(
	metricType metric.Type,
	metricID id.RawID,
) idKey {
	idHash := hash.Murmur3Hash128(metricID)
	return idKey{
		metricType: metricType,
		idHash:     idHash,
	}
}
type forwardedAggregationBucket struct {
	timeNanos int64
	values    []float64
}
type forwardedAggregationBuckets []forwardedAggregationBucket
type forwardedAggregationWithKey struct {
	key aggregationKey
	totalRefCnt int
	currRefCnt        int
	cachedValueArrays [][]float64
	buckets           forwardedAggregationBuckets
}
func (agg *forwardedAggregationWithKey) reset() {
	agg.currRefCnt = 0
	for i := 0; i < len(agg.buckets); i++ {
		agg.buckets[i].values = agg.buckets[i].values[:0]
		agg.cachedValueArrays = append(agg.cachedValueArrays, agg.buckets[i].values)
		agg.buckets[i].values = nil
	}
	agg.buckets = agg.buckets[:0]
}
func (agg *forwardedAggregationWithKey) add(timeNanos int64, value float64) {
	for i := 0; i < len(agg.buckets); i++ {
		if agg.buckets[i].timeNanos == timeNanos {
			agg.buckets[i].values = append(agg.buckets[i].values, value)
			return
		}
	}
	var values []float64
	if numCachedValueArrays := len(agg.cachedValueArrays); numCachedValueArrays > 0 {
		values = agg.cachedValueArrays[numCachedValueArrays-1]
		values = values[:0]
		agg.cachedValueArrays = agg.cachedValueArrays[:numCachedValueArrays-1]
	} else {
		values = make([]float64, 0, initialValueArrayCapacity)
	}
	values = append(values, value)
	bucket := forwardedAggregationBucket{
		timeNanos: timeNanos,
		values:    values,
	}
	agg.buckets = append(agg.buckets, bucket)
}
type forwardedAggregationMetrics struct {
	added                  tally.Counter
	removed                tally.Counter
	write                  tally.Counter
	onDoneNoWrite          tally.Counter
	onDoneWriteSuccess     tally.Counter
	onDoneWriteErrors      tally.Counter
	onDoneUnexpectedRefCnt tally.Counter
}
func newForwardedAggregationMetrics(scope tally.Scope) *forwardedAggregationMetrics {
	return &forwardedAggregationMetrics{
		added:                  scope.Counter("added"),
		removed:                scope.Counter("removed"),
		write:                  scope.Counter("write"),
		onDoneNoWrite:          scope.Counter("on-done-not-write"),
		onDoneWriteSuccess:     scope.Counter("on-done-write-success"),
		onDoneWriteErrors:      scope.Counter("on-done-write-errors"),
		onDoneUnexpectedRefCnt: scope.Counter("on-done-unexpected-refcnt"),
	}
}
type forwardedAggregation struct {
	metricType metric.Type
	metricID   id.RawID
	shard      uint32
	client     client.AdminClient
	byKey    []forwardedAggregationWithKey
	metrics  *forwardedAggregationMetrics
	writeFn  writeForwardedMetricFn
	onDoneFn onForwardedAggregationDoneFn
}
func newForwardedAggregation(
	metricType metric.Type,
	metricID id.RawID,
	shard uint32,
	client client.AdminClient,
	fm *forwardedAggregationMetrics,
) *forwardedAggregation {
	agg := &forwardedAggregation{
		metricType: metricType,
		metricID:   metricID,
		shard:      shard,
		client:     client,
		byKey:      make([]forwardedAggregationWithKey, 0, 2),
		metrics:    fm,
	}
	agg.writeFn = agg.write
	agg.onDoneFn = agg.onDone
	return agg
}
func (agg *forwardedAggregation) writeForwardedMetricFn() writeForwardedMetricFn {
	return agg.writeFn
}
func (agg *forwardedAggregation) onAggregationKeyDoneFn() onForwardedAggregationDoneFn {
	return agg.onDoneFn
}
func (agg *forwardedAggregation) clear() { *agg = forwardedAggregation{} }
func (agg *forwardedAggregation) reset() {
	for i := 0; i < len(agg.byKey); i++ {
		agg.byKey[i].reset()
	}
}