Merge #154481

154481: sidetransport: pace sidetransport sender messages r=iskettaneh a=iskettaneh

This PR paces sending sidetransport updates to all nodes. This way, we will have a lower number of runnable goroutines at any given time.

Before this change, you can see how at fixed intervals, multiple goroutines unblock at the exact same time:
<img width="951" height="489" alt="image" src="https://github.com/user-attachments/assets/6b225ed1-17a7-4a2d-9794-8b175818a200" />

After this change, that unblocking happens in a slightly staggered fashion:

<img width="957" height="584" alt="image" src="https://github.com/user-attachments/assets/7fe7b21f-dbd9-454c-ad9e-2e698db86754" />

This showed a significant reduction in the number of runnable goroutines.

More results can be found [in this document](https://docs.google.com/document/d/1mzl3vyPD8KASi2c14tJ8iRxbtGtGg7OEQvkdYk-Fhpk/edit?tab=t.0).

Fixes: #148211

Release note: None

Co-authored-by: iskettaneh <[email protected]>

Author: craig[bot]

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/policy.go

@@ -22,7 +22,10 @@ const (
 // global reads. It accounts for network latency, clock offset, and both Raft
 // and side-transport propagation delays.
 func computeLeadTimeForGlobalReads(
-	networkRTT time.Duration, maxClockOffset time.Duration, sideTransportCloseInterval time.Duration,
+	networkRTT time.Duration,
+	maxClockOffset time.Duration,
+	sideTransportCloseInterval time.Duration,
+	sideTransportPacingInterval time.Duration,
 ) time.Duration {
 	// The LEAD_FOR_GLOBAL_READS calculation is more complex. Instead of the
 	// policy defining an offset from the publisher's perspective, the
@@ -61,10 +64,13 @@ func computeLeadTimeForGlobalReads(
 	//  # the worst-case.
 	//  side_propagation_time = max_network_rtt * 0.5 + side_transport_close_interval
 	//
+	//  # We pace sending side-transport updates up to side_transport_pacing_refresh_interval.
+	//  # We have to also include it into our calculation.
+	//
 	//  # Combine, we get the following result
 	//  closed_ts_at_sender = now + max_offset + max(
 	//    max_network_rtt * 1.5 + raft_overhead,
-	//    max_network_rtt * 0.5 + side_transport_close_interval,
+	//    max_network_rtt * 0.5 + side_transport_close_interval + side_transport_pacing_refresh_interval,
 	//  )
 	//
 	// By default, this leads to a closed timestamp target that leads the
@@ -86,7 +92,7 @@ func computeLeadTimeForGlobalReads(
 	raftTransportPropTime := (networkRTT*3)/2 + raftTransportOverhead
 
 	// See side_propagation_time.
-	sideTransportPropTime := networkRTT/2 + sideTransportCloseInterval
+	sideTransportPropTime := networkRTT/2 + sideTransportCloseInterval + sideTransportPacingInterval
 
 	// See propagation_time.
 	maxTransportPropTime := max(sideTransportPropTime, raftTransportPropTime)
@@ -106,6 +112,7 @@ func TargetForPolicy(
 	lagTargetDuration time.Duration,
 	leadTargetOverride time.Duration,
 	sideTransportCloseInterval time.Duration,
+	sideTransportPacingInterval time.Duration,
 	policy ctpb.RangeClosedTimestampPolicy,
 ) hlc.Timestamp {
 	var targetOffsetTime time.Duration
@@ -121,7 +128,7 @@ func TargetForPolicy(
 			break
 		}
 		targetOffsetTime = computeLeadTimeForGlobalReads(computeNetworkRTTBasedOnPolicy(policy),
-			maxClockOffset, sideTransportCloseInterval)
+			maxClockOffset, sideTransportCloseInterval, sideTransportPacingInterval)
 	default:
 		panic("unexpected RangeClosedTimestampPolicy")
 	}

pkg/kv/kvserver/closedts/policy_test.go

@@ -28,11 +28,12 @@ func TestTargetForPolicy(t *testing.T) {
 	maxClockOffset := millis(500)
 
 	for _, tc := range []struct {
-		lagTargetNanos             time.Duration
-		leadTargetOverride         time.Duration
-		sideTransportCloseInterval time.Duration
-		rangePolicy                ctpb.RangeClosedTimestampPolicy
-		expClosedTSTarget          hlc.Timestamp
+		lagTargetNanos              time.Duration
+		leadTargetOverride          time.Duration
+		sideTransportCloseInterval  time.Duration
+		sideTransportPacingInterval time.Duration
+		rangePolicy                 ctpb.RangeClosedTimestampPolicy
+		expClosedTSTarget           hlc.Timestamp
 	}{
 		{
 			lagTargetNanos:    secs(3),
@@ -52,6 +53,16 @@ func TestTargetForPolicy(t *testing.T) {
 					millis(275) /* sideTransportPropTime */ +
 					millis(25) /* bufferTime */).Nanoseconds(), 0),
 		},
+		{
+			sideTransportCloseInterval:  millis(200),
+			sideTransportPacingInterval: millis(10),
+			rangePolicy:                 ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO,
+			expClosedTSTarget: now.
+				Add((maxClockOffset +
+					millis(275) /* sideTransportPropTime */ +
+					millis(10) /* sideTransportPacing */ +
+					millis(25) /* bufferTime */).Nanoseconds(), 0),
+		},
 		{
 			sideTransportCloseInterval: millis(50),
 			rangePolicy:                ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO,
@@ -61,10 +72,11 @@ func TestTargetForPolicy(t *testing.T) {
 					millis(25) /* bufferTime */).Nanoseconds(), 0),
 		},
 		{
-			leadTargetOverride:         millis(1234),
-			sideTransportCloseInterval: millis(200),
-			rangePolicy:                ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO,
-			expClosedTSTarget:          now.Add(millis(1234).Nanoseconds(), 0),
+			leadTargetOverride:          millis(1234),
+			sideTransportCloseInterval:  millis(200),
+			sideTransportPacingInterval: millis(10),
+			rangePolicy:                 ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO,
+			expClosedTSTarget:           now.Add(millis(1234).Nanoseconds(), 0),
 		},
 	} {
 		t.Run("", func(t *testing.T) {
@@ -74,6 +86,7 @@ func TestTargetForPolicy(t *testing.T) {
 				tc.lagTargetNanos,
 				tc.leadTargetOverride,
 				tc.sideTransportCloseInterval,
+				tc.sideTransportPacingInterval,
 				tc.rangePolicy,
 			)
 			require.Equal(t, tc.expClosedTSTarget, target)

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