optimize AddRange

BEFORE:

~/g/s/g/g/m/m/cache ❯❯❯ taskset --cpu-list 5 go test -bench  . -benchmem
goos: linux
goarch: amd64
pkg: github.com/grafana/metrictank/mdata/cache
BenchmarkAddingManyChunksOneByOne 	   10000	   1225701 ns/op	   21688 B/op	       5 allocs/op
BenchmarkAddingManyChunksAtOnce   	 1000000	      1146 ns/op	     118 B/op	       1 allocs/op
PASS
ok  	github.com/grafana/metrictank/mdata/cache	14.713s

AFTER:

~/g/s/g/g/m/m/cache ❯❯❯ taskset --cpu-list 5 go test -bench  . -benchmem
goos: linux
goarch: amd64
pkg: github.com/grafana/metrictank/mdata/cache
BenchmarkAddingManyChunksOneByOne 	   10000	   1226088 ns/op	   21688 B/op	       5 allocs/op
BenchmarkAddingManyChunksAtOnce   	 1000000	      1043 ns/op	     118 B/op	       1 allocs/op
PASS
ok  	github.com/grafana/metrictank/mdata/cache	14.620s

Author: Dieterbe

mdata/cache/ccache_metric.go

@@ -78,10 +78,17 @@ func (mc *CCacheMetric) AddRange(prev uint32, itergens []chunk.IterGen) {
 		return
 	}
 
+	if len(itergens) == 1 {
+		mc.Add(prev, itergens[0])
+		return
+	}
+
 	mc.Lock()
 	defer mc.Unlock()
 
-	ts := itergens[0].Ts
+	// handle the first one
+	itergen := itergens[0]
+	ts := itergen.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
@@ -93,46 +100,74 @@ func (mc *CCacheMetric) AddRange(prev uint32, itergens []chunk.IterGen) {
 		}
 	}
 
-	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
-		}
+	// if the previous chunk is cached, link it
+	if _, ok := mc.chunks[prev]; ok {
+		mc.chunks[prev].Next = ts
+	} else {
+		prev = 0
+	}
 
+	// add chunk if we don't have it yet (most likely)
+	if _, ok := mc.chunks[ts]; !ok {
 		mc.chunks[ts] = &CCacheChunk{
 			Ts:    ts,
-			Prev:  0,
-			Next:  0,
+			Prev:  prev,
+			Next:  itergens[1].Ts,
 			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
+
+	// handle the 2nd until the last-but-one
+	for i := 1; i < len(itergens)-1; i++ {
+		itergen := itergens[i]
+		ts := itergen.Ts
+		// add chunk if we don't have it yet (most likely)
+		if _, ok := mc.chunks[ts]; !ok {
+			mc.chunks[ts] = &CCacheChunk{
+				Ts:    ts,
+				Prev:  prev,
+				Next:  itergens[i+1].Ts,
+				Itgen: itergen,
+			}
 		}
 		prev = ts
 	}
 
-	nextTs := mc.nextTs(ts)
+	// handle the last one
+	itergen = itergens[len(itergens)-1]
+	ts = itergen.Ts
 
 	// if nextTs() can't figure out the end date it returns ts
-	if nextTs > ts {
+	next := mc.nextTsCore(itergen, ts, prev, 0)
+	if next == ts {
+		next = 0
+	} else {
 		// 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
+		if _, ok := mc.chunks[next]; ok {
+			mc.chunks[next].Prev = ts
+		} else {
+			next = 0
 		}
 	}
 
+	// add chunk if we don't have it yet (most likely)
+	if _, ok := mc.chunks[ts]; !ok {
+		mc.chunks[ts] = &CCacheChunk{
+			Ts:    ts,
+			Prev:  prev,
+			Next:  next,
+			Itgen: itergen,
+		}
+	}
 	// regenerate the list of sorted keys after adding a chunk
 	mc.generateKeys()
 
 	return
 }
 
+// Add adds a chunk to the cache
 func (mc *CCacheMetric) Add(prev uint32, itergen chunk.IterGen) {
 	ts := itergen.Ts
 
@@ -201,24 +236,29 @@ func (mc *CCacheMetric) generateKeys() {
 // assumes we already have at least a read lock
 func (mc *CCacheMetric) nextTs(ts uint32) uint32 {
 	chunk := mc.chunks[ts]
-	span := chunk.Itgen.Span
+	return mc.nextTsCore(chunk.Itgen, chunk.Ts, chunk.Prev, chunk.Next)
+}
+
+// nextTsCore returns the ts of the next chunk, given a chunks key properties
+// (to the extent we know them). It guesses if necessary.
+// assumes we already have at least a read lock
+func (mc *CCacheMetric) nextTsCore(itgen chunk.IterGen, ts, prev, next uint32) uint32 {
+	span := itgen.Span
 	if span > 0 {
 		// if the chunk is span-aware we don't need anything else
-		return chunk.Ts + span
+		return ts + span
 	}
 
-	if chunk.Next == 0 {
-		if chunk.Prev == 0 {
-			// if a chunk has no next and no previous chunk we have to assume it's length is 0
-			return chunk.Ts
-		} else {
-			// if chunk has no next chunk, but has a previous one, we assume the length of this one is same as the previous one
-			return chunk.Ts + (chunk.Ts - chunk.Prev)
-		}
-	} else {
-		// if chunk has a next chunk, then that's the ts we need
-		return chunk.Next
+	// if chunk has a next chunk, then that's the ts we need
+	if next != 0 {
+		return next
+	}
+	// if chunk has no next chunk, but has a previous one, we assume the length of this one is same as the previous one
+	if prev != 0 {
+		return ts + (ts - prev)
 	}
+	// if a chunk has no next and no previous chunk we have to assume it's length is 0
+	return ts
 }
 
 // lastTs returns the last Ts of this metric cache