<commit_before>package cache

import (
	"context"
	"flag"
	"runtime"
	"sync"

	"github.com/grafana/metrictank/mdata/cache/accnt"
	"github.com/grafana/metrictank/mdata/chunk"
	"github.com/grafana/metrictank/stats"
	"github.com/grafana/metrictank/tracing"
	opentracing "github.com/opentracing/opentracing-go"
	"github.com/raintank/worldping-api/pkg/log"
	"github.com/rakyll/globalconf"
	"gopkg.in/raintank/schema.v1"
)

var (
	maxSize      uint64
	searchFwdBug = stats.NewCounter32("recovered_errors.cache.metric.searchForwardBug")
)

func init() {
	flags := flag.NewFlagSet("chunk-cache", flag.ExitOnError)
	// (1024 ^ 3) * 4 = 4294967296 = 4G
	flags.Uint64Var(&maxSize, "max-size", 4294967296, "Maximum size of chunk cache in bytes")
	globalconf.Register("chunk-cache", flags)
}

type CCache struct {
	sync.RWMutex

	// one CCacheMetric struct per metric key, indexed by the key
	metricCache map[schema.AMKey]*CCacheMetric

	// sets of metric keys, indexed by their raw metric keys
	metricRawKeys map[schema.MKey]map[schema.Archive]struct{}

	// accounting for the cache. keeps track of when data needs to be evicted
	// and what should be evicted
	accnt accnt.Accnt

	// channel that's only used to signal go routines to stop
	stop chan interface{}

	tracer opentracing.Tracer
}

func NewCCache() *CCache {
	cc := &CCache{
		metricCache:   make(map[schema.AMKey]*CCacheMetric),
		metricRawKeys: make(map[schema.MKey]map[schema.Archive]struct{}),
		accnt:         accnt.NewFlatAccnt(maxSize),
		stop:          make(chan interface{}),
		tracer:        opentracing.NoopTracer{},
	}
	go cc.evictLoop()
	return cc
}

func (c *CCache) SetTracer(t opentracing.Tracer) {
	c.tracer = t
}

func (c *CCache) evictLoop() {
	evictQ := c.accnt.GetEvictQ()
	for {
		select {
		case target := <-evictQ:
			c.evict(target)
		case _ = <-c.stop:
			return
		}
	}
}

// takes a raw key and deletes all archives associated with it from cache
func (c *CCache) DelMetric(rawMetric schema.MKey) (int, int) {
	archives, series := 0, 0

	c.Lock()
	defer c.Unlock()

	archs, ok := c.metricRawKeys[rawMetric]
	if !ok {
		return archives, series
	}

	metric := schema.AMKey{MKey: rawMetric}
	for arch := range archs {
		metric.Archive = arch
		delete(c.metricCache, metric)
		c.accnt.DelMetric(metric)
		archives++
	}

	delete(c.metricRawKeys, rawMetric)
	series++

	return series, archives
}

// adds the given chunk to the cache, but only if the metric is sufficiently hot
func (c *CCache) CacheIfHot(metric schema.AMKey, prev uint32, itergen chunk.IterGen) {
	c.RLock()

	var met *CCacheMetric
	var ok bool

	// if this metric is not cached at all it is not hot
	if met, ok = c.metricCache[metric]; !ok {
		c.RUnlock()
		return
	}

	// if the previous chunk is not cached we consider the metric not hot enough to cache this chunk
	// only works reliably if the last chunk of that metric is span aware, otherwise lastTs() will be guessed
	// conservatively which means that the returned value will probably be lower than the real last ts
	if met.lastTs() < itergen.Ts {
		c.RUnlock()
		return
	}

	accnt.CacheChunkPushHot.Inc()

	c.RUnlock()
	met.Add(prev, itergen)
}

func (c *CCache) Add(metric schema.AMKey, prev uint32, itergen chunk.IterGen) {
	c.Lock()
	defer c.Unlock()

	ccm, ok := c.metricCache[metric]
	if !ok {
		ccm = NewCCacheMetric()
		ccm.Init(metric.MKey, prev, itergen)
		c.metricCache[metric] = ccm

		// if we do not have this raw key yet, create the entry with the association
		ccms, ok := c.metricRawKeys[metric.MKey]
		if !ok {
			c.metricRawKeys[metric.MKey] = map[schema.Archive]struct{}{
				metric.Archive: struct{}{},
			}
		} else {
			// otherwise, make sure the association exists
			ccms[metric.Archive] = struct{}{}
		}
	} else {
		ccm.Add(prev, itergen)
	}

	c.accnt.AddChunk(metric, itergen.Ts, itergen.Size())
}

