Merge #154270

154270: takepacer: move task pacer to its own util package r=iskettaneh a=iskettaneh

Before this commit, the taskPacer lived inside kvserver, which made it unusable from packages that cannot include the kvserver package. This opens the door to using the taskPacer in more packages.

Epic: None

Release note: None

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

Author: craig[bot]

pkg/BUILD.bazel

@@ -817,6 +817,7 @@ ALL_TESTS = [
     "//pkg/util/syncutil/singleflight:singleflight_test",
     "//pkg/util/syncutil:syncutil_test",
     "//pkg/util/sysutil:sysutil_test",
+    "//pkg/util/taskpacer:taskpacer_test",
     "//pkg/util/timeofday:timeofday_test",
     "//pkg/util/timetz:timetz_test",
     "//pkg/util/timeutil/pgdate:pgdate_test",
@@ -2794,6 +2795,8 @@ GO_TARGETS = [
     "//pkg/util/system:system",
     "//pkg/util/sysutil:sysutil",
     "//pkg/util/sysutil:sysutil_test",
+    "//pkg/util/taskpacer:taskpacer",
+    "//pkg/util/taskpacer:taskpacer_test",
     "//pkg/util/timeofday:timeofday",
     "//pkg/util/timeofday:timeofday_test",
     "//pkg/util/timetz:timetz",

pkg/kv/kvserver/BUILD.bazel

@@ -107,7 +107,6 @@ go_library(
         "stores_base.go",
         "stores_server.go",
         "stores_store_liveness.go",
-        "task_pacer.go",
         "testing_knobs.go",
         "ts_maintenance_queue.go",
         ":gen-refreshraftreason-stringer",  # keep
@@ -247,6 +246,7 @@ go_library(
         "//pkg/util/stop",
         "//pkg/util/syncutil",
         "//pkg/util/syncutil/singleflight",
+        "//pkg/util/taskpacer",
         "//pkg/util/timeutil",
         "//pkg/util/tracing",
         "//pkg/util/tracing/tracingpb",
@@ -388,7 +388,6 @@ go_test(
         "store_replica_btree_test.go",
         "store_test.go",
         "stores_test.go",
-        "task_pacer_test.go",
         "testutils_test.go",
         "ts_maintenance_queue_test.go",
         "txn_recovery_integration_test.go",

pkg/kv/kvserver/store.go

@@ -99,6 +99,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/stop"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil/singleflight"
+	"github.com/cockroachdb/cockroach/pkg/util/taskpacer"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing/tracingpb"
@@ -2577,7 +2578,7 @@ func (s *Store) startRangefeedUpdater(ctx context.Context) {
 		//   the rate of processing in accordance with the time remaining until the
 		//   refresh interval ends.
 		conf := newRangeFeedUpdaterConf(s.cfg.Settings)
-		pacer := NewTaskPacer(conf)
+		pacer := taskpacer.New(conf)
 		for {
 			// Configuration may have changed between runs, load it unconditionally.
 			// This will block until an "active" configuration exists, i.e. a one with

pkg/kv/kvserver/store_raft.go

@@ -25,6 +25,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/mon"
 	"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/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/errors"
@@ -207,7 +208,7 @@ func (qs *raftReceiveQueues) SetEnforceMaxLen(enforceMaxLen bool) {
 }
 
 // raftTickPacerConf is a configuration struct for the raft tick pacer.
-// It implements the taskPacerConfig interface.
+// It implements the taskpacer.Config interface.
 type raftTickPacerConf struct {
 	store *Store
 }
@@ -216,11 +217,11 @@ func newRaftTickPacerConf(s *Store) raftTickPacerConf {
 	return raftTickPacerConf{store: s}
 }
 
-func (r raftTickPacerConf) getRefresh() time.Duration {
+func (r raftTickPacerConf) GetRefresh() time.Duration {
 	return r.store.cfg.RaftTickInterval
 }
 
-func (r raftTickPacerConf) getSmear() time.Duration {
+func (r raftTickPacerConf) GetSmear() time.Duration {
 	return r.store.cfg.RaftTickSmearInterval
 }
 
@@ -931,7 +932,7 @@ func (s *Store) raftTickLoop(ctx context.Context) {
 	// Create a config that will be used by the taskPacer, which allows us to pace
 	// the enqueuing of Raft ticks.
 	conf := newRaftTickPacerConf(s)
-	pacer := NewTaskPacer(conf)
+	pacer := taskpacer.New(conf)
 
 	for {
 		select {

pkg/kv/kvserver/store_rangefeed.go

@@ -15,7 +15,7 @@ import (
 
 // rangeFeedUpdaterConf provides configuration for the rangefeed updater job,
 // and allows watching for when it is updated.
-// rangeFeedUpdaterConf Implements the taskPacerConfig interface.
+// rangeFeedUpdaterConf Implements the taskpacer.Config interface.
 type rangeFeedUpdaterConf struct {
 	settings *cluster.Settings
 	changed  <-chan struct{}
@@ -41,8 +41,8 @@ func newRangeFeedUpdaterConf(st *cluster.Settings) rangeFeedUpdaterConf {
 // configuration, and returns it.
 func (r rangeFeedUpdaterConf) wait(ctx context.Context) error {
 	for {
-		refresh := r.getRefresh()
-		smear := r.getSmear()
+		refresh := r.GetRefresh()
+		smear := r.GetSmear()
 		if refresh != 0 && smear != 0 {
 			return nil
 		}
@@ -55,18 +55,18 @@ func (r rangeFeedUpdaterConf) wait(ctx context.Context) error {
 	}
 }
 
-// getRefresh returns the refresh interval for the rangefeed updater.
-func (r rangeFeedUpdaterConf) getRefresh() time.Duration {
+// GetRefresh returns the refresh interval for the rangefeed updater.
+func (r rangeFeedUpdaterConf) GetRefresh() time.Duration {
 	refresh := RangeFeedRefreshInterval.Get(&r.settings.SV)
 	if refresh <= 0 {
 		refresh = closedts.SideTransportCloseInterval.Get(&r.settings.SV)
 	}
 	return refresh
 }
 
-// getSmear returns the smear interval for the rangefeed updater.
-func (r rangeFeedUpdaterConf) getSmear() time.Duration {
-	refresh := r.getRefresh()
+// GetSmear returns the smear interval for the rangefeed updater.
+func (r rangeFeedUpdaterConf) GetSmear() time.Duration {
+	refresh := r.GetRefresh()
 	smear := RangeFeedSmearInterval.Get(&r.settings.SV)
 	if smear <= 0 || smear > refresh {
 		smear = refresh

pkg/kv/kvserver/store_rangefeed_test.go

@@ -95,7 +95,7 @@ func TestRangeFeedUpdaterConf(t *testing.T) {
 			if len(tc.updates) != 0 {
 				<-conf.changed // we must observe an update, otherwise the test times out
 			}
-			refresh, smear := conf.getRefresh(), conf.getSmear()
+			refresh, smear := conf.GetRefresh(), conf.GetSmear()
 			assert.Equal(t, tc.want, [...]time.Duration{refresh, smear})
 
 			ctx, cancel := context.WithTimeout(ctx, time.Millisecond)
@@ -105,7 +105,7 @@ func TestRangeFeedUpdaterConf(t *testing.T) {
 			if tc.waitErr != nil {
 				return
 			}
-			refresh, smear = conf.getRefresh(), conf.getSmear()
+			refresh, smear = conf.GetRefresh(), conf.GetSmear()
 			assert.Equal(t, tc.want, [...]time.Duration{refresh, smear})
 		})
 	}

pkg/util/taskpacer/BUILD.bazel

@@ -0,0 +1,22 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "taskpacer",
+    srcs = ["pacer.go"],
+    importpath = "github.com/cockroachdb/cockroach/pkg/util/taskpacer",
+    visibility = ["//visibility:public"],
+)
+
+go_test(
+    name = "taskpacer_test",
+    size = "small",
+    srcs = ["pacer_test.go"],
+    embed = [":taskpacer"],
+    deps = [
+        "//pkg/util/leaktest",
+        "//pkg/util/log",
+        "//pkg/util/timeutil",
+        "@com_github_stretchr_testify//assert",
+        "@com_github_stretchr_testify//require",
+    ],
+)

pkg/kv/kvserver/task_pacer.go renamed to pkg/util/taskpacer/pacer.go

@@ -3,44 +3,49 @@
 // Use of this software is governed by the CockroachDB Software License
 // included in the /LICENSE file.
 
-package kvserver
+package taskpacer
 
 import "time"
 
-// taskPacerConfig is the interface for the configuration of a taskPacer.
-type taskPacerConfig interface {
-	// getRefresh returns the interval between tasks.
-	getRefresh() time.Duration
+// Config is the interface for the configuration of a Pacer.
+type Config interface {
+	// GetRefresh returns the interval between tasks.
+	GetRefresh() time.Duration
 
-	// getSmear returns the interval between batches inside a task.
-	getSmear() time.Duration
+	// GetSmear returns the interval between batches inside a task.
+	GetSmear() time.Duration
 }
 
-// taskPacer controls the pacing of tasks to prevent overloading the system.
-type taskPacer struct {
+// Pacer controls the pacing of tasks to prevent overloading the system.
+// It's typically used to avoid having a sudden spike in runnable goroutines.
+// By spreading out work over time, we can keep the number of runnable
+// goroutines smaller, which reduces the overhead on the Go scheduler.
+type Pacer struct {
 	// taskStartTime is the time at which the task started.
 	taskStartTime time.Time
-	// conf is the configuration for the taskPacer.
-	conf taskPacerConfig
+	// conf is the configuration for the Pacer.
+	conf Config
 }
 
-func NewTaskPacer(conf taskPacerConfig) *taskPacer {
-	return &taskPacer{
+// New creates a new task pacer with the given configuration.
+func New(conf Config) *Pacer {
+	return &Pacer{
 		conf: conf,
 	}
 }
 
-func (tp *taskPacer) StartTask(now time.Time) {
+// StartTask marks the start of a new task at the given time.
+func (tp *Pacer) StartTask(now time.Time) {
 	tp.taskStartTime = now
 }
 
 // Pace returns the amount of work that should be done and the time by which it
 // should be done.
-// See the test TestTaskPacer for examples of how to use this method.
-func (tp *taskPacer) Pace(now time.Time, workLeft int) (todo int, by time.Time) {
+// See the test TestPacer for examples of how to use this method.
+func (tp *Pacer) Pace(now time.Time, workLeft int) (todo int, by time.Time) {
 	deadline := tp.GetDeadline()
 	timeLeft := deadline.Sub(now)
-	quantum := tp.conf.getSmear()
+	quantum := tp.conf.GetSmear()
 	if workLeft <= 0 || timeLeft <= 0 { // ran out of work or time
 		return workLeft, now
 	} else if quantum <= 0 { // smearing is disabled
@@ -62,6 +67,6 @@ func (tp *taskPacer) Pace(now time.Time, workLeft int) (todo int, by time.Time)
 
 // GetDeadline returns the time at which the current batch of work should be
 // done.
-func (tp *taskPacer) GetDeadline() time.Time {
-	return tp.taskStartTime.Add(tp.conf.getRefresh())
+func (tp *Pacer) GetDeadline() time.Time {
+	return tp.taskStartTime.Add(tp.conf.GetRefresh())
 }

pkg/kv/kvserver/task_pacer_test.go renamed to pkg/util/taskpacer/pacer_test.go

@@ -3,7 +3,7 @@
 // Use of this software is governed by the CockroachDB Software License
 // included in the /LICENSE file.
 
-package kvserver
+package taskpacer
 
 import (
 	"testing"
@@ -16,30 +16,30 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-// MockTaskPacerConfig is a test implementation of a task pacer configuration.
-type MockTaskPacerConfig struct {
+// MockConfig is a test implementation of a task pacer configuration.
+type MockConfig struct {
 	// refresh is the interval between task batches
 	refresh time.Duration
 	// smear is the interval between task batches
 	smear time.Duration
 }
 
-func newMockTaskPacerConfig(refresh, smear time.Duration) *MockTaskPacerConfig {
-	return &MockTaskPacerConfig{
+func newMockConfig(refresh, smear time.Duration) *MockConfig {
+	return &MockConfig{
 		refresh: refresh,
 		smear:   smear,
 	}
 }
 
-func (tp *MockTaskPacerConfig) getRefresh() time.Duration {
+func (tp *MockConfig) GetRefresh() time.Duration {
 	return tp.refresh
 }
 
-func (tp *MockTaskPacerConfig) getSmear() time.Duration {
+func (tp *MockConfig) GetSmear() time.Duration {
 	return tp.smear
 }
 
-func TestTaskPacer(t *testing.T) {
+func TestPacer(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
@@ -62,7 +62,7 @@ func TestTaskPacer(t *testing.T) {
 		wantBy:   []time.Duration{200},
 		wantDone: 55,
 	}, {
-		// If smear is set to 0, the taskPacer should not smear the work over
+		// If smear is set to 0, the Pacer should not smear the work over
 		// time.
 		desc:     "zero-smear",
 		deadline: 200, smear: 0, work: 1234,
@@ -143,8 +143,8 @@ func TestTaskPacer(t *testing.T) {
 			start := timeutil.Unix(946684800, 0) // Jan 1, 2000
 			now := start
 
-			conf := newMockTaskPacerConfig(tc.deadline, tc.smear)
-			pacer := NewTaskPacer(conf)
+			conf := newMockConfig(tc.deadline, tc.smear)
+			pacer := New(conf)
 			pacer.StartTask(now)
 
 			for work, startAt := tc.work, now; work != 0; {
@@ -171,9 +171,9 @@ func TestTaskPacer(t *testing.T) {
 	}
 }
 
-// TestTaskPacerAccommodatesConfChanges tests that the taskPacer can accommodate
+// TestPacerAccommodatesConfChanges tests that the Pacer can accommodate
 // changing the refresh and the smear intervals mid-run.
-func TestTaskPacerAccommodatesConfChanges(t *testing.T) {
+func TestPacerAccommodatesConfChanges(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
@@ -242,8 +242,8 @@ func TestTaskPacerAccommodatesConfChanges(t *testing.T) {
 			start := timeutil.Unix(946684800, 0) // Jan 1, 2000
 			now := start
 
-			conf := newMockTaskPacerConfig(tc.refreshes[0], tc.smears[0])
-			pacer := NewTaskPacer(conf)
+			conf := newMockConfig(tc.refreshes[0], tc.smears[0])
+			pacer := New(conf)
 			pacer.StartTask(now)
 
 			index := 0
@@ -274,5 +274,4 @@ func TestTaskPacerAccommodatesConfChanges(t *testing.T) {
 			assert.Equal(t, tc.wantDone, now.Sub(start))
 		})
 	}
-
 }