| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | package metrix |
| 4 | |
| 5 | import ( |
| 6 | "sync" |
| 7 | "sync/atomic" |
| 8 | "testing" |
| 9 | |
| 10 | "github.com/stretchr/testify/require" |
| 11 | ) |
| 12 | |
| 13 | func TestCollectorStoreConcurrencyScenarios(t *testing.T) { |
| 14 | tests := map[string]struct { |
| 15 | run func(t *testing.T) |
| 16 | }{ |
| 17 | "single writer with concurrent readers is race-safe": { |
| 18 | run: func(t *testing.T) { |
| 19 | s := NewCollectorStore() |
| 20 | cc := cycleController(t, s) |
| 21 | g := s.Write().SnapshotMeter("collector").Gauge("load") |
| 22 | |
| 23 | const cycles = 200 |
| 24 | const readers = 8 |
| 25 | |
| 26 | var writeDone atomic.Bool |
| 27 | var writerWG sync.WaitGroup |
| 28 | var readerWG sync.WaitGroup |
| 29 | |
| 30 | writerWG.Go(func() { |
| 31 | for i := 1; i <= cycles; i++ { |
| 32 | cc.BeginCycle() |
| 33 | g.Observe(SampleValue(i)) |
| 34 | cc.CommitCycleSuccess() |
| 35 | } |
| 36 | writeDone.Store(true) |
| 37 | }) |
| 38 | |
| 39 | readerWG.Add(readers) |
| 40 | for range readers { |
| 41 | go func() { |
| 42 | defer readerWG.Done() |
| 43 | for !writeDone.Load() { |
| 44 | r := s.Read() |
| 45 | _ = r.CollectMeta() |
| 46 | _, _ = r.Value("collector.load", nil) |
| 47 | r.ForEachSeries(func(_ string, _ LabelView, _ SampleValue) {}) |
| 48 | r.ForEachSeriesIdentity(func(_ SeriesIdentity, _ SeriesMeta, _ string, _ LabelView, _ SampleValue) {}) |
| 49 | |
| 50 | raw := s.Read(ReadRaw()) |
| 51 | _ = raw.CollectMeta() |
| 52 | _, _ = raw.Value("collector.load", nil) |
| 53 | } |
| 54 | }() |
| 55 | } |
| 56 | |
| 57 | writerWG.Wait() |
| 58 | readerWG.Wait() |
| 59 | |
| 60 | meta := s.Read().CollectMeta() |
| 61 | require.Equal(t, CollectStatusSuccess, meta.LastAttemptStatus, "unexpected collect metadata: %#v", meta) |
| 62 | require.Equal(t, uint64(cycles), meta.LastAttemptSeq, "unexpected collect metadata: %#v", meta) |
| 63 | require.Equal(t, uint64(cycles), meta.LastSuccessSeq, "unexpected collect metadata: %#v", meta) |
| 64 | mustValue(t, s.Read(), "collector.load", nil, SampleValue(cycles)) |
| 65 | }, |
| 66 | }, |
| 67 | } |
| 68 | |
| 69 | for name, tc := range tests { |
| 70 | t.Run(name, tc.run) |
| 71 | } |
| 72 | } |