func (cc *CCache) Reset() (int, int) {
	cc.Lock()
	cc.accnt.Reset()
	series := len(cc.metricRawKeys)
	archives := len(cc.metricCache)
	cc.metricCache = make(map[schema.AMKey]*CCacheMetric)<commit_msg>ccache: gofmt -w -s<commit_after>package cache

import (
	"context"
	"flag"
	"runtime"
	"sync"

	"github.com/grafana/metrictank/mdata/cache/accnt"
	"github.com/grafana/metrictank/mdata/chunk"
	"github.com/grafana/metrictank/stats"
	"github.com/grafana/metrictank/tracing"
	opentracing "github.com/opentracing/opentracing-go"
	"github.com/raintank/worldping-api/pkg/log"
	"github.com/rakyll/globalconf"
	"gopkg.in/raintank/schema.v1"
)

var (
	maxSize      uint64
	searchFwdBug = stats.NewCounter32("recovered_errors.cache.metric.searchForwardBug")
)

func init() {
	flags := flag.NewFlagSet("chunk-cache", flag.ExitOnError)
	// (1024 ^ 3) * 4 = 4294967296 = 4G
	flags.Uint64Var(&maxSize, "max-size", 4294967296, "Maximum size of chunk cache in bytes")
	globalconf.Register("chunk-cache", flags)
}

type CCache struct {
	sync.RWMutex

	// one CCacheMetric struct per metric key, indexed by the key
	metricCache map[schema.AMKey]*CCacheMetric

	// sets of metric keys, indexed by their raw metric keys
	metricRawKeys map[schema.MKey]map[schema.Archive]struct{}

	// accounting for the cache. keeps track of when data needs to be evicted
	// and what should be evicted
	accnt accnt.Accnt

	// channel that's only used to signal go routines to stop
	stop chan interface{}

	tracer opentracing.Tracer
}

func NewCCache() *CCache {
	cc := &CCache{
		metricCache:   make(map[schema.AMKey]*CCacheMetric),
		metricRawKeys: make(map[schema.MKey]map[schema.Archive]struct{}),
		accnt:         accnt.NewFlatAccnt(maxSize),
		stop:          make(chan interface{}),
		tracer:        opentracing.NoopTracer{},
	}
	go cc.evictLoop()
	return cc
}

func (c *CCache) SetTracer(t opentracing.Tracer) {
	c.tracer = t
}

func (c *CCache) evictLoop() {
	evictQ := c.accnt.GetEvictQ()
	for {
		select {
		case target := <-evictQ:
			c.evict(target)
		case _ = <-c.stop:
			return
		}
	}
}

// takes a raw key and deletes all archives associated with it from cache
func (c *CCache) DelMetric(rawMetric schema.MKey) (int, int) {
	archives, series := 0, 0

	c.Lock()
	defer c.Unlock()

	archs, ok := c.metricRawKeys[rawMetric]
	if !ok {
		return archives, series
	}

	metric := schema.AMKey{MKey: rawMetric}
	for arch := range archs {
		metric.Archive = arch
		delete(c.metricCache, metric)
		c.accnt.DelMetric(metric)
		archives++
	}

	delete(c.metricRawKeys, rawMetric)
	series++

	return series, archives
}

// adds the given chunk to the cache, but only if the metric is sufficiently hot
func (c *CCache) CacheIfHot(metric schema.AMKey, prev uint32, itergen chunk.IterGen) {
	c.RLock()

	var met *CCacheMetric
	var ok bool

	// if this metric is not cached at all it is not hot
	if met, ok = c.metricCache[metric]; !ok {
		c.RUnlock()
		return
	}

	// if the previous chunk is not cached we consider the metric not hot enough to cache this chunk
	// only works reliably if the last chunk of that metric is span aware, otherwise lastTs() will be guessed
	// conservatively which means that the returned value will probably be lower than the real last ts
	if met.lastTs() < itergen.Ts {
		c.RUnlock()
		return
	}

	accnt.CacheChunkPushHot.Inc()

	c.RUnlock()
	met.Add(prev, itergen)
}

func (c *CCache) Add(metric schema.AMKey, prev uint32, itergen chunk.IterGen) {
	c.Lock()
	defer c.Unlock()

	ccm, ok := c.metricCache[metric]
	if !ok {
		ccm = NewCCacheMetric()
		ccm.Init(metric.MKey, prev, itergen)
		c.metricCache[metric] = ccm

		// if we do not have this raw key yet, create the entry with the association
		ccms, ok := c.metricRawKeys[metric.MKey]
		if !ok {
			c.metricRawKeys[metric.MKey] = map[schema.Archive]struct{}{
				metric.Archive: {},
			}
		} else {
			// otherwise, make sure the association exists
			ccms[metric.Archive] = struct{}{}
		}
	} else {
		ccm.Add(prev, itergen)
	}

	c.accnt.AddChunk(metric, itergen.Ts, itergen.Size())
}

func (cc *CCache) Reset() (int, int) {
	cc.Lock()
	cc.accnt.Reset()
	series := len(cc.metricRawKeys)
	archives := len(cc.metricCache)
	cc.metricCache = make(map[schema.AMKey]*CCacheMetric)
{'fraction_non_alphanumeric': 0.12369467926404774, 'fraction_numerical': 0.009945300845350571, 'mean_word_length': 3.8231414868105515, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2603295', 'n_tokens_mistral': 3117, 'n_tokens_neox': 2917, 'n_words': 781}