mix: update comments and clean up commented out codes, todos

compress.go

@@ -361,12 +361,18 @@ func (whisper *Whisper) fetchCompressed(start, end int64, archive *archiveInfo)
 		}
 	}
 
+	// Start live aggregation. This probably has a read peformance hit.
 	if base := whisper.archives[0]; base != archive {
 		var dps []dataPoint
+
+		// Mix aggregation is triggered when block in base archive is rotated and also
+		// depends on the sufficiency of data points. This could results to a over
+		// long gap when fetching data from higer archives, depending on different
+		// retention policy. Therefore cwhisper needs to do live aggregation.
 		if whisper.aggregationMethod == Mix {
-			var overflow = len(dst) > 0 && dst[len(dst)-1].interval < int(end)
+			var baseLookupNeeded = len(dst) > 0 && dst[len(dst)-1].interval < int(end)
 			var inBase bool
-			if overflow {
+			if baseLookupNeeded {
 				bstart, bend := base.getOverallRange()
 				inBase = int64(bstart) <= end || end <= int64(bend)
 			}
@@ -385,6 +391,7 @@ func (whisper *Whisper) fetchCompressed(start, end int64, archive *archiveInfo)
 			}
 		}
 
+		// This would benefits both mix and no-mix aggregations.
 		if base.hasBuffer() {
 			for _, p := range unpackDataPoints(base.buffer) {
 				if p.interval != 0 && int(start) <= p.interval && p.interval <= int(end) {
@@ -393,7 +400,6 @@ func (whisper *Whisper) fetchCompressed(start, end int64, archive *archiveInfo)
 			}
 		}
 
-		// live aggregate
 		var pinterval int
 		var vals []float64
 		for i, dp := range dps {
@@ -514,6 +520,7 @@ func (whisper *Whisper) archiveUpdateManyCompressed(archive *archiveInfo, points
 			aggregateValue := aggregate(whisper.aggregationMethod, knownValues)
 			point := &TimeSeriesPoint{lowerIntervalStart, aggregateValue}
 
+			// TODO: consider migrating to a non-recursive propagation implementation like mix policy
 			if err := whisper.archiveUpdateManyCompressed(lower, []*TimeSeriesPoint{point}); err != nil {
 				return err
 			}

compress_test.go

@@ -912,7 +912,6 @@ func TestEstimatePointSize(t *testing.T) {
 	return
 }
 
-// TODO: srcMix is empty, investigate why
 func TestFillCompressedMix(t *testing.T) {
 	srcPath := "fill-mix.src.cwsp"
 	dstPath := "fill-mix.dst.cwsp"
@@ -954,11 +953,9 @@ func TestFillCompressedMix(t *testing.T) {
 	nowNext := func() time.Time { now++; return Now() }
 	defer func() { Now = func() time.Time { return time.Now() } }()
 
-	// twoMonthsAgo := Now().Add(time.Hour * 24 * -60)
 	limit = 300 + rand.Intn(100)
 	for i, end := 0, 60*60*24*80; i < end; i++ {
 		points = append(points, &TimeSeriesPoint{
-			// Time:  int(twoMonthsAgo.Add(time.Second * time.Duration(i)).Unix()),
 			Time:  int(nowNext().Unix()),
 			Value: rand.NormFloat64(),
 		})
@@ -1078,8 +1075,6 @@ func TestFillCompressedMix(t *testing.T) {
 	compare(dstMix, oldDstMix, int(Now().Add(time.Hour*24*-2+time.Hour).Unix()), int(Now().Unix()))
 }
 
-// TODO: check if there are duplicated timestamps by directly reading
-// data from blocks
 func TestFetchCompressedMix(t *testing.T) {
 	srcPath := "fetch-mix.cwsp"
 	os.Remove(srcPath)
@@ -1119,7 +1114,6 @@ func TestFetchCompressedMix(t *testing.T) {
 
 	for i := 0; i < 4*60*60; i++ {
 		points = append(points, &TimeSeriesPoint{
-			// Time:  int(start.Add(time.Second * time.Duration(i)).Unix()),
 			Time:  int(Now().Unix()),
 			Value: float64(i),
 		})
@@ -1143,7 +1137,6 @@ func TestFetchCompressedMix(t *testing.T) {
 		t.Error(err)
 	}
 
-	// data, err := srcMix.Fetch(int(start.Unix()), int(Now().Unix()))
 	t.Run("Check1stArchive", func(t *testing.T) {
 		data, err := srcMix.FetchByAggregation(now-10, now, &MixAggregationSpec{Method: Min})
 		if err != nil {

debug.go

@@ -121,11 +121,6 @@ func (archive *archiveInfo) dumpInfoCompressed() {
 	fmt.Printf("points_per_block:     %d\n", archive.calculateSuitablePointsPerBlock(archive.whisper.pointsPerBlock))
 	fmt.Printf("compression_ratio:    %f (%d/%d)\n", float64(archive.blockSize*archive.blockCount)/float64(archive.Size()), archive.blockSize*archive.blockCount, archive.Size())
 	if archive.aggregationSpec != nil {
-		// if archive.aggregationSpec.Method == Percentile {
-		// 	fmt.Printf("aggregation:          p%.2f\n", archive.aggregationSpec.Percentile)
-		// } else {
-		// 	fmt.Printf("aggregation:          %s\n", archive.aggregationSpec.Method)
-		// }
 		fmt.Printf("aggregation:          %s\n", archive.aggregationSpec)
 	}
 

whisper.go

@@ -173,7 +173,7 @@ type archiveInfo struct {
 	//  is necessary to have it, so it's possible to optimize away buffers in lower
 	//  archives.
 	buffer     []byte
-	bufferSize int
+	bufferSize int // dynamically calculated in Whisper.initMetaInfo
 
 	blockRanges []blockRange // TODO: remove: sorted by start
 	blockSize   int
@@ -643,10 +643,6 @@ func (whisper *Whisper) initMetaInfo() {
 		}
 
 		if i == 0 {
-			if whisper.aggregationMethod == Mix {
-				// TODO
-				// arc.bufferSize =
-			}
 			continue
 		}