balancer/rls/internal/adaptive/lookback.go

/*
 *
 * Copyright 2020 gRPC authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

package adaptive

import "time"

// lookback implements a moving sum over an int64 timeline.
type lookback struct {
	bins  int64         // Number of bins to use for lookback.
	width time.Duration // Width of each bin.

	head  int64   // Absolute bin index (time * bins / duration) of the current head bin.
	total int64   // Sum over all the values in buf, within the lookback window behind head.
	buf   []int64 // Ring buffer for keeping track of the sum elements.
}

// newLookback creates a new lookback for the given duration with a set number
// of bins.
func newLookback(bins int64, duration time.Duration) *lookback {
	return &lookback{
		bins:  bins,
		width: duration / time.Duration(bins),
		buf:   make([]int64, bins),
	}
}

// add is used to increment the lookback sum.
func (l *lookback) add(t time.Time, v int64) {
	pos := l.advance(t)

	if (l.head - pos) >= l.bins {
		// Do not increment counters if pos is more than bins behind head.
		return
	}
	l.buf[pos%l.bins] += v
	l.total += v
}

// sum returns the sum of the lookback buffer at the given time or head,
// whichever is greater.
func (l *lookback) sum(t time.Time) int64 {
	l.advance(t)
	return l.total
}

// advance prepares the lookback buffer for calls to add() or sum() at time t.
// If head is greater than t then the lookback buffer will be untouched. The
// absolute bin index corresponding to t is returned. It will always be less
// than or equal to head.
func (l *lookback) advance(t time.Time) int64 {
	ch := l.head                               // Current head bin index.
	nh := t.UnixNano() / l.width.Nanoseconds() // New head bin index.

	if nh <= ch {
		// Either head unchanged or clock jitter (time has moved backwards). Do
		// not advance.
		return nh
	}

	jmax := min(l.bins, nh-ch)
	for j := int64(0); j < jmax; j++ {
		i := (ch + j + 1) % l.bins
		l.total -= l.buf[i]
		l.buf[i] = 0
	}
	l.head = nh
	return nh
}

func min(x int64, y int64) int64 {
	if x < y {
		return x
	}
	return y
}