loadgen: route room-read/read-receipt/members through the batched pacer

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},
 		},

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)
+}

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)},
 		},

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())

tools/loadgen/members_generator.go

@@ -164,63 +164,69 @@ 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")
 	}
 	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{}
 	if g.cfg.MaxInFlight > 0 {
 		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)
+				}
 			}
 		}
 	}

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(

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
+}

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)

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) {

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
+}

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()})

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) {