track chunk total points more efficiently

also, this way we're not forced to read out the value
note: i think some of the AggMetrics benchmarks have bugs making them
misleading, that's something to look into later.
but at least all times have decreased.

BEFORE:

BenchmarkProcess-8   	 2000000	       717 ns/op
PASS
ok  	github.com/raintank/metrictank/input	2.687s
BenchmarkAggMetrics1000Metrics1Day-8                  	       1	4787832311 ns/op
BenchmarkAggMetrics1kSeries2Chunks1kQueueSize-8       	2000000000	         0.04 ns/op
BenchmarkAggMetrics10kSeries2Chunks10kQueueSize-8     	       1	1210286443 ns/op
BenchmarkAggMetrics100kSeries2Chunks100kQueueSize-8   	       1	13937332075 ns/op
PASS
ok  	github.com/raintank/metrictank/mdata	20.655s

AFTER:

BenchmarkProcess-8   	 3000000	       500 ns/op
PASS
ok  	github.com/raintank/metrictank/input	2.715s
BenchmarkAggMetrics1000Metrics1Day-8                  	       1	3977004221 ns/op
BenchmarkAggMetrics1kSeries2Chunks1kQueueSize-8       	1000000000	         0.06 ns/op
BenchmarkAggMetrics10kSeries2Chunks10kQueueSize-8     	2000000000	         0.39 ns/op
BenchmarkAggMetrics100kSeries2Chunks100kQueueSize-8   	       1	9796636066 ns/op
PASS
ok  	github.com/raintank/metrictank/mdata	42.775s

input/input_test.go

@@ -8,7 +8,6 @@ import (
 	"github.com/raintank/met/helper"
 	"github.com/raintank/metrictank/idx/memory"
 	"github.com/raintank/metrictank/mdata"
-	"github.com/raintank/metrictank/mdata/chunk"
 	"github.com/raintank/metrictank/usage"
 	"gopkg.in/raintank/schema.v1"
 )
@@ -116,10 +115,6 @@ func BenchmarkProcess(b *testing.B) {
 	}
 
 	b.ResetTimer()
-	go func() {
-		for range chunk.TotalPoints {
-		}
-	}()
 	for i := 0; i < b.N; i++ {
 		in.Process(datas[i])
 	}

mdata/aggmetric_test.go

@@ -3,7 +3,6 @@ package mdata
 import (
 	"fmt"
 	"github.com/raintank/met/helper"
-	"github.com/raintank/metrictank/mdata/chunk"
 	"testing"
 )
 
@@ -175,10 +174,6 @@ func BenchmarkAggMetrics1000Metrics1Day(b *testing.B) {
 	}
 
 	metrics := NewAggMetrics(dnstore, chunkSpan, numChunks, chunkMaxStale, metricMaxStale, ttl, 0, aggSettings)
-	go func() {
-		for range chunk.TotalPoints {
-		}
-	}()
 	defer metrics.Stop()
 
 	maxT := 3600 * 24 * uint32(b.N) // b.N in days
@@ -216,10 +211,6 @@ func BenchmarkAggMetrics1kSeries2Chunks1kQueueSize(b *testing.B) {
 	}
 
 	metrics := NewAggMetrics(dnstore, chunkSpan, numChunks, chunkMaxStale, metricMaxStale, ttl, 0, aggSettings)
-	go func() {
-		for range chunk.TotalPoints {
-		}
-	}()
 	defer metrics.Stop()
 
 	maxT := uint32(1200)
@@ -257,10 +248,6 @@ func BenchmarkAggMetrics10kSeries2Chunks10kQueueSize(b *testing.B) {
 	}
 
 	metrics := NewAggMetrics(dnstore, chunkSpan, numChunks, chunkMaxStale, metricMaxStale, ttl, 0, aggSettings)
-	go func() {
-		for range chunk.TotalPoints {
-		}
-	}()
 	defer metrics.Stop()
 
 	maxT := uint32(1200)
@@ -298,10 +285,6 @@ func BenchmarkAggMetrics100kSeries2Chunks100kQueueSize(b *testing.B) {
 	}
 
 	metrics := NewAggMetrics(dnstore, chunkSpan, numChunks, chunkMaxStale, metricMaxStale, ttl, 0, aggSettings)
-	go func() {
-		for range chunk.TotalPoints {
-		}
-	}()
 	defer metrics.Stop()
 
 	maxT := uint32(1200)

mdata/chunk/chunk.go

@@ -2,17 +2,13 @@ package chunk
 
 import (
 	"fmt"
+	"sync/atomic"
 	"time"
 
 	"github.com/dgryski/go-tsz"
 )
 
-var TotalPoints chan int
-
-func init() {
-	// measurements can lag a bit, that's ok
-	TotalPoints = make(chan int, 1000)
-}
+var totalPoints uint64
 
 // Chunk is a chunk of data. not concurrency safe.
 type Chunk struct {
@@ -42,9 +38,13 @@ func (c *Chunk) Push(t uint32, v float64) error {
 	c.NumPoints += 1
 	c.LastTs = t
 	c.LastWrite = uint32(time.Now().Unix())
-	TotalPoints <- 1
+	atomic.AddUint64(&totalPoints, 1)
 	return nil
 }
 func (c *Chunk) Clear() {
-	TotalPoints <- -1 * int(c.NumPoints)
+	atomic.AddUint64(&totalPoints, ^uint64(c.NumPoints-1))
+}
+
+func TotalPoints() uint64 {
+	return atomic.LoadUint64(&totalPoints)
 }

metrictank.go

@@ -464,15 +464,14 @@ func initMetrics(stats met.Backend) {
 
 	// run a collector for some global stats
 	go func() {
-		currentPoints := 0
 		var m runtime.MemStats
 		var promotionReadyAtTs uint32
 
 		ticker := time.Tick(time.Duration(1) * time.Second)
 		for {
 			select {
 			case now := <-ticker:
-				points.Value(int64(currentPoints))
+				points.Value(int64(chunk.TotalPoints()))
 				runtime.ReadMemStats(&m)
 				alloc.Value(int64(m.Alloc))
 				totalAlloc.Value(int64(m.TotalAlloc))
@@ -501,8 +500,6 @@ func initMetrics(stats met.Backend) {
 				} else {
 					clusterPromoWait.Value(int64(promotionReadyAtTs - unix))
 				}
-			case update := <-chunk.TotalPoints:
-				currentPoints += update
 			case promotionReadyAtTs = <-promotionReadyAtChan:
 			}
 		}