Add method to cache many chunks at once

Author: replay

mdata/cache/ccache_metric.go

@@ -69,6 +69,66 @@ func (mc *CCacheMetric) Del(ts uint32) int {
 	return len(mc.chunks)
 }
 
+func (mc *CCacheMetric) AddSlice(prev uint32, itergens []chunk.IterGen) {
+	if len(itergens) == 0 {
+		return
+	}
+
+	mc.Lock()
+	defer mc.Unlock()
+
+	ts := itergens[0].Ts
+
+	// if previous chunk has not been passed we try to be smart and figure it out.
+	// this is common in a scenario where a metric continuously gets queried
+	// for a range that starts less than one chunkspan before now().
+	if prev == 0 {
+		res, ok := mc.seekDesc(ts - 1)
+		if ok {
+			prev = res
+		}
+	}
+
+	for _, itergen := range itergens {
+		ts = itergen.Ts
+
+		if _, ok := mc.chunks[ts]; ok {
+			// chunk is already present. no need to error on that, just ignore it
+			continue
+		}
+
+		mc.chunks[ts] = &CCacheChunk{
+			Ts:    ts,
+			Prev:  0,
+			Next:  0,
+			Itgen: itergen,
+		}
+
+		// if the previous chunk is cached, link in both directions
+		if _, ok := mc.chunks[prev]; ok {
+			mc.chunks[prev].Next = ts
+			mc.chunks[ts].Prev = prev
+		}
+		prev = ts
+	}
+
+	nextTs := mc.nextTs(ts)
+
+	// if nextTs() can't figure out the end date it returns ts
+	if nextTs > ts {
+		// if the next chunk is cached, link in both directions
+		if _, ok := mc.chunks[nextTs]; ok {
+			mc.chunks[nextTs].Prev = ts
+			mc.chunks[ts].Next = nextTs
+		}
+	}
+
+	// regenerate the list of sorted keys after adding a chunk
+	mc.generateKeys()
+
+	return
+}
+
 func (mc *CCacheMetric) Add(prev uint32, itergen chunk.IterGen) {
 	ts := itergen.Ts
 

mdata/cache/ccache_metric_test.go

@@ -0,0 +1,47 @@
+package cache
+
+import (
+	"testing"
+
+	"github.com/grafana/metrictank/mdata/chunk"
+	"gopkg.in/raintank/schema.v1"
+)
+
+func generateChunks(b *testing.B) []chunk.IterGen {
+	res := make([]chunk.IterGen, 0, b.N)
+
+	// the metric is initialized with the first chunk at ts 1,
+	// so the generated slice should start at ts 2
+	for i := uint32(2); i < uint32(b.N+2); i++ {
+		c := getItgen(b, []uint32{1}, i, true)
+		res = append(res, c)
+	}
+	return res
+}
+
+func initMetric(b *testing.B) *CCacheMetric {
+	ccm := NewCCacheMetric()
+	mkey, _ := schema.MKeyFromString("1.12345678901234567890123456789012")
+	c := getItgen(b, []uint32{1}, uint32(1), true)
+	ccm.Init(mkey, 0, c)
+	return ccm
+}
+
+func BenchmarkAddingManyChunksOneByOne(b *testing.B) {
+	ccm := initMetric(b)
+	chunks := generateChunks(b)
+	prev := uint32(1)
+	b.ResetTimer()
+	for _, chunk := range chunks {
+		ccm.Add(prev, chunk)
+		prev = chunk.Ts
+	}
+}
+
+func BenchmarkAddingManyChunksAtOnce(b *testing.B) {
+	ccm := initMetric(b)
+	chunks := generateChunks(b)
+	prev := uint32(1)
+	b.ResetTimer()
+	ccm.AddSlice(prev, chunks)
+}

mdata/cache/ccache_test.go

@@ -11,7 +11,7 @@ import (
 )
 
 // getItgen returns an IterGen which holds a chunk which has directly encoded all values
-func getItgen(t *testing.T, values []uint32, ts uint32, spanaware bool) chunk.IterGen {
+func getItgen(t testing.TB, values []uint32, ts uint32, spanaware bool) chunk.IterGen {
 	var b []byte
 	buf := new(bytes.Buffer)
 	if spanaware {