func (tsaggr *TimeslotAggregator) signal() {
	tsaggr.sigCh <- struct{}{}  
	tsaggr.sigCh <- struct{}{}  
}
signal wakes up the processor routine
to advance the timeslot, if needed. Only
the processor can do this, because its the
only writer to the data structure
package load
import (
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"github.com/uber-common/bark"
	"github.com/uber/cherami-server/common"
)
type (
	metricData struct {
		hostID string
		groupTag string
		metricName MetricName
		metricValue int64
		timestamp int64
	}
	timeslotData struct {
		n   int64  
		sum int64  
	}
	timeslot struct {
		beginTime int64                     
		data      map[string]*timeslotData  
	}
	TimeslotAggregator struct {
		currIdx   int64  
		started   int64
		timeslots [historySize]timeslot  
		dataCh    chan *metricData       
		sigCh     chan struct{}          
		stopCh    chan struct{}
		stopWG    sync.WaitGroup
		clock     common.TimeSource
		logger    bark.Logger
	}
)
const historySize = 10  
const maxClockSkew = int64(time.Second * 5)
const aggregationPeriod = int64(time.Minute)  
const dataChannelSize = 256                   
func NewTimeSlotAggregator(clock common.TimeSource, logger bark.Logger) *TimeslotAggregator {
	aggr := new(TimeslotAggregator)
	aggr.clock = clock
	aggr.logger = logger
	aggr.stopCh = make(chan struct{})
	aggr.sigCh = make(chan struct{})
	aggr.dataCh = make(chan *metricData, dataChannelSize)
	return aggr
}
func (tsaggr *TimeslotAggregator) Start() {
	if !atomic.CompareAndSwapInt64(&tsaggr.started, 0, 1) {
		return
	}
	tsaggr.stopWG.Add(1)
	go tsaggr.processor()
	tsaggr.logger.Infof("Timeslot metrics aggregator started")
}
func (tsaggr *TimeslotAggregator) Stop() {
	close(tsaggr.stopCh)
	if !common.AwaitWaitGroup(&tsaggr.stopWG, time.Second) {
		tsaggr.logger.Warn("Timed out waiting for timeslotAggregator to stop")
	}
	tsaggr.logger.Infof("Timeslot metrics aggregator stopped")
}
func (tsaggr *TimeslotAggregator) Put(hostID string, groupTag string, metricName MetricName, metricValue int64, timestamp int64) error {
	data := new(metricData)
	data.hostID = hostID
	data.groupTag = groupTag
	data.metricName = metricName
	data.metricValue = metricValue
	data.timestamp = timestamp
	select {
	case tsaggr.dataCh <- data:
		return nil
	default:
		return ErrBackpressure
	}
}
func (tsaggr *TimeslotAggregator) Get(hostID string, groupTag string, metricName MetricName, resultType AggregateType) (int64, error) {
	now := tsaggr.clock.Now().UnixNano()
	currIdx := atomic.LoadInt64(&tsaggr.currIdx)
	currSlot := &tsaggr.timeslots[currIdx]
	if tsaggr.isTimeslotExpired(currSlot, now) {
		tsaggr.signal()  
		currIdx = atomic.LoadInt64(&tsaggr.currIdx)
		currSlot = &tsaggr.timeslots[currIdx]
	}
	currIdx = prevSlotIdx(currIdx)
	currSlot = &tsaggr.timeslots[currIdx]
	oldestDataTime := tsaggr.getOldestDataTimeForAggrType(resultType, now)
	key := buildKey(hostID, groupTag, metricName)
	minSlots := tsaggr.minSlotsForAggrType(resultType)
	maxSlots := tsaggr.maxSlotsForAggrType(resultType)
	var nSlots int
	var result timeslotData
	for nSlots < maxSlots {
		slotEndTime := currSlot.beginTime + aggregationPeriod
		if slotEndTime <= oldestDataTime {
			break  
		}
		if d, ok := currSlot.data[key]; ok {
			result.n += d.n
			result.sum += d.sum
			nSlots++
		}
		currIdx = prevSlotIdx(currIdx)
		currSlot = &tsaggr.timeslots[currIdx]
	}
	if nSlots < minSlots {
		return 0, ErrNoData
	}
	if resultType == OneMinSum || resultType == FiveMinSum {
		return result.sum, nil
	}
	if result.n < tsaggr.minDataPointsForAggrType(resultType) {
		return 0, ErrNoData
	}
	return result.sum / result.n, nil
}
func (tsaggr *TimeslotAggregator) processor() {
	defer tsaggr.stopWG.Done()
	for {
		select {
		case data := <-tsaggr.dataCh:
			tsaggr.add(data)
		case <-tsaggr.sigCh:
			tsaggr.advanceTimeslotIfExpired(tsaggr.clock.Now().UnixNano())
		case <-tsaggr.stopCh:
			return
		}
	}
}
func (tsaggr *TimeslotAggregator) add(data *metricData) {
	now := tsaggr.clock.Now().UnixNano()
	diff := now - data.timestamp
	if diff > maxClockSkew {
		return
	}
	currSlot := tsaggr.advanceTimeslotIfExpired(now)
	if getBeginTime(currSlot) == 0 {
		setBeginTime(currSlot, now)
		currSlot.data = make(map[string]*timeslotData)
	}
	key := buildKey(data.hostID, data.groupTag, data.metricName)
	tsdata, ok := currSlot.data[key]
	if !ok {
		tsdata = new(timeslotData)
		currSlot.data[key] = tsdata
	}
	tsdata.n++
	tsdata.sum += data.metricValue
}
func (tsaggr *TimeslotAggregator) advanceTimeslotIfExpired(now int64) *timeslot {
	currIdx := atomic.LoadInt64(&tsaggr.currIdx)
	currSlot := &tsaggr.timeslots[currIdx]
	if tsaggr.isTimeslotExpired(currSlot, now) {
		currIdx = nextSlotIdx(currIdx)
		atomic.StoreInt64(&tsaggr.currIdx, currIdx)
		currSlot = &tsaggr.timeslots[currIdx]
		setBeginTime(currSlot, 0)
	}
	return currSlot
}