diff --git a/pkg/types/encoding/carbonapi_v2/packed_stream_test.go b/pkg/types/encoding/carbonapi_v2/packed_stream_test.go index 19ca039e..84bff4ce 100644 --- a/pkg/types/encoding/carbonapi_v2/packed_stream_test.go +++ b/pkg/types/encoding/carbonapi_v2/packed_stream_test.go @@ -55,6 +55,45 @@ func TestPackedStreamDensityAndSplitFields(t *testing.T) { } } +func TestPackedStreamMixedMetricOwnership(t *testing.T) { + rng := rand.New(rand.NewSource(20261022)) + msg := &pb.MultiFetchResponse{} + // Alternate packed and dense storage, and exercise both buffer growth + // and shorter successors. Retained dense arrays must remain independent. + for i := 0; i < 60; i++ { + n := []int{6000, 6000, 256, 128, 127, 0}[i%6] + m := &pb.FetchResponse{Name: string(rune('a' + i)), Values: make([]float64, n), IsAbsent: make([]bool, n)} + for j := range m.Values { + m.IsAbsent[j] = (i+j)%3 == 0 + if i%6 == 1 || j%997 == 0 { + m.Values[j] = math.Float64frombits(rng.Uint64()) + } + } + msg.Metrics = append(msg.Metrics, m) + } + body, err := msg.MarshalVT() + if err != nil { + t.Fatal(err) + } + want, err := RenderDecoder(body) + if err != nil { + t.Fatal(err) + } + for _, chunk := range []int{1, 31, 32768} { + got, err := RenderStreamDecoderPacked(&fragmentedReader{data: body, chunk: chunk}) + if err != nil { + t.Fatal(err) + } + // Empty repeated fields retain the reference decoder's nil slices. + for i := range want { + if want[i].IsAbsent == nil && got[i].IsAbsent != nil { + t.Fatalf("metric %d: empty absence field became non-nil", i) + } + } + equalStreamMetrics(t, unpackStream(got), want) + } +} + func BenchmarkSparseStream(b *testing.B) { msg := &pb.MultiFetchResponse{} for i := 0; i < 100; i++ { diff --git a/pkg/types/encoding/carbonapi_v2/stream.go b/pkg/types/encoding/carbonapi_v2/stream.go index 419ff1aa..c0913f1b 100644 --- a/pkg/types/encoding/carbonapi_v2/stream.go +++ b/pkg/types/encoding/carbonapi_v2/stream.go @@ -36,6 +36,7 @@ func renderStreamDecoder(src io.Reader, packed bool) ([]types.Metric, error) { defer func() { r.Reset(nil); streamReaders.Put(r) }() s := protoStream{r: r, left: math.MaxInt64} metrics := make([]types.Metric, 0) + var absentScratch []bool for { tag, err := s.varint() if err == io.EOF { @@ -60,7 +61,7 @@ func renderStreamDecoder(src io.Reader, packed bool) ([]types.Metric, error) { if err != nil { return nil, err } - child := protoStream{r: r, left: n} + child := protoStream{r: r, left: n, absentScratch: absentScratch} m, err := child.metric(packed) if err != nil { return nil, err @@ -68,14 +69,21 @@ func renderStreamDecoder(src io.Reader, packed bool) ([]types.Metric, error) { s.left -= n if packed { m.PackPoints() + // Packed metrics own a bitmap, so their temporary bool storage can + // serve the next metric. Dense metrics retain it and must detach. + absentScratch = child.absentScratch + if m.IsAbsent != nil || cap(absentScratch) > maxStreamPointChunk { + absentScratch = nil + } } metrics = append(metrics, m) } } type protoStream struct { - r *bufio.Reader - left int64 + r *bufio.Reader + left int64 + absentScratch []bool } func validTag(tag uint64) error { @@ -281,7 +289,12 @@ func (s *protoStream) bools(dst []bool, n int64) ([]bool, error) { return dst, nil } start := len(dst) - dst = append(dst, make([]bool, int(n))...) + if int(n) <= cap(dst)-start { + // read overwrites every byte before any bool is inspected. + dst = dst[:start+int(n)] + } else { + dst = append(dst, make([]bool, int(n))...) + } if n == 0 { return dst, nil } @@ -307,6 +320,9 @@ func (s *protoStream) bools(dst []bool, n int64) ([]bool, error) { } func (s *protoStream) metric(packed bool) (types.Metric, error) { var m types.Metric + if packed { + m.IsAbsent = s.absentScratch[:0] + } var points types.PointPacker packing := packed for s.left > 0 { @@ -406,6 +422,10 @@ func (s *protoStream) metric(packed bool) (types.Metric, error) { } } } + if len(m.IsAbsent) == 0 { + m.IsAbsent = nil + } + s.absentScratch = m.IsAbsent if packing { points.Finish(&m) } diff --git a/pkg/types/packed_merge_bench_test.go b/pkg/types/packed_merge_bench_test.go new file mode 100644 index 00000000..327949eb --- /dev/null +++ b/pkg/types/packed_merge_bench_test.go @@ -0,0 +1,37 @@ +package types + +import ( + "testing" + + "github.com/bookingcom/carbonapi/pkg/cfg" +) + +func BenchmarkPackedReplicaMerge(b *testing.B) { + for _, density := range []string{"zero", "sparse", "dense"} { + for _, mode := range []cfg.ReplicaMatchMode{cfg.ReplicaMatchModeNormal, cfg.ReplicaMatchModeCheck, cfg.ReplicaMatchModeMajority} { + b.Run(density+"/"+string(mode), func(b *testing.B) { + metrics := make([]Metric, 3) + for replica := range metrics { + m := Metric{StepTime: 60, Values: make([]float64, 10080), IsAbsent: make([]bool, 10080)} + for i := range m.Values { + m.IsAbsent[i] = (i+replica)%3 == 0 + if density == "dense" || density == "sparse" && i%997 == 0 { + m.Values[i] = 17 + } + } + m.PackPoints() + metrics[replica] = m + } + c := cfg.RenderReplicaMismatchConfig{RenderReplicaMatchMode: mode} + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + m, stats := mergeMetrics(metrics, c) + if len(m.Values) != 10080 || stats.MismatchCount != 0 { + b.Fatal("invalid merge") + } + } + }) + } + } +} diff --git a/pkg/types/packed_points_test.go b/pkg/types/packed_points_test.go index 351e3f5f..b4fd5cc2 100644 --- a/pkg/types/packed_points_test.go +++ b/pkg/types/packed_points_test.go @@ -11,6 +11,49 @@ import ( "go.uber.org/zap" ) +func TestPackedZeroBlockMergeOracle(t *testing.T) { + rng := rand.New(rand.NewSource(20261022)) + for seed := 0; seed < 80; seed++ { + input := make([][]Metric, 2+seed%3) + for replica := range input { + n := []int{128, 129, 255, 256, 10080}[seed%5] + m := Metric{Name: "zero-blocks", StepTime: 60, Values: make([]float64, n), IsAbsent: make([]bool, n)} + if seed%7 == 0 && replica == 1 { + m.StepTime = 120 + } + for i := range m.Values { + m.IsAbsent[i] = rng.Intn(3) == 0 + if seed%4 != 0 && i%997 == 0 { + m.Values[i] = float64(replica + 1) + } + } + if seed%4 == 2 { + m.Values[63] = math.Float64frombits(1 << 63) + m.Values[64] = math.Float64frombits(0x7ff8000000000123) + m.Values[127] = math.Inf(replica%2*2 - 1) + } + input[replica] = []Metric{m} + } + for _, mode := range []cfg.ReplicaMatchMode{cfg.ReplicaMatchModeNormal, cfg.ReplicaMatchModeCheck, cfg.ReplicaMatchModeMajority} { + for _, approx := range []bool{false, true} { + c := cfg.RenderReplicaMismatchConfig{RenderReplicaMatchMode: mode, RenderReplicaMismatchApproximateCheck: approx} + want, ws := MergeMetrics(copyPackedOracleInput(input, false), c, zap.NewNop()) + for _, mixed := range []bool{false, true} { + pending := copyPackedOracleInput(input, true) + if mixed { + pending[1][0].materializePoints() + } + got, gs := MergeMetrics(pending, c, zap.NewNop()) + if gs != ws { + t.Fatalf("seed %d: stats differ: %+v / %+v", seed, gs, ws) + } + assertPackedBits(t, got, want) + } + } + } + } +} + func copyPackedOracleInput(input [][]Metric, packed bool) [][]Metric { out := make([][]Metric, len(input)) for i, ms := range input { diff --git a/pkg/types/types.go b/pkg/types/types.go index bd6d6b2f..0b47a07d 100644 --- a/pkg/types/types.go +++ b/pkg/types/types.go @@ -11,6 +11,7 @@ package types import ( "errors" "math" + "math/bits" "sort" "github.com/bookingcom/carbonapi/pkg/cfg" @@ -319,7 +320,7 @@ func mergeMetrics(metrics []Metric, replicaMismatchConfig cfg.RenderReplicaMisma metric = metrics[0] metric.materializePoints() var cursors []packedPointCursor - for _, replica := range metrics[1:] { + for _, replica := range metrics { if replica.packed != nil { cursors = make([]packedPointCursor, len(metrics)) for i := range cursors { @@ -337,13 +338,26 @@ func mergeMetrics(metrics []Metric, replicaMismatchConfig cfg.RenderReplicaMisma } valuesForPoint := make([]float64, 0, len(metrics)) isMismatchFindConfig := replicaMatchMode != cfg.ReplicaMatchModeNormal - for i := range metric.Values { + for i := 0; i < len(metric.Values); i++ { if cursors != nil && i&63 == 0 { - for j := 1; j < replicaLimit; j++ { + for j := 0; j < replicaLimit; j++ { if p := metrics[j].packed; p != nil { cursors[j].advanceToBlock(p, i>>6) } } + if metrics[0].packed != nil { + if absent, ok := packedZeroBlockAbsent(metrics[:replicaLimit], cursors, i>>6); ok { + // Every value in this block is +0. Only absence healing can + // change the result; no mismatch comparison is necessary. + healedBits := metrics[0].packed.absent[i>>6] &^ absent + for healedBits != 0 { + metric.IsAbsent[i+bits.TrailingZeros64(healedBits)] = false + healedBits &= healedBits - 1 + } + i += min(64, len(metric.Values)-i) - 1 + continue + } + } } pointExists := !metric.IsAbsent[i] shouldLookForMismatch := isMismatchFindConfig @@ -411,6 +425,20 @@ func mergeMetrics(metrics []Metric, replicaMismatchConfig cfg.RenderReplicaMisma } } +// packedZeroBlockAbsent combines absence flags only when every compatible +// replica stores exclusively positive-zero values in this block. Nonzero bits, +// including negative zero and NaNs, stay on the point-wise merge path. +func packedZeroBlockAbsent(metrics []Metric, cursors []packedPointCursor, block int) (uint64, bool) { + absent := ^uint64(0) + for i, m := range metrics { + if m.packed == nil || m.packed.values[cursors[i].offset] != 0 { + return 0, false + } + absent &= m.packed.absent[block] + } + return absent, true +} + // Info contains metadata about a metric in Graphite. type Info struct { Host string