kv,rangefeed: integrate catchup scans with elastic cpu

Part of #65957.

Changefeed backfills, given their scan-heavy nature, can be fairly
CPU-intensive. In #89656 we introduced a roachtest demonstrating the
latency impact backfills can have on a moderately CPU-saturated cluster.
Similar to what we saw for backups, this CPU heavy nature can elevate Go
scheduling latencies which in turn translates to foreground latency
impact. This commit integrates rangefeed catchup scan with the elastic
CPU limiter we introduced in #86638; this is one of two optional halves
of changefeed backfills. The second half is the initial scan -- scan
requests issued over some keyspan as of some timestamp. For that we
simply rely on the existing slots mechanism but now setting a lower
priority bit (BulkNormalPri) -- #88733. Experimentally we observed that
during initial scans the encoding routines in changefeedccl are the most
impactful CPU-wise, something #89589 can help with. We leave admission
integration of parallel worker goroutines to future work (#90089).

Unlike export requests rangefeed catchup scans are non-premptible. The
rangefeed RPC is a streaming one, and the catchup scan is done during
stream setup. So we don't have resumption tokens to propagate up to the
caller like we did for backups. We still want CPU-bound work going
through admission control to only use 100ms of CPU time, to exercise
better control over scheduling latencies. To that end, we introduce the
following component used within the rangefeed catchup iterator.

    // Pacer is used in tight loops (CPU-bound) for non-premptible
    // elastic work. Callers are expected to invoke Pace() every loop
    // iteration and Close() once done. Internally this type integrates
    // with elastic CPU work queue, acquiring tokens for the CPU work
    // being done, and blocking if tokens are unavailable. This allows
    // for a form of cooperative scheduling with elastic CPU granters.
    type Pacer struct
    func (p *Pacer) Pace(ctx context.Context) error { ... }
    func (p *Pacer) Close() { ... }

Release note: None

pkg/ccl/changefeedccl/changefeedbase/settings.go

@@ -108,7 +108,7 @@ var FrontierCheckpointMaxBytes = settings.RegisterByteSizeSetting(
 	settings.TenantWritable,
 	"changefeed.frontier_checkpoint_max_bytes",
 	"controls the maximum size of the checkpoint as a total size of key bytes",
-	1<<20,
+	1<<20, // 1 MiB
 )
 
 // ScanRequestLimit is the number of Scan requests that can run at once.
@@ -126,7 +126,7 @@ var ScanRequestSize = settings.RegisterIntSetting(
 	settings.TenantWritable,
 	"changefeed.backfill.scan_request_size",
 	"the maximum number of bytes returned by each scan request",
-	16<<20,
+	16<<20, // 16 MiB
 )
 
 // SinkThrottleConfig describes throttling configuration for the sink.

pkg/clusterversion/cockroach_versions.go

@@ -301,7 +301,6 @@ const (
 	// SupportAssumeRoleAuth is the version where assume role authorization is
 	// supported in cloud storage and KMS.
 	SupportAssumeRoleAuth
-
 	// FixUserfileRelatedDescriptorCorruption adds a migration which uses
 	// heuristics to identify invalid table descriptors for userfile-related
 	// descriptors.

pkg/kv/kvserver/client_replica_circuit_breaker_test.go

@@ -442,7 +442,7 @@ func TestReplicaCircuitBreaker_RangeFeed(t *testing.T) {
 	defer cancel()
 	stream1 := &dummyStream{t: t, ctx: ctx, name: "rangefeed1", recv: make(chan *roachpb.RangeFeedEvent)}
 	require.NoError(t, tc.Stopper().RunAsyncTask(ctx, "stream1", func(ctx context.Context) {
-		err := tc.repls[0].RangeFeed(args, stream1).GoError()
+		err := tc.repls[0].RangeFeed(ctx, args, stream1).GoError()
 		if ctx.Err() != nil {
 			return // main goroutine stopping
 		}
@@ -496,7 +496,7 @@ func TestReplicaCircuitBreaker_RangeFeed(t *testing.T) {
 	// the breaker.
 	stream2 := &dummyStream{t: t, ctx: ctx, name: "rangefeed2", recv: make(chan *roachpb.RangeFeedEvent)}
 	require.NoError(t, tc.Stopper().RunAsyncTask(ctx, "stream2", func(ctx context.Context) {
-		err := tc.repls[0].RangeFeed(args, stream2).GoError()
+		err := tc.repls[0].RangeFeed(ctx, args, stream2).GoError()
 		if ctx.Err() != nil {
 			return // main goroutine stopping
 		}

pkg/kv/kvserver/kvadmission/kvadmission.go

@@ -51,30 +51,52 @@ var elasticCPUDurationPerExportRequest = settings.RegisterDurationSetting(
 	},
 )
 
+// elasticCPUDurationPerRangefeedScanUnit controls how many CPU tokens are
+// allotted for each unit of work during rangefeed catchup scans.
+var elasticCPUDurationPerRangefeedScanUnit = settings.RegisterDurationSetting(
+	settings.SystemOnly,
+	"kvadmission.elastic_cpu.duration_per_rangefeed_scan_unit",
+	"controls how many CPU tokens are allotted for each unit of work during rangefeed catchup scans",
+	admission.MaxElasticCPUDuration,
+	func(duration time.Duration) error {
+		if duration < admission.MinElasticCPUDuration {
+			return fmt.Errorf("minimum CPU duration allowed is %s, got %s",
+				admission.MinElasticCPUDuration, duration)
+		}
+		if duration > admission.MaxElasticCPUDuration {
+			return fmt.Errorf("maximum CPU duration allowed is %s, got %s",
+				admission.MaxElasticCPUDuration, duration)
+		}
+		return nil
+	},
+)
+
 // Controller provides admission control for the KV layer.
 type Controller interface {
 	// AdmitKVWork must be called before performing KV work.
 	// BatchRequest.AdmissionHeader and BatchRequest.Replica.StoreID must be
 	// populated for admission to work correctly. If err is non-nil, the
 	// returned handle can be ignored. If err is nil, AdmittedKVWorkDone must be
 	// called after the KV work is done executing.
-	AdmitKVWork(
-		ctx context.Context, tenantID roachpb.TenantID, ba *roachpb.BatchRequest,
-	) (Handle, error)
+	AdmitKVWork(context.Context, roachpb.TenantID, *roachpb.BatchRequest) (Handle, error)
 	// AdmittedKVWorkDone is called after the admitted KV work is done
 	// executing.
 	AdmittedKVWorkDone(Handle, *StoreWriteBytes)
+	// AdmitRangefeedRequest must be called before serving rangefeed requests.
+	// It returns a Pacer that's used within rangefeed catchup scans (typically
+	// CPU-intensive and affects scheduling latencies negatively).
+	AdmitRangefeedRequest(roachpb.TenantID, *roachpb.RangeFeedRequest) *Pacer
 	// SetTenantWeightProvider is used to set the provider that will be
 	// periodically polled for weights. The stopper should be used to terminate
 	// the periodic polling.
-	SetTenantWeightProvider(provider TenantWeightProvider, stopper *stop.Stopper)
+	SetTenantWeightProvider(TenantWeightProvider, *stop.Stopper)
 	// SnapshotIngested informs admission control about a range snapshot
 	// ingestion.
-	SnapshotIngested(storeID roachpb.StoreID, ingestStats pebble.IngestOperationStats)
+	SnapshotIngested(roachpb.StoreID, pebble.IngestOperationStats)
 	// FollowerStoreWriteBytes informs admission control about writes
 	// replicated to a raft follower, that have not been subject to admission
 	// control.
-	FollowerStoreWriteBytes(storeID roachpb.StoreID, followerWriteBytes FollowerStoreWriteBytes)
+	FollowerStoreWriteBytes(roachpb.StoreID, FollowerStoreWriteBytes)
 }
 
 // TenantWeightProvider can be periodically asked to provide the tenant
@@ -266,6 +288,27 @@ func (n *controllerImpl) AdmittedKVWorkDone(ah Handle, writeBytes *StoreWriteByt
 	}
 }
 
+// AdmitRangefeedRequest implements the Controller interface.
+func (n *controllerImpl) AdmitRangefeedRequest(
+	tenantID roachpb.TenantID, request *roachpb.RangeFeedRequest,
+) *Pacer {
+	// TODO(irfansharif): We need to version gate/be defensive when integrating
+	// rangefeeds since admission headers will not be fully set on older version
+	// nodes. See EnableRangefeedElasticCPUControl in cockroach_versions.go.
+	// Consider a cluster setting too.
+
+	return &Pacer{
+		unit: elasticCPUDurationPerRangefeedScanUnit.Get(&n.settings.SV),
+		wi: admission.WorkInfo{
+			TenantID:        tenantID,
+			Priority:        admissionpb.WorkPriority(request.AdmissionHeader.Priority),
+			CreateTime:      request.AdmissionHeader.CreateTime,
+			BypassAdmission: false,
+		},
+		wq: n.elasticCPUWorkQueue,
+	}
+}
+
 // SetTenantWeightProvider implements the Controller interface.
 func (n *controllerImpl) SetTenantWeightProvider(
 	provider TenantWeightProvider, stopper *stop.Stopper,
@@ -382,3 +425,67 @@ func (wb *StoreWriteBytes) Release() {
 	}
 	storeWriteBytesPool.Put(wb)
 }
+
+// Pacer is used in tight loops (CPU-bound) for non-premptible elastic work.
+// Callers are expected to invoke Pace() every loop iteration and Close() once
+// done. Internally this type integrates with elastic CPU work queue, acquiring
+// tokens for the CPU work being done, and blocking if tokens are unavailable.
+// This allows for a form of cooperative scheduling with elastic CPU granters.
+type Pacer struct {
+	unit time.Duration
+	wi   admission.WorkInfo
+	wq   *admission.ElasticCPUWorkQueue
+
+	cur *admission.ElasticCPUWorkHandle
+}
+
+// Pace is part of the Pacer interface.
+func (p *Pacer) Pace(ctx context.Context) error {
+	if p == nil {
+		return nil
+	}
+
+	if overLimit, _ := p.cur.OverLimit(); overLimit {
+		p.wq.AdmittedWorkDone(p.cur)
+		p.cur = nil
+	}
+
+	if p.cur == nil {
+		handle, err := p.wq.Admit(ctx, p.unit, p.wi)
+		if err != nil {
+			return err
+		}
+		p.cur = handle
+	}
+	return nil
+}
+
+// Close is part of the Pacer interface.
+func (p *Pacer) Close() {
+	if p == nil || p.cur == nil {
+		return
+	}
+
+	p.wq.AdmittedWorkDone(p.cur)
+	p.cur = nil
+}
+
+type pacerKey struct{}
+
+// ContextWithPacer returns a Context wrapping the supplied Pacer, if any.
+func ContextWithPacer(ctx context.Context, h *Pacer) context.Context {
+	if h == nil {
+		return ctx
+	}
+	return context.WithValue(ctx, pacerKey{}, h)
+}
+
+// PacerFromContext returns the Pacer contained in the Context, if any.
+func PacerFromContext(ctx context.Context) *Pacer {
+	val := ctx.Value(pacerKey{})
+	h, ok := val.(*Pacer)
+	if !ok {
+		return nil
+	}
+	return h
+}

pkg/kv/kvserver/rangefeed/BUILD.bazel

@@ -17,6 +17,7 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//pkg/keys",
+        "//pkg/kv/kvserver/kvadmission",
         "//pkg/roachpb",
         "//pkg/settings",
         "//pkg/storage",

pkg/kv/kvserver/rangefeed/catchup_scan.go

@@ -12,7 +12,9 @@ package rangefeed
 
 import (
 	"bytes"
+	"context"
 
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvadmission"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/storage/enginepb"
@@ -62,6 +64,7 @@ type CatchUpIterator struct {
 	close     func()
 	span      roachpb.Span
 	startTime hlc.Timestamp // exclusive
+	pacer     *kvadmission.Pacer
 }
 
 // NewCatchUpIterator returns a CatchUpIterator for the given Reader over the
@@ -70,7 +73,11 @@ type CatchUpIterator struct {
 // NB: startTime is exclusive, i.e. the first possible event will be emitted at
 // Timestamp.Next().
 func NewCatchUpIterator(
-	reader storage.Reader, span roachpb.Span, startTime hlc.Timestamp, closer func(),
+	reader storage.Reader,
+	span roachpb.Span,
+	startTime hlc.Timestamp,
+	closer func(),
+	pacer *kvadmission.Pacer,
 ) *CatchUpIterator {
 	return &CatchUpIterator{
 		simpleCatchupIter: storage.NewMVCCIncrementalIterator(reader,
@@ -89,13 +96,15 @@ func NewCatchUpIterator(
 		close:     closer,
 		span:      span,
 		startTime: startTime,
+		pacer:     pacer,
 	}
 }
 
 // Close closes the iterator and calls the instantiator-supplied close
 // callback.
 func (i *CatchUpIterator) Close() {
 	i.simpleCatchupIter.Close()
+	i.pacer.Close()
 	if i.close != nil {
 		i.close()
 	}
@@ -117,7 +126,9 @@ type outputEventFn func(e *roachpb.RangeFeedEvent) error
 // For example, with MVCC range tombstones [a-f)@5 and [a-f)@3 overlapping point
 // keys a@6, a@4, and b@2, the emitted order is [a-f)@3,[a-f)@5,a@4,a@6,b@2 because
 // the start key "a" is ordered before all of the timestamped point keys.
-func (i *CatchUpIterator) CatchUpScan(outputFn outputEventFn, withDiff bool) error {
+func (i *CatchUpIterator) CatchUpScan(
+	ctx context.Context, outputFn outputEventFn, withDiff bool,
+) error {
 	var a bufalloc.ByteAllocator
 	// MVCCIterator will encounter historical values for each key in
 	// reverse-chronological order. To output in chronological order, store
@@ -319,6 +330,10 @@ func (i *CatchUpIterator) CatchUpScan(outputFn outputEventFn, withDiff bool) err
 				i.Next()
 			}
 		}
+
+		if err := i.pacer.Pace(ctx); err != nil {
+			return errors.Wrap(err, "automatic pacing: %v")
+		}
 	}
 
 	// Output events for the last key encountered.

pkg/kv/kvserver/rangefeed/catchup_scan_bench_test.go

@@ -45,13 +45,14 @@ func runCatchUpBenchmark(b *testing.B, emk engineMaker, opts benchOptions) (numE
 		EndKey: endKey,
 	}
 
+	ctx := context.Background()
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
 		func() {
-			iter := rangefeed.NewCatchUpIterator(eng, span, opts.ts, nil)
+			iter := rangefeed.NewCatchUpIterator(eng, span, opts.ts, nil, nil)
 			defer iter.Close()
 			counter := 0
-			err := iter.CatchUpScan(func(*roachpb.RangeFeedEvent) error {
+			err := iter.CatchUpScan(ctx, func(*roachpb.RangeFeedEvent) error {
 				counter++
 				return nil
 			}, opts.withDiff)

pkg/kv/kvserver/rangefeed/catchup_scan_test.go

@@ -111,11 +111,11 @@ func TestCatchupScan(t *testing.T) {
 	}
 	testutils.RunTrueAndFalse(t, "withDiff", func(t *testing.T, withDiff bool) {
 		span := roachpb.Span{Key: testKey1, EndKey: roachpb.KeyMax}
-		iter := NewCatchUpIterator(eng, span, ts1, nil)
+		iter := NewCatchUpIterator(eng, span, ts1, nil, nil)
 		defer iter.Close()
 		var events []roachpb.RangeFeedValue
 		// ts1 here is exclusive, so we do not want the versions at ts1.
-		require.NoError(t, iter.CatchUpScan(func(e *roachpb.RangeFeedEvent) error {
+		require.NoError(t, iter.CatchUpScan(ctx, func(e *roachpb.RangeFeedEvent) error {
 			events = append(events, *e.Val)
 			return nil
 		}, withDiff))
@@ -154,10 +154,10 @@ func TestCatchupScanInlineError(t *testing.T) {
 
 	// Run a catchup scan across the span and watch it error.
 	span := roachpb.Span{Key: keys.LocalMax, EndKey: keys.MaxKey}
-	iter := NewCatchUpIterator(eng, span, hlc.Timestamp{}, nil)
+	iter := NewCatchUpIterator(eng, span, hlc.Timestamp{}, nil, nil)
 	defer iter.Close()
 
-	err := iter.CatchUpScan(nil, false)
+	err := iter.CatchUpScan(ctx, nil, false)
 	require.Error(t, err)
 	require.Contains(t, err.Error(), "unexpected inline value")
 }
@@ -194,11 +194,11 @@ func TestCatchupScanSeesOldIntent(t *testing.T) {
 
 	// Run a catchup scan across the span and watch it succeed.
 	span := roachpb.Span{Key: keys.LocalMax, EndKey: keys.MaxKey}
-	iter := NewCatchUpIterator(eng, span, tsCutoff, nil)
+	iter := NewCatchUpIterator(eng, span, tsCutoff, nil, nil)
 	defer iter.Close()
 
 	keys := map[string]struct{}{}
-	require.NoError(t, iter.CatchUpScan(func(e *roachpb.RangeFeedEvent) error {
+	require.NoError(t, iter.CatchUpScan(ctx, func(e *roachpb.RangeFeedEvent) error {
 		keys[string(e.Val.Key)] = struct{}{}
 		return nil
 	}, true /* withDiff */))

pkg/kv/kvserver/rangefeed/registry.go

@@ -300,7 +300,7 @@ func (r *registration) disconnect(pErr *roachpb.Error) {
 // have been emitted.
 func (r *registration) outputLoop(ctx context.Context) error {
 	// If the registration has a catch-up scan, run it.
-	if err := r.maybeRunCatchUpScan(); err != nil {
+	if err := r.maybeRunCatchUpScan(ctx); err != nil {
 		err = errors.Wrap(err, "catch-up scan failed")
 		log.Errorf(ctx, "%v", err)
 		return err
@@ -372,7 +372,7 @@ func (r *registration) drainAllocations(ctx context.Context) {
 //
 // If the registration does not have a catchUpIteratorConstructor, this method
 // is a no-op.
-func (r *registration) maybeRunCatchUpScan() error {
+func (r *registration) maybeRunCatchUpScan(ctx context.Context) error {
 	catchUpIter := r.detachCatchUpIter()
 	if catchUpIter == nil {
 		return nil
@@ -383,7 +383,7 @@ func (r *registration) maybeRunCatchUpScan() error {
 		r.metrics.RangeFeedCatchUpScanNanos.Inc(timeutil.Since(start).Nanoseconds())
 	}()
 
-	return catchUpIter.CatchUpScan(r.stream.Send, r.withDiff)
+	return catchUpIter.CatchUpScan(ctx, r.stream.Send, r.withDiff)
 }
 
 // ID implements interval.Interface.

pkg/kv/kvserver/rangefeed/registry_test.go

@@ -278,7 +278,7 @@ func TestRegistrationCatchUpScan(t *testing.T) {
 	}, hlc.Timestamp{WallTime: 4}, iter, true /* withDiff */)
 
 	require.Zero(t, r.metrics.RangeFeedCatchUpScanNanos.Count())
-	require.NoError(t, r.maybeRunCatchUpScan())
+	require.NoError(t, r.maybeRunCatchUpScan(context.Background()))
 	require.True(t, iter.closed)
 	require.NotZero(t, r.metrics.RangeFeedCatchUpScanNanos.Count())