sidetransport: pace sidetransport sender messages

Before this commit, every sidetransport sending interval, we used to
wake up up-to N goroutines to send the sidetransport messages where
N=number of nodes. This adds an overhead on the go scheduler as the
number of nodes in the cluster gets larger.

This commit addresses that by adding a small pacing duration while we
wake up the goroutines to send messages.

Fixes: #148211

Release note: None

Author: iskettaneh

docs/generated/settings/settings-for-tenants.txt

@@ -84,6 +84,8 @@ kv.closed_timestamp.follower_reads.enabled (alias: kv.closed_timestamp.follower_
 kv.closed_timestamp.lead_for_global_reads_auto_tune.enabled	boolean	false	if enabled, observed network latency between leaseholders and their furthest follower will be used to adjust closed timestamp policies for rangesranges configured to serve global reads. kv.closed_timestamp.lead_for_global_reads_override takes precedence if set.	system-visible
 kv.closed_timestamp.lead_for_global_reads_override	duration	0s	if nonzero, overrides the lead time that global_read ranges use to publish closed timestamps	system-visible
 kv.closed_timestamp.side_transport_interval	duration	200ms	the interval at which the closed timestamp side-transport attempts to advance each range's closed timestamp; set to 0 to disable the side-transport	system-visible
+kv.closed_timestamp.side_transport_pacing_refresh_interval	duration	10ms	the refresh interval for the task pacer that controls pacing of sending sidetransport updates to avoid overloading the system when many connections are waiting	system-visible
+kv.closed_timestamp.side_transport_pacing_smear_interval	duration	1ms	the smear interval for the task pacer that controls the amount of time each paced batch is going to take when broadcasting sidetransport updates	system-visible
 kv.closed_timestamp.target_duration	duration	3s	if nonzero, attempt to provide closed timestamp notifications for timestamps trailing cluster time by approximately this duration	system-visible
 kv.dist_sender.circuit_breaker.cancellation.enabled	boolean	true	when enabled, in-flight requests will be cancelled when the circuit breaker trips	application
 kv.dist_sender.circuit_breaker.cancellation.write_grace_period	duration	10s	how long after the circuit breaker trips to cancel write requests (these can't retry internally, so should be long enough to allow quorum/lease recovery)	application

docs/generated/settings/settings.html

@@ -100,6 +100,8 @@
 <tr><td><div id="setting-kv-closed-timestamp-lead-for-global-reads-auto-tune-enabled" class="anchored"><code>kv.closed_timestamp.lead_for_global_reads_auto_tune.enabled</code></div></td><td>boolean</td><td><code>false</code></td><td>if enabled, observed network latency between leaseholders and their furthest follower will be used to adjust closed timestamp policies for rangesranges configured to serve global reads. kv.closed_timestamp.lead_for_global_reads_override takes precedence if set.</td><td>Advanced/Self-hosted (read-write); Basic/Standard (read-only)</td></tr>
 <tr><td><div id="setting-kv-closed-timestamp-lead-for-global-reads-override" class="anchored"><code>kv.closed_timestamp.lead_for_global_reads_override</code></div></td><td>duration</td><td><code>0s</code></td><td>if nonzero, overrides the lead time that global_read ranges use to publish closed timestamps</td><td>Advanced/Self-hosted (read-write); Basic/Standard (read-only)</td></tr>
 <tr><td><div id="setting-kv-closed-timestamp-side-transport-interval" class="anchored"><code>kv.closed_timestamp.side_transport_interval</code></div></td><td>duration</td><td><code>200ms</code></td><td>the interval at which the closed timestamp side-transport attempts to advance each range&#39;s closed timestamp; set to 0 to disable the side-transport</td><td>Advanced/Self-hosted (read-write); Basic/Standard (read-only)</td></tr>
+<tr><td><div id="setting-kv-closed-timestamp-side-transport-pacing-refresh-interval" class="anchored"><code>kv.closed_timestamp.side_transport_pacing_refresh_interval</code></div></td><td>duration</td><td><code>10ms</code></td><td>the refresh interval for the task pacer that controls pacing of sending sidetransport updates to avoid overloading the system when many connections are waiting</td><td>Advanced/Self-hosted (read-write); Basic/Standard (read-only)</td></tr>
+<tr><td><div id="setting-kv-closed-timestamp-side-transport-pacing-smear-interval" class="anchored"><code>kv.closed_timestamp.side_transport_pacing_smear_interval</code></div></td><td>duration</td><td><code>1ms</code></td><td>the smear interval for the task pacer that controls the amount of time each paced batch is going to take when broadcasting sidetransport updates</td><td>Advanced/Self-hosted (read-write); Basic/Standard (read-only)</td></tr>
 <tr><td><div id="setting-kv-closed-timestamp-target-duration" class="anchored"><code>kv.closed_timestamp.target_duration</code></div></td><td>duration</td><td><code>3s</code></td><td>if nonzero, attempt to provide closed timestamp notifications for timestamps trailing cluster time by approximately this duration</td><td>Advanced/Self-hosted (read-write); Basic/Standard (read-only)</td></tr>
 <tr><td><div id="setting-kv-dist-sender-circuit-breaker-cancellation-enabled" class="anchored"><code>kv.dist_sender.circuit_breaker.cancellation.enabled</code></div></td><td>boolean</td><td><code>true</code></td><td>when enabled, in-flight requests will be cancelled when the circuit breaker trips</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>
 <tr><td><div id="setting-kv-dist-sender-circuit-breaker-cancellation-write-grace-period" class="anchored"><code>kv.dist_sender.circuit_breaker.cancellation.write_grace_period</code></div></td><td>duration</td><td><code>10s</code></td><td>how long after the circuit breaker trips to cancel write requests (these can&#39;t retry internally, so should be long enough to allow quorum/lease recovery)</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>

pkg/kv/kvserver/closedts/setting.go

@@ -98,3 +98,28 @@ var PolicySwitchWhenLatencyExceedsBucketFraction = settings.RegisterFloatSetting
 		"exceeded before the closed timestamp policy will be changed",
 	0.2,
 )
+
+// SideTransportPacingRefreshInterval controls the task pacer refresh interval
+// for pacing broadcast updates in the side transport. This determines how long
+// the pacer takes to complete signaling all waiting connections.
+var SideTransportPacingRefreshInterval = settings.RegisterDurationSetting(
+	settings.SystemVisible,
+	"kv.closed_timestamp.side_transport_pacing_refresh_interval",
+	"the refresh interval for the task pacer that controls pacing of sending "+
+		"sidetransport updates to avoid overloading the system when many connections are waiting",
+	10*time.Millisecond,
+	settings.WithPublic,
+)
+
+// SideTransportPacingSmearInterval controls the time interval between batches
+// when broadcasting side transport updates. The pacer uses this to spread out
+// work over time, sleeping between batches to avoid creating spikes in runnable
+// goroutines when many connections are waiting for updates.
+var SideTransportPacingSmearInterval = settings.RegisterDurationSetting(
+	settings.SystemVisible,
+	"kv.closed_timestamp.side_transport_pacing_smear_interval",
+	"the smear interval for the task pacer that controls the amount of time "+
+		"each paced batch is going to take when broadcasting sidetransport updates",
+	1*time.Millisecond,
+	settings.WithPublic,
+)

pkg/kv/kvserver/closedts/sidetransport/BUILD.bazel

@@ -28,6 +28,7 @@ go_library(
         "//pkg/util/netutil",
         "//pkg/util/stop",
         "//pkg/util/syncutil",
+        "//pkg/util/taskpacer",
         "//pkg/util/timeutil",
         "@com_github_cockroachdb_errors//:errors",
     ],
@@ -51,6 +52,7 @@ go_test(
         "//pkg/rpc/rpcbase",
         "//pkg/settings/cluster",
         "//pkg/testutils",
+        "//pkg/testutils/skip",
         "//pkg/util/hlc",
         "//pkg/util/leaktest",
         "//pkg/util/log",

pkg/kv/kvserver/closedts/sidetransport/sender.go

@@ -33,6 +33,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/netutil"
 	"github.com/cockroachdb/cockroach/pkg/util/stop"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
+	"github.com/cockroachdb/cockroach/pkg/util/taskpacer"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/errors"
 )
@@ -116,6 +117,21 @@ type connTestingKnobs struct {
 	sleepOnErrOverride time.Duration
 }
 
+// pacerConfig implements taskpacer.Config using cluster settings.
+type pacerConfig struct {
+	st *cluster.Settings
+}
+
+// GetRefresh implements the taskpacer.Config interface
+func (c *pacerConfig) GetRefresh() time.Duration {
+	return closedts.SideTransportPacingRefreshInterval.Get(&c.st.SV)
+}
+
+// GetSmear implements the taskpacer.Config interface
+func (c *pacerConfig) GetSmear() time.Duration {
+	return closedts.SideTransportPacingSmearInterval.Get(&c.st.SV)
+}
+
 // trackedRange contains the information that the side-transport last published
 // about a particular range.
 type trackedRange struct {
@@ -212,7 +228,7 @@ func newSenderWithConnFactory(
 		st:          st,
 		clock:       clock,
 		connFactory: connFactory,
-		buf:         newUpdatesBuf(),
+		buf:         newUpdatesBuf(st),
 	}
 	s.trackedMu.tracked = make(map[roachpb.RangeID]trackedRange)
 	s.trackedMu.lastClosed = make(map[ctpb.RangeClosedTimestampPolicy]hlc.Timestamp)
@@ -237,6 +253,12 @@ func (s *Sender) Run(ctx context.Context, nodeID roachpb.NodeID) {
 	}
 	closedts.SideTransportCloseInterval.SetOnChange(&s.st.SV, confChanged)
 
+	// Set up callback for pacing refresh interval changes.
+	pacingConfChanged := func(ctx context.Context) {
+		s.buf.updatePacer(s.st)
+	}
+	closedts.SideTransportPacingRefreshInterval.SetOnChange(&s.st.SV, pacingConfChanged)
+
 	_ /* err */ = s.stopper.RunAsyncTask(ctx, "closedts side-transport publisher",
 		func(ctx context.Context) {
 			defer func() {
@@ -540,6 +562,10 @@ type updatesBuf struct {
 	mu struct {
 		syncutil.Mutex
 
+		// pacer controls the timing of broadcast updates to avoid overloading the
+		// system.
+		pacer *taskpacer.Pacer
+
 		// We use two condition variables that we atomically swap to avoid signaling
 		// the same goroutine multiple times. Without this, a goroutine could:
 		// 1. Wake up from a signal.
@@ -580,14 +606,22 @@ type updatesBuf struct {
 // little while and not have to send a snapshot when it resumes.
 const updatesBufSize = 50
 
-func newUpdatesBuf() *updatesBuf {
+func newUpdatesBuf(st *cluster.Settings) *updatesBuf {
 	buf := &updatesBuf{}
 	buf.mu.updated1.L = &buf.mu
 	buf.mu.updated2.L = &buf.mu
 	buf.mu.data = make([]*ctpb.Update, updatesBufSize)
+	buf.mu.pacer = taskpacer.New(&pacerConfig{st: st})
 	return buf
 }
 
+// updatePacer atomically replaces the task pacer with a new one using the current cluster settings.
+func (b *updatesBuf) updatePacer(st *cluster.Settings) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.mu.pacer = taskpacer.New(&pacerConfig{st: st})
+}
+
 // Push adds a new update to the back of the buffer.
 func (b *updatesBuf) Push(ctx context.Context, update *ctpb.Update) {
 	b.mu.Lock()
@@ -632,20 +666,39 @@ func (b *updatesBuf) Push(ctx context.Context, update *ctpb.Update) {
 	// Notify everybody who might have been waiting for this message - we expect
 	// all the connections to be blocked waiting.
 	b.mu.Unlock()
-	b.PaceBroadcastUpdate(condVar, numWaiters)
+	b.PaceBroadcastUpdate(ctx, condVar, numWaiters)
 }
 
-// PaceBroadcastUpdate paces the conditional variable signaling to avoid
-// overloading the system.
-func (b *updatesBuf) PaceBroadcastUpdate(condVar *sync.Cond, numWaiters *int) {
-	// TODO(ibrahim): We can just hook this with the taskPacer.
+// PaceBroadcastUpdate paces the conditional variable signaling to avoid overloading the system.
+func (b *updatesBuf) PaceBroadcastUpdate(ctx context.Context, condVar *sync.Cond, numWaiters *int) {
 	b.mu.Lock()
 	originalNumWaiters := *numWaiters
-	for originalNumWaiters > 0 {
-		condVar.Signal()
-		originalNumWaiters--
+	if originalNumWaiters <= 0 {
+		b.mu.Unlock()
+		return
 	}
+
+	// Get the current pacer (which uses the cluster setting refresh interval).
+	pacer := b.mu.pacer
 	b.mu.Unlock()
+
+	pacer.StartTask(timeutil.Now())
+
+	workLeft := originalNumWaiters
+	for workLeft > 0 {
+		todo, by := pacer.Pace(timeutil.Now(), workLeft)
+
+		b.mu.Lock()
+		for i := 0; i < todo && workLeft > 0; i++ {
+			condVar.Signal()
+			workLeft--
+		}
+		b.mu.Unlock()
+
+		if workLeft > 0 && timeutil.Now().Before(by) {
+			time.Sleep(by.Sub(timeutil.Now()))
+		}
+	}
 }
 
 func (b *updatesBuf) lastIdxLocked() int {
@@ -743,6 +796,14 @@ func (b *updatesBuf) Close() {
 	b.mu.updated2.Broadcast()
 }
 
+// TestingGetTotalNumWaiters returns the total number of goroutines waiting
+// on the buffer. For testing purposes only.
+func (b *updatesBuf) TestingGetTotalNumWaiters() int {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.mu.numWaiters1 + b.mu.numWaiters2
+}
+
 // connFactory is capable of creating new connections to specific nodes.
 type connFactory interface {
 	new(*Sender, roachpb.NodeID) conn

pkg/kv/kvserver/closedts/sidetransport/sender_test.go

@@ -24,6 +24,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/rpc/rpcbase"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
@@ -927,3 +928,109 @@ func TestAllUpdatesBufAreSignalled(t *testing.T) {
 	wg.Wait()
 	require.Equal(t, int64(numGoroutines*numMsgs), counter.Load())
 }
+
+// TestPaceUpdateSignalling verifies that the task pacer properly spaces out
+// signal calls after an update on the updatesBuf.
+func TestPaceUpdateSignalling(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	// Flaky under duress.
+	skip.UnderDuress(t)
+
+	// Create a mock sender to get access to the buffer.
+	ctx := context.Background()
+	connFactory := &mockConnFactory{}
+	st := cluster.MakeTestingClusterSettings()
+	closedts.SideTransportPacingRefreshInterval.Override(ctx, &st.SV, 250*time.Millisecond)
+	s, stopper := newMockSenderWithSt(connFactory, st)
+	defer stopper.Stop(ctx)
+
+	// numWaiters controls how many goroutines will wait on updatesBuf.
+	numWaiters := 1000
+
+	// testPacing tests that when multiple goroutines are waiting on
+	// s.buf.GetBySeq, they receive their items spaced out by at least
+	// expectedMinSpread. If expectedMinSpread is 0, then pacing is disabled,
+	// and we expect all goroutines to receive their items within a few ms of
+	// each other.
+	//
+	// seqNum is the sequence number to wait for.
+	testPacing := func(seqNum ctpb.SeqNum, assertionFunc func(timeSpread time.Duration)) {
+		// Track the times when goroutines receive items from the buffer.
+		var receiveTimes []time.Time
+		var mu syncutil.Mutex
+
+		// Create numWaiters goroutines that wait on s.buf.GetBySeq and record
+		// receive times.
+		done := make(chan struct{}, numWaiters)
+		for i := 0; i < numWaiters; i++ {
+			go func() {
+				// Wait for the specified sequence number.
+				_, ok := s.buf.GetBySeq(ctx, seqNum)
+				require.Equal(t, true, ok)
+				if ok {
+					mu.Lock()
+					receiveTimes = append(receiveTimes, time.Now())
+					mu.Unlock()
+				}
+				done <- struct{}{}
+			}()
+		}
+
+		// Wait until all goroutines are waiting on the buffer.
+		testutils.SucceedsSoon(t, func() error {
+			if s.buf.TestingGetTotalNumWaiters() < numWaiters {
+				return errors.New("not all goroutines are waiting yet")
+			}
+			return nil
+		})
+
+		// Publish an item to the buffer, which should trigger the paced signalling.
+		s.publish(ctx)
+
+		// Wait for all goroutines to finish.
+		for i := 0; i < numWaiters; i++ {
+			<-done
+		}
+
+		// Verify that all goroutines received the message.
+		require.Len(t, receiveTimes, numWaiters)
+
+		// Find min and max receive times.
+		minTime := receiveTimes[0]
+		maxTime := receiveTimes[0]
+		for _, t := range receiveTimes[1:] {
+			if t.Before(minTime) {
+				minTime = t
+			}
+			if t.After(maxTime) {
+				maxTime = t
+			}
+		}
+
+		// Verify that the time spread matches expectations.
+		timeSpread := maxTime.Sub(minTime)
+		assertionFunc(timeSpread)
+	}
+
+	// Test with 250ms pacing interval - expect at least 125ms spread just to be
+	// conservative. In practice, it should be closer to 250ms.
+	closedts.SideTransportPacingRefreshInterval.Override(ctx, &st.SV, 250*time.Millisecond)
+	testPacing(1 /* seqNum */, func(timeSpread time.Duration) {
+		require.GreaterOrEqual(t, timeSpread, 125*time.Millisecond)
+	})
+
+	// Change to 100ms pacing interval - expect at least 50ms spread.
+	closedts.SideTransportPacingRefreshInterval.Override(ctx, &st.SV, 100*time.Millisecond)
+	testPacing(2 /* seqNum */, func(timeSpread time.Duration) {
+		require.GreaterOrEqual(t, timeSpread, 50*time.Millisecond)
+	})
+
+	// Change to 0ms (disabled) pacing interval - expect all goroutines to be
+	// woken within a few milliseconds of each other.
+	closedts.SideTransportPacingRefreshInterval.Override(ctx, &st.SV, 0)
+	testPacing(3 /* seqNum */, func(timeSpread time.Duration) {
+		require.LessOrEqual(t, timeSpread, 25*time.Millisecond)
+	})
+}