package obs
import (
"sort"
"strings"
"sync"
"time"
)
// trafficFresh is how long a computed traffic page is served as it is,
// the live view's own tick (web's trafficEvery), so the stream still
// moves every time it asks;
// trafficStale is how long past that it is still served while a
// refresh runs behind it. The page took 0.8-1.5 s to compute from the
// log on the forge, against a 250 ms budget, and its numbers move by the
// minute, not by the request.
const (
trafficFresh = 30 * time.Second
trafficStale = 10 * time.Minute
)
type trafficEntry struct {
t *Traffic
at time.Time
pending bool
}
// trafficMemo keeps the last answer per range and filter set.
type trafficMemo struct {
mu sync.Mutex
m map[string]*trafficEntry
}
func trafficKey(q TrafficQuery) string {
keys := make([]string, 0, len(q.Filters))
for k, v := range q.Filters {
if v != "" {
keys = append(keys, k+"="+v)
}
}
sort.Strings(keys)
return q.Range.Name + "|" + strings.Join(keys, "&")
}
// get answers q from the memo when it can: fresh as it is, stale while
// one refresh runs behind it. compute is the uncached read. A query that
// names its own Now is a question about the past, and is never memoised.
func (m *trafficMemo) get(q TrafficQuery, now time.Time, compute func(TrafficQuery) (*Traffic, error)) (*Traffic, error) {
if !q.Now.IsZero() {
return compute(q)
}
key := trafficKey(q)
m.mu.Lock()
if m.m == nil {
m.m = map[string]*trafficEntry{}
}
e := m.m[key]
switch {
case e != nil && now.Sub(e.at) < trafficFresh:
t := e.t
m.mu.Unlock()
return t, nil
case e != nil && now.Sub(e.at) < trafficFresh+trafficStale:
t := e.t
if !e.pending {
e.pending = true
go m.refresh(key, q, compute)
}
m.mu.Unlock()
return t, nil
}
m.mu.Unlock()
t, err := compute(q)
if err != nil {
return nil, err
}
m.mu.Lock()
m.m[key] = &trafficEntry{t: t, at: time.Now()}
m.mu.Unlock()
return t, nil
}
func (m *trafficMemo) refresh(key string, q TrafficQuery, compute func(TrafficQuery) (*Traffic, error)) {
t, err := compute(q)
m.mu.Lock()
defer m.mu.Unlock()
e := m.m[key]
if err != nil {
if e != nil {
e.pending = false
}
return
}
m.m[key] = &trafficEntry{t: t, at: time.Now()}
}
// trafficWarmEvery is how often WarmTraffic recomputes each range: inside
// trafficFresh+trafficStale, so the unfiltered page is always answered
// from the memo, the first view after a restart included.
const trafficWarmEvery = 4 * time.Minute
// WarmTraffic computes every named range, unfiltered, now and then
// every trafficWarmEvery until stop closes. The traffic page took 1-1.8 s
// to compute on its first view of each range; this pays that in the
// background, one range at a time.
func (r *Recorder) WarmTraffic(stop <-chan struct{}) {
warm := func() {
for _, rg := range TrafficRanges {
select {
case <-stop:
return
default:
}
r.mu.Lock()
sink := r.sink
r.mu.Unlock()
t, ok := sink.(interface {
Traffic(TrafficQuery) (*Traffic, error)
})
if !ok {
return
}
q := TrafficQuery{Range: rg}
if res, err := t.Traffic(q); err == nil {
r.traffic.put(q, res)
}
}
}
warm()
tick := time.NewTicker(trafficWarmEvery)
defer tick.Stop()
for {
select {
case <-stop:
return
case <-tick.C:
warm()
}
}
}
// put stores a computed answer for q.
func (m *trafficMemo) put(q TrafficQuery, t *Traffic) {
m.mu.Lock()
defer m.mu.Unlock()
if m.m == nil {
m.m = map[string]*trafficEntry{}
}
m.m[trafficKey(q)] = &trafficEntry{t: t, at: time.Now()}
}