diff --git a/tools/loadgen/maxrps_readreceipt.go b/tools/loadgen/maxrps_readreceipt.go index 365de3c10..9f79b2731 100644 --- a/tools/loadgen/maxrps_readreceipt.go +++ b/tools/loadgen/maxrps_readreceipt.go @@ -25,6 +25,7 @@ func buildReadReceiptInputs(targetRPS int, hold time.Duration, c *ReadReceiptCol AttemptedOps: len(samples) + failed, FailedOps: failed, Saturation: c.Saturation(), + EmitUnderrun: c.UnderrunCount(), Latencies: []seriesSamples{ {Name: "read-receipt", Samples: samples}, }, diff --git a/tools/loadgen/maxrps_readreceipt_test.go b/tools/loadgen/maxrps_readreceipt_test.go index 94bae4cc7..cc9c336df 100644 --- a/tools/loadgen/maxrps_readreceipt_test.go +++ b/tools/loadgen/maxrps_readreceipt_test.go @@ -33,3 +33,11 @@ func TestBuildReadReceiptInputs(t *testing.T) { assert.Empty(t, in.Pending) assert.False(t, in.Inconclusive) } + +func TestBuildReadReceiptInputs_PopulatesEmitUnderrun(t *testing.T) { + c := NewReadReceiptCollector() + c.RecordUnderrun(5) + c.RecordUnderrun(4) + in := buildReadReceiptInputs(2000, 30*time.Second, c) + assert.Equal(t, 9, in.EmitUnderrun) +} diff --git a/tools/loadgen/maxrps_roomread.go b/tools/loadgen/maxrps_roomread.go index 4a8889f9a..c9533308d 100644 --- a/tools/loadgen/maxrps_roomread.go +++ b/tools/loadgen/maxrps_roomread.go @@ -33,6 +33,7 @@ func buildRoomReadInputs(targetRPS int, hold time.Duration, c *RoomReadCollector AttemptedOps: len(samples) + failed, FailedOps: failed, Saturation: c.SaturationCount(), + EmitUnderrun: c.UnderrunCount(), Latencies: []seriesSamples{ {Name: "room-read", Samples: roomReadLatencies(samples)}, }, diff --git a/tools/loadgen/maxrps_roomread_test.go b/tools/loadgen/maxrps_roomread_test.go index 09cf6bad9..b24ceb74f 100644 --- a/tools/loadgen/maxrps_roomread_test.go +++ b/tools/loadgen/maxrps_roomread_test.go @@ -33,6 +33,14 @@ func TestBuildRoomReadInputs_MapsCollector(t *testing.T) { assert.Len(t, in.Latencies[0].Samples, 2) } +func TestBuildRoomReadInputs_PopulatesEmitUnderrun(t *testing.T) { + c := NewRoomReadCollector() + c.RecordUnderrun(7) + c.RecordUnderrun(3) + in := buildRoomReadInputs(2000, 30*time.Second, c) + assert.Equal(t, 10, in.EmitUnderrun) +} + func TestRoomReadWorkload_Label(t *testing.T) { w := &roomReadWorkload{} assert.Equal(t, "room-read", w.Label()) diff --git a/tools/loadgen/members_generator.go b/tools/loadgen/members_generator.go index 81c6f318b..8c9527ba6 100644 --- a/tools/loadgen/members_generator.go +++ b/tools/loadgen/members_generator.go @@ -164,7 +164,11 @@ func NewSustainedMembersGenerator(cfg *SustainedMembersConfig, seed int64) *Sust } // Run drives the publish loop. Returns ErrPoolsExhausted if every room runs -// out of candidates before ctx is cancelled. +// out of candidates before ctx is cancelled. The arrival rate is paced by the +// batched pacer (events released per coarse tick), so achieved RPS is not +// capped by single-ticker resolution; events the pacer cannot release on +// schedule are tallied as the "underrun" publish-error reason (a load-box +// diagnostic, not a real publish failure). func (g *SustainedMembersGenerator) Run(ctx context.Context) error { if g.cfg.Rate <= 0 { return fmt.Errorf("rate must be > 0") @@ -172,11 +176,9 @@ func (g *SustainedMembersGenerator) Run(ctx context.Context) error { if g.cfg.UsersPerAdd <= 0 { return fmt.Errorf("usersPerAdd must be > 0") } - interval := time.Second / time.Duration(g.cfg.Rate) - if interval <= 0 { - interval = time.Nanosecond - } - tick := time.NewTicker(interval) + + p := newPacer(g.cfg.Rate, time.Now()) + tick := time.NewTicker(p.interval) defer tick.Stop() var sem chan struct{} @@ -184,43 +186,47 @@ func (g *SustainedMembersGenerator) Run(ctx context.Context) error { sem = make(chan struct{}, g.cfg.MaxInFlight) } var wg sync.WaitGroup + drain := func() { + done := make(chan struct{}) + go func() { wg.Wait(); close(done) }() + select { + case <-done: + case <-time.After(drainGracePeriod): + } + } for { select { case <-ctx.Done(): - done := make(chan struct{}) - go func() { wg.Wait(); close(done) }() - select { - case <-done: - case <-time.After(drainGracePeriod): - } + drain() return nil case <-tick.C: - roomID, accounts, ok := g.takeNext() - if !ok { - // Drain in-flight before returning so prior publishes complete. - done := make(chan struct{}) - go func() { wg.Wait(); close(done) }() - select { - case <-done: - case <-time.After(drainGracePeriod): - } - return ErrPoolsExhausted - } - if sem == nil { - g.publishOne(ctx, roomID, accounts) - continue + emit, underrun := p.tick(time.Now()) + if underrun > 0 { + g.cfg.Metrics.MemberPublishErrors.WithLabelValues("underrun").Add(float64(underrun)) } - select { - case sem <- struct{}{}: - wg.Add(1) - go func() { - defer func() { <-sem; wg.Done() }() + for i := 0; i < emit; i++ { + roomID, accounts, ok := g.takeNext() + if !ok { + // Drain in-flight before returning so prior publishes complete. + drain() + return ErrPoolsExhausted + } + if sem == nil { g.publishOne(ctx, roomID, accounts) - }() - default: - g.cfg.Metrics.MemberPublishErrors.WithLabelValues("saturated").Inc() - g.giveBack(roomID, accounts) + continue + } + select { + case sem <- struct{}{}: + wg.Add(1) + go func() { + defer func() { <-sem; wg.Done() }() + g.publishOne(ctx, roomID, accounts) + }() + default: + g.cfg.Metrics.MemberPublishErrors.WithLabelValues("saturated").Inc() + g.giveBack(roomID, accounts) + } } } } diff --git a/tools/loadgen/metrics.go b/tools/loadgen/metrics.go index 1132a7f17..f42835898 100644 --- a/tools/loadgen/metrics.go +++ b/tools/loadgen/metrics.go @@ -74,7 +74,7 @@ func NewMetrics() *Metrics { []string{"preset", "phase", "inject", "shape"}, ) m.MemberPublishErrors = prometheus.NewCounterVec( - prometheus.CounterOpts{Name: "loadgen_member_publish_errors_total", Help: "Member-add publish-side errors by reason (publish|room_service|timeout|marshal|saturated)."}, + prometheus.CounterOpts{Name: "loadgen_member_publish_errors_total", Help: "Member-add publish-side errors by reason (publish|room_service|timeout|marshal|saturated|underrun)."}, []string{"reason"}, ) m.MemberE1Latency = prometheus.NewHistogramVec( diff --git a/tools/loadgen/readreceipt_collector.go b/tools/loadgen/readreceipt_collector.go index 3fecde8d0..dc4670233 100644 --- a/tools/loadgen/readreceipt_collector.go +++ b/tools/loadgen/readreceipt_collector.go @@ -13,6 +13,7 @@ type ReadReceiptCollector struct { samples []time.Duration errors map[errClass]int saturation int + underrun int } // NewReadReceiptCollector returns an empty collector. @@ -48,6 +49,17 @@ func (c *ReadReceiptCollector) RecordSaturation() { c.saturation++ } +// RecordUnderrun adds n events that the pacer could not release on schedule +// (the load box fell behind the target cadence). n<=0 ticks are no-ops. +func (c *ReadReceiptCollector) RecordUnderrun(n int) { + if n <= 0 { + return + } + c.mu.Lock() + defer c.mu.Unlock() + c.underrun += n +} + // Samples returns a defensive copy of the latency tape. func (c *ReadReceiptCollector) Samples() []time.Duration { c.mu.Lock() @@ -74,3 +86,10 @@ func (c *ReadReceiptCollector) Saturation() int { defer c.mu.Unlock() return c.saturation } + +// UnderrunCount returns the total emit-underrun events. +func (c *ReadReceiptCollector) UnderrunCount() int { + c.mu.Lock() + defer c.mu.Unlock() + return c.underrun +} diff --git a/tools/loadgen/readreceipt_collector_test.go b/tools/loadgen/readreceipt_collector_test.go index c45194705..08001544a 100644 --- a/tools/loadgen/readreceipt_collector_test.go +++ b/tools/loadgen/readreceipt_collector_test.go @@ -7,6 +7,15 @@ import ( "github.com/stretchr/testify/assert" ) +func TestReadReceiptCollector_Underrun(t *testing.T) { + c := NewReadReceiptCollector() + assert.Equal(t, 0, c.UnderrunCount()) + c.RecordUnderrun(5) + c.RecordUnderrun(0) // zero is a no-op tick, must not change the tally + c.RecordUnderrun(4) + assert.Equal(t, 9, c.UnderrunCount()) +} + func TestReadReceiptCollector_SamplesAndErrors(t *testing.T) { c := NewReadReceiptCollector() c.RecordSample(15 * time.Millisecond) diff --git a/tools/loadgen/readreceipt_generator.go b/tools/loadgen/readreceipt_generator.go index 803199ef5..998b947db 100644 --- a/tools/loadgen/readreceipt_generator.go +++ b/tools/loadgen/readreceipt_generator.go @@ -67,57 +67,21 @@ func NewReadReceiptGenerator(cfg *ReadReceiptGeneratorConfig, seed int64) *ReadR } } -// Run drives the open-loop publisher until ctx cancels. +// Run drives the open-loop publisher until ctx cancels. MaxInFlight>0 uses the +// batched pacer (so achieved RPS is not capped by single-ticker resolution); +// MaxInFlight<=0 selects the legacy serial path, retained for bisection — it +// will not ramp past the single-ticker ceiling. func (g *ReadReceiptGenerator) Run(ctx context.Context) error { if g.cfg.Rate <= 0 { return fmt.Errorf("rate must be > 0") } - interval := time.Second / time.Duration(g.cfg.Rate) - if interval <= 0 { - interval = time.Nanosecond - } - tick := time.NewTicker(interval) - defer tick.Stop() - if g.cfg.MaxInFlight <= 0 { - for { - select { - case <-ctx.Done(): - return nil - case <-tick.C: - g.requestOne(ctx) - } - } - } - - sem := make(chan struct{}, g.cfg.MaxInFlight) - var wg sync.WaitGroup - for { - select { - case <-ctx.Done(): - done := make(chan struct{}) - go func() { wg.Wait(); close(done) }() - select { - case <-done: - case <-time.After(drainGracePeriod): - } - return nil - case <-tick.C: - select { - case sem <- struct{}{}: - wg.Add(1) - go func() { - defer func() { - <-sem - wg.Done() - }() - g.requestOne(ctx) - }() - default: - g.cfg.Collector.RecordSaturation() - } - } + serialDispatch(ctx, g.cfg.Rate, g.requestOne) + return nil } + pacedDispatch(ctx, g.cfg.Rate, g.cfg.MaxInFlight, + g.cfg.Collector.RecordUnderrun, g.cfg.Collector.RecordSaturation, g.requestOne) + return nil } func (g *ReadReceiptGenerator) requestOne(ctx context.Context) { diff --git a/tools/loadgen/roomread_collector.go b/tools/loadgen/roomread_collector.go index 0a4f4b31b..81031cd60 100644 --- a/tools/loadgen/roomread_collector.go +++ b/tools/loadgen/roomread_collector.go @@ -19,6 +19,7 @@ type RoomReadCollector struct { samples []RoomReadSample errors map[errClass]int saturation int + underrun int } // NewRoomReadCollector returns an empty collector. @@ -54,6 +55,17 @@ func (c *RoomReadCollector) RecordSaturation() { c.saturation++ } +// RecordUnderrun adds n events that the pacer could not release on schedule +// (the load box fell behind the target cadence). n<=0 ticks are no-ops. +func (c *RoomReadCollector) RecordUnderrun(n int) { + if n <= 0 { + return + } + c.mu.Lock() + defer c.mu.Unlock() + c.underrun += n +} + // Samples returns a defensive copy of the sample tape. func (c *RoomReadCollector) Samples() []RoomReadSample { c.mu.Lock() @@ -84,3 +96,10 @@ func (c *RoomReadCollector) SaturationCount() int { defer c.mu.Unlock() return c.saturation } + +// UnderrunCount returns the total emit-underrun events. +func (c *RoomReadCollector) UnderrunCount() int { + c.mu.Lock() + defer c.mu.Unlock() + return c.underrun +} diff --git a/tools/loadgen/roomread_collector_test.go b/tools/loadgen/roomread_collector_test.go index 3c011c446..ffffc88ba 100644 --- a/tools/loadgen/roomread_collector_test.go +++ b/tools/loadgen/roomread_collector_test.go @@ -25,6 +25,15 @@ func TestRoomReadCollector_Aggregates(t *testing.T) { assert.Equal(t, 2, c.SaturationCount()) } +func TestRoomReadCollector_Underrun(t *testing.T) { + c := NewRoomReadCollector() + assert.Equal(t, 0, c.UnderrunCount()) + c.RecordUnderrun(5) + c.RecordUnderrun(0) // zero is a no-op tick, must not change the tally + c.RecordUnderrun(4) + assert.Equal(t, 9, c.UnderrunCount()) +} + func TestRoomReadCollector_SamplesIsCopy(t *testing.T) { c := NewRoomReadCollector() c.RecordSample(RoomReadSample{Latency: time.Millisecond, At: time.Now()}) diff --git a/tools/loadgen/roomread_generator.go b/tools/loadgen/roomread_generator.go index 76c472a67..f82e694c0 100644 --- a/tools/loadgen/roomread_generator.go +++ b/tools/loadgen/roomread_generator.go @@ -74,57 +74,21 @@ func newRoomReadGenerator(cfg *roomReadGeneratorConfig, seed int64) *roomReadGen return &roomReadGenerator{cfg: *cfg, rng: r, zipf: z, roomSubs: lookup} } -// Run drives the open-loop publisher until ctx cancels. Mirrors HistoryGenerator.Run. +// Run drives the open-loop publisher until ctx cancels. Mirrors HistoryGenerator.Run: +// MaxInFlight>0 uses the batched pacer (so achieved RPS is not capped by +// single-ticker resolution); MaxInFlight<=0 selects the legacy serial path, +// retained for bisection — it will not ramp past the single-ticker ceiling. func (g *roomReadGenerator) Run(ctx context.Context) error { if g.cfg.Rate <= 0 { return fmt.Errorf("rate must be > 0") } - interval := time.Second / time.Duration(g.cfg.Rate) - if interval <= 0 { - interval = time.Nanosecond - } - tick := time.NewTicker(interval) - defer tick.Stop() - if g.cfg.MaxInFlight <= 0 { - for { - select { - case <-ctx.Done(): - return nil - case <-tick.C: - g.requestOne(ctx) - } - } - } - - sem := make(chan struct{}, g.cfg.MaxInFlight) - var wg sync.WaitGroup - for { - select { - case <-ctx.Done(): - done := make(chan struct{}) - go func() { wg.Wait(); close(done) }() - select { - case <-done: - case <-time.After(drainGracePeriod): - } - return nil - case <-tick.C: - select { - case sem <- struct{}{}: - wg.Add(1) - go func() { - defer func() { - <-sem - wg.Done() - }() - g.requestOne(ctx) - }() - default: - g.cfg.Collector.RecordSaturation() - } - } + serialDispatch(ctx, g.cfg.Rate, g.requestOne) + return nil } + pacedDispatch(ctx, g.cfg.Rate, g.cfg.MaxInFlight, + g.cfg.Collector.RecordUnderrun, g.cfg.Collector.RecordSaturation, g.requestOne) + return nil } func (g *roomReadGenerator) requestOne(ctx context.Context) {