admission: move yielding into ElasticCPUWorkHandle

ElasticCPUWorkHandle.Overlimit is expected to be called in a tight loop,
so yielding there is conceptually the right place. More importantly,
this will allow in the future for KV work that is not holding latches
to also yield.

As part of this change, elastic work that does not wish to wait in
admission control queues (due to cluster settings), is now accounted for
in the elastic tokens, and in the admission.elastic_cpu_bypassed.utilization
metric. One side-effect of this accounting is that work that needs to
wait in admission queues may have fewer tokens available to it, and may
wait longer. This is considered acceptable since:
- Elastic work that bypasses queueing is still elastic work, and our
  overarching goal is to reduce impact to foreground work.
- Due to the default on use of runtime.Yield, all elastic work yields,
  which allows the system to run at higher elastic CPU utilization without
  impacting the latency of foreground work.

Epic: none

Release note: None

Author: sumeerbhola

pkg/backup/datadriven_test.go

@@ -27,7 +27,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/kv"
-	"github.com/cockroachdb/cockroach/pkg/kv/bulk"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvclient/kvcoord"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver"
@@ -200,8 +199,7 @@ func (d *datadrivenTestState) addCluster(t *testing.T, cfg clusterCfg) error {
 	// and overload the host. Generally overload would mean bulk work, which only
 	// uses strictly spare capacitym gets starved, but these tests expect it to
 	// still run (just slowly, along with everything else).
-	bulk.YieldIfNoPacer.Override(context.Background(), &settings.SV, false)
-	admission.YieldInPacer.Override(context.Background(), &settings.SV, false)
+	admission.YieldForElasticCPU.Override(context.Background(), &settings.SV, false)
 	params.ServerArgs.Settings = settings
 
 	clusterSize := cfg.nodes

pkg/kv/bulk/cpu_pacer.go

@@ -25,26 +25,18 @@ var cpuPacerRequestDuration = settings.RegisterDurationSetting(
 	50*time.Millisecond,
 )
 
-// YieldIfNoPacer is exported so it can be overridden in tests.
-var YieldIfNoPacer = settings.RegisterBoolSetting(
-	settings.ApplicationLevel,
-	"bulkio.elastic_cpu_control.always_yield.enabled",
-	"if true, yield the CPU as needed even when time-based elastic pacing is not enabled",
-	true,
-)
-
-// NewCPUPacer creates a new AC pacer for SST batcher. It may return an empty
-// Pacer which noops if pacing is disabled or its arguments are nil.
+// NewCPUPacer creates a new AC pacer for SST batcher. It will return an empty
+// Pacer which noops if db or db.AdmissionPacerFactory is nil.
+//
+// The setting specifies where waiting in the elastic admission control queue
+// is enabled. If disabled, the CPU consumed will be accounted for in
+// admission control, but pacing will not wait in admission control.
 func NewCPUPacer(ctx context.Context, db *kv.DB, setting *settings.BoolSetting) *admission.Pacer {
-	if db == nil || db.AdmissionPacerFactory == nil || !setting.Get(db.SettingsValues()) {
-		log.Dev.Infof(ctx, "admission control is not configured to pace bulk ingestion")
-
-		if db != nil && YieldIfNoPacer.Get(db.SettingsValues()) {
-			// Return a Pacer that just yields.
-			return &admission.Pacer{Yield: true}
-		}
+	if db == nil || db.AdmissionPacerFactory == nil {
+		log.Dev.Infof(ctx, "admission control is not configured to pace this bulk work")
 		return nil
 	}
+	bypassACQueue := !setting.Get(db.SettingsValues())
 	tenantID, ok := roachpb.ClientTenantFromContext(ctx)
 	if !ok {
 		tenantID = roachpb.SystemTenantID
@@ -55,6 +47,6 @@ func NewCPUPacer(ctx context.Context, db *kv.DB, setting *settings.BoolSetting)
 			TenantID:        tenantID,
 			Priority:        admissionpb.BulkNormalPri,
 			CreateTime:      timeutil.Now().UnixNano(),
-			BypassAdmission: false,
+			BypassAdmission: bypassACQueue,
 		})
 }

pkg/kv/kvserver/kvadmission/kvadmission.go

@@ -396,7 +396,7 @@ func (n *controllerImpl) AdmitKVWork(
 			// reads, so in some sense, the integration is incomplete. This is
 			// probably harmless.
 			elasticWorkHandle, err := n.elasticCPUGrantCoordinator.ElasticCPUWorkQueue.Admit(
-				ctx, admitDuration, admissionInfo,
+				ctx, admitDuration, admissionInfo, false,
 			)
 			if err != nil {
 				return Handle{}, err

pkg/server/testserver.go

@@ -26,7 +26,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/jobs"
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/kv"
-	"github.com/cockroachdb/cockroach/pkg/kv/bulk"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvclient/rangefeed"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvprober"
@@ -1486,8 +1485,7 @@ func (ts *testServer) StartSharedProcessTenant(
 
 	// Disable yield AC for tenant servers in tests, for the same reason as the
 	// system tenant (see comment in serverutils.NewServer).
-	admission.YieldInPacer.Override(ctx, &sqlServer.cfg.Settings.SV, false)
-	bulk.YieldIfNoPacer.Override(ctx, &sqlServer.cfg.Settings.SV, false)
+	admission.YieldForElasticCPU.Override(ctx, &sqlServer.cfg.Settings.SV, false)
 
 	hts := &httpTestServer{}
 	hts.t.authentication = sqlServerWrapper.authentication
@@ -1879,8 +1877,7 @@ func (ts *testServer) StartTenant(
 
 	// Disable yield AC for tenant servers in tests, for the same reason as the
 	// system tenant (see comment in serverutils.NewServer).
-	admission.YieldInPacer.Override(ctx, &st.SV, false)
-	bulk.YieldIfNoPacer.Override(ctx, &st.SV, false)
+	admission.YieldForElasticCPU.Override(ctx, &st.SV, false)
 
 	hts := &httpTestServer{}
 	hts.t.authentication = sw.authentication

pkg/testutils/serverutils/BUILD.bazel

@@ -24,7 +24,6 @@ go_library(
         "//pkg/config/zonepb",
         "//pkg/keys",
         "//pkg/kv",
-        "//pkg/kv/bulk",
         "//pkg/kv/kvpb",
         "//pkg/kv/kvprober",
         "//pkg/kv/kvserver/liveness/livenesspb",

pkg/testutils/serverutils/test_server_shim.go

@@ -22,7 +22,6 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/base"
 	"github.com/cockroachdb/cockroach/pkg/kv"
-	"github.com/cockroachdb/cockroach/pkg/kv/bulk"
 	"github.com/cockroachdb/cockroach/pkg/multitenant/tenantcapabilitiespb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/security/securitytest"
@@ -427,8 +426,7 @@ func NewServer(params base.TestServerArgs) (TestServerInterface, error) {
 	// system components/rangefeeds/etc of several testservers/tenants is enough
 	// to have a perpetually non-empty runnable queue in CIs for 10+ minutes. As
 	// such, we have to disable yield AC if we want background work to run at all.
-	admission.YieldInPacer.Override(context.Background(), &srv.(TestServerInterfaceRaw).ClusterSettings().SV, false)
-	bulk.YieldIfNoPacer.Override(context.Background(), &srv.(TestServerInterfaceRaw).ClusterSettings().SV, false)
+	admission.YieldForElasticCPU.Override(context.Background(), &srv.(TestServerInterfaceRaw).ClusterSettings().SV, false)
 	srv = wrapTestServer(srv.(TestServerInterfaceRaw), tcfg)
 	return srv.(TestServerInterface), nil
 }

pkg/util/admission/BUILD.bazel

@@ -24,6 +24,7 @@ go_library(
         "testing_knobs.go",
         "tokens_linear_model.go",
         "work_queue.go",
+        "yield.go",
     ],
     importpath = "github.com/cockroachdb/cockroach/pkg/util/admission",
     visibility = ["//visibility:public"],

pkg/util/admission/elastic_cpu_grant_coordinator.go

@@ -84,10 +84,10 @@ func (e *ElasticCPUGrantCoordinator) tryGrant() {
 	e.elasticCPUGranter.tryGrant()
 }
 
-// YieldInPacer is exported so it can be overridden in tests.
-var YieldInPacer = settings.RegisterBoolSetting(
+// YieldForElasticCPU is exported so it can be overridden in tests.
+var YieldForElasticCPU = settings.RegisterBoolSetting(
 	settings.ApplicationLevel,
-	"admission.elastic_cpu.yield_in_pacer.enabled",
+	"admission.elastic_cpu.yield.enabled",
 	"when true, time-based elastic CPU pacing additionally yields CPU as-needed according to the scheduler",
 	true,
 )
@@ -101,6 +101,6 @@ func (e *ElasticCPUGrantCoordinator) NewPacer(unit time.Duration, wi WorkInfo) *
 		unit:  unit,
 		wi:    wi,
 		wq:    e.ElasticCPUWorkQueue,
-		Yield: YieldInPacer.Get(&e.ElasticCPUWorkQueue.settings.SV),
+		yield: YieldForElasticCPU.Get(&e.ElasticCPUWorkQueue.settings.SV),
 	}
 }

pkg/util/admission/elastic_cpu_granter.go

@@ -8,6 +8,7 @@ package admission
 import (
 	"context"
 	"runtime"
+	"sync/atomic"
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
@@ -250,14 +251,26 @@ func (e *elasticCPUGranter) computeUtilizationMetric() {
 	currentCumAcquiredNanos := e.metrics.AcquiredNanos.Count()
 	currentCumUsedNanos := currentCumAcquiredNanos - currentCumReturnedNanos
 
+	intervalUsedFracFunc := func(used int64) float64 {
+		return float64(used) /
+			(float64(e.metrics.MaxAvailableNanos.Count()) * elasticCPUUtilizationMetricInterval.Seconds())
+	}
 	if e.metrics.lastCumUsedNanos != 0 {
 		intervalUsedNanos := currentCumUsedNanos - e.metrics.lastCumUsedNanos
-		intervalUsedPercent := float64(intervalUsedNanos) /
-			(float64(e.metrics.MaxAvailableNanos.Count()) * elasticCPUUtilizationMetricInterval.Seconds())
-		e.metrics.Utilization.Update(intervalUsedPercent)
-		e.metrics.lastCumUsedNanos = currentCumUsedNanos
+		intervalUsedFrac := intervalUsedFracFunc(intervalUsedNanos)
+		e.metrics.Utilization.Update(intervalUsedFrac)
 	}
 	e.metrics.lastCumUsedNanos = currentCumUsedNanos
+
+	// Compute the utilization of work that bypassed admission.
+	curBypassedNanos := e.metrics.bypassedAdmissionCumNanos.Load()
+	if curBypassedNanos != 0 {
+		intervalBypassedNanos := curBypassedNanos - e.metrics.lastBypassedAdmissionCumNanos
+		intervalBypassedFrac := intervalUsedFracFunc(intervalBypassedNanos)
+		e.metrics.BypassedAdmissionUtilization.Update(intervalBypassedFrac)
+
+	}
+	e.metrics.lastBypassedAdmissionCumNanos = curBypassedNanos
 }
 
 // TODO(irfansharif): Provide separate enums for different elastic CPU token
@@ -324,6 +337,13 @@ var ( // granter-side metrics (some of these have parallels on the requester sid
 		Unit:        metric.Unit_PERCENT,
 	}
 
+	elasticCPUGranterBypassedUtilization = metric.Metadata{
+		Name:        "admission.elastic_cpu_bypassed.utilization",
+		Help:        "CPU utilization by elastic work that bypassed admission",
+		Measurement: "CPU Time",
+		Unit:        metric.Unit_PERCENT,
+	}
+
 	elasticCPUGranterUtilizationLimit = metric.Metadata{
 		Name:        "admission.elastic_cpu.utilization_limit",
 		Help:        "Utilization limit set for the elastic CPU work",
@@ -344,9 +364,15 @@ type elasticCPUGranterMetrics struct {
 	NanosExhaustedDuration *metric.Counter
 	OverLimitDuration      metric.IHistogram
 
-	Utilization      *metric.GaugeFloat64 // updated every elasticCPUUtilizationMetricInterval, using fields below
+	Utilization                  *metric.GaugeFloat64 // updated every elasticCPUUtilizationMetricInterval, using fields below
+	BypassedAdmissionUtilization *metric.GaugeFloat64
+
 	everyInterval    util.EveryN[crtime.Mono]
 	lastCumUsedNanos int64
+
+	// Used for computing BypassedAdmissionUtilization.
+	bypassedAdmissionCumNanos     atomic.Int64
+	lastBypassedAdmissionCumNanos int64
 }
 
 const elasticCPUUtilizationMetricInterval = 10 * time.Second
@@ -365,9 +391,10 @@ func makeElasticCPUGranterMetrics() *elasticCPUGranterMetrics {
 			Duration:     base.DefaultHistogramWindowInterval(),
 			BucketConfig: metric.IOLatencyBuckets,
 		}),
-		Utilization:      metric.NewGaugeFloat64(elasticCPUGranterUtilization),
-		UtilizationLimit: metric.NewGaugeFloat64(elasticCPUGranterUtilizationLimit),
-		everyInterval:    util.EveryMono(elasticCPUUtilizationMetricInterval),
+		Utilization:                  metric.NewGaugeFloat64(elasticCPUGranterUtilization),
+		BypassedAdmissionUtilization: metric.NewGaugeFloat64(elasticCPUGranterBypassedUtilization),
+		UtilizationLimit:             metric.NewGaugeFloat64(elasticCPUGranterUtilizationLimit),
+		everyInterval:                util.EveryMono(elasticCPUUtilizationMetricInterval),
 	}
 
 	metrics.MaxAvailableNanos.Inc(int64(runtime.GOMAXPROCS(0)) * time.Second.Nanoseconds())

pkg/util/admission/elastic_cpu_work_handle.go

@@ -30,6 +30,10 @@ type ElasticCPUWorkHandle struct {
 	// duration to count against what it was allotted, but we still want to
 	// track to deduct an appropriate number of granter tokens.
 	preWork time.Duration
+	// yield decides whether Overlimit will call runtime.Yield.
+	yield bool
+	// bypassedAdmission is true if the allotment happened without waiting.
+	bypassedAdmission bool
 
 	// This handle is used in tight loops that are sensitive to per-iteration
 	// overhead (checking against the running time too can have an effect). To
@@ -49,9 +53,10 @@ type ElasticCPUWorkHandle struct {
 }
 
 func newElasticCPUWorkHandle(
-	tenantID roachpb.TenantID, allotted time.Duration,
+	tenantID roachpb.TenantID, allotted time.Duration, yield bool, bypassedAdmission bool,
 ) *ElasticCPUWorkHandle {
-	h := &ElasticCPUWorkHandle{tenantID: tenantID, allotted: allotted}
+	h := &ElasticCPUWorkHandle{
+		tenantID: tenantID, allotted: allotted, yield: yield, bypassedAdmission: bypassedAdmission}
 	h.cpuStart = grunning.Time()
 	return h
 }
@@ -85,6 +90,13 @@ func (h *ElasticCPUWorkHandle) runningTime() time.Duration {
 	return grunning.Elapsed(h.cpuStart, grunning.Time())
 }
 
+// SetYield sets the behavior of whether to call runtime.Yield. It should only
+// be used when it is not possible to set the value correctly at construction
+// time.
+func (h *ElasticCPUWorkHandle) SetYield(yield bool) {
+	h.yield = yield
+}
+
 // OverLimit is used to check whether we're over the allotted elastic CPU
 // tokens. If StartTimer was invoked, we start measuring on-CPU time only after
 // the invocation. It also returns the total time difference between how long we
@@ -103,7 +115,9 @@ func (h *ElasticCPUWorkHandle) OverLimit() (overLimit bool, difference time.Dura
 	if h == nil { // not applicable
 		return false, time.Duration(0)
 	}
-
+	if h.yield {
+		runtimeYield()
+	}
 	// What we're effectively doing is just:
 	//
 	// 		runningTime := h.runningTime()
@@ -190,7 +204,7 @@ func ElasticCPUWorkHandleFromContext(ctx context.Context) *ElasticCPUWorkHandle
 // TestingNewElasticCPUHandle exports the ElasticCPUWorkHandle constructor for
 // testing purposes.
 func TestingNewElasticCPUHandle() *ElasticCPUWorkHandle {
-	return newElasticCPUWorkHandle(roachpb.SystemTenantID, 420*time.Hour) // use a very high allotment
+	return newElasticCPUWorkHandle(roachpb.SystemTenantID, 420*time.Hour, false, false) // use a very high allotment
 }
 
 // TestingNewElasticCPUHandleWithCallback constructs an ElasticCPUWorkHandle