gpool: register gpool into resource manager (#40410)

ref #40412
pingcap · Jan 10, 2023 · f5362f9 · f5362f9
1 parent e2a14ce
commit f5362f9
Show file tree

Hide file tree

Showing 13 changed files with 58 additions and 35 deletions.
diff --git a/resourcemanager/schedule.go b/resourcemanager/schedule.go
@@ -47,7 +47,7 @@ func (*ResourceManager) exec(pool *util.PoolContainer, cmd scheduler.Command) {
 	if cmd == scheduler.Hold {
 		return
 	}
-	if time.Since(pool.Pool.LastTunerTs()) > 200*time.Millisecond {
+	if time.Since(pool.Pool.LastTunerTs()) > util.MinSchedulerInterval.Load() {
 		con := pool.Pool.Cap()
 		switch cmd {
 		case scheduler.Downclock:

diff --git a/resourcemanager/scheduler/BUILD.bazel b/resourcemanager/scheduler/BUILD.bazel
@@ -11,6 +11,5 @@ go_library(
     deps = [
         "//resourcemanager/util",
         "//util/cpu",
-        "@org_uber_go_atomic//:atomic",
     ],
 )
diff --git a/resourcemanager/scheduler/cpu_scheduler.go b/resourcemanager/scheduler/cpu_scheduler.go
@@ -31,7 +31,7 @@ func NewCPUScheduler() *CPUScheduler {
 
 // Tune is to tune the goroutine pool
 func (*CPUScheduler) Tune(_ util.Component, pool util.GorotinuePool) Command {
-	if time.Since(pool.LastTunerTs()) < minCPUSchedulerInterval.Load() {
+	if time.Since(pool.LastTunerTs()) < util.MinSchedulerInterval.Load() {
 		return Hold
 	}
 	if cpu.GetCPUUsage() < 0.5 {

diff --git a/resourcemanager/scheduler/scheduler.go b/resourcemanager/scheduler/scheduler.go
@@ -15,14 +15,7 @@
 package scheduler
 
 import (
-	"time"
-
 	"github.com/pingcap/tidb/resourcemanager/util"
-	"go.uber.org/atomic"
-)
-
-var (
-	minCPUSchedulerInterval = atomic.NewDuration(time.Minute)
 )
 
 // Command is the command for scheduler

diff --git a/resourcemanager/util/BUILD.bazel b/resourcemanager/util/BUILD.bazel
@@ -9,7 +9,10 @@ go_library(
     ],
     importpath = "github.com/pingcap/tidb/resourcemanager/util",
     visibility = ["//visibility:public"],
-    deps = ["@com_github_pingcap_errors//:errors"],
+    deps = [
+        "@com_github_pingcap_errors//:errors",
+        "@org_uber_go_atomic//:atomic",
+    ],
 )
 
 go_test(

diff --git a/resourcemanager/util/mock_gpool.go b/resourcemanager/util/mock_gpool.go
@@ -26,8 +26,8 @@ func NewMockGPool(name string) *MockGPool {
 	return &MockGPool{name: name}
 }
 
-// Release is only for test
-func (*MockGPool) Release() {
+// ReleaseAndWait is only for test
+func (*MockGPool) ReleaseAndWait() {
 	panic("implement me")
 }
 

diff --git a/resourcemanager/util/util.go b/resourcemanager/util/util.go
@@ -14,24 +14,23 @@
 
 package util
 
-import "time"
+import (
+	"time"
+
+	"go.uber.org/atomic"
+)
+
+var (
+	// MinSchedulerInterval is the minimum interval between two scheduling.
+	MinSchedulerInterval = atomic.NewDuration(200 * time.Millisecond)
+)
 
 // GorotinuePool is a pool interface
 type GorotinuePool interface {
-	Release()
+	ReleaseAndWait()
 	Tune(size int)
 	LastTunerTs() time.Time
-	MaxInFlight() int64
-	InFlight() int64
-	MinRT() uint64
-	MaxPASS() uint64
 	Cap() int
-	// LongRTT is to represent the baseline latency by tracking a measurement of the long term, less volatile RTT.
-	LongRTT() float64
-	UpdateLongRTT(f func(float64) float64)
-	// ShortRTT is to represent the current system latency by tracking a measurement of the short time, and more volatile RTT.
-	ShortRTT() uint64
-	GetQueueSize() int64
 	Running() int
 	Name() string
 }

diff --git a/util/gpool/BUILD.bazel b/util/gpool/BUILD.bazel
@@ -8,4 +8,5 @@ go_library(
     ],
     importpath = "github.com/pingcap/tidb/util/gpool",
     visibility = ["//visibility:public"],
+    deps = ["@org_uber_go_atomic//:atomic"],
 )
diff --git a/util/gpool/gpool.go b/util/gpool/gpool.go
@@ -18,6 +18,8 @@ import (
 	"errors"
 	"sync/atomic"
 	"time"
+
+	atomicutil "go.uber.org/atomic"
 )
 
 const (
@@ -44,13 +46,16 @@ var (
 
 // BasePool is base class of pool
 type BasePool struct {
-	name      string
-	generator atomic.Uint64
+	name       string
+	lastTuneTs atomicutil.Time
+	generator  atomic.Uint64
 }
 
 // NewBasePool is to create a new BasePool.
 func NewBasePool() BasePool {
-	return BasePool{}
+	return BasePool{
+		lastTuneTs: *atomicutil.NewTime(time.Now()),
+	}
 }
 
 // SetName is to set name.
@@ -67,3 +72,13 @@ func (p *BasePool) Name() string {
 func (p *BasePool) NewTaskID() uint64 {
 	return p.generator.Add(1)
 }
+
+// LastTunerTs returns the last time when the pool was tuned.
+func (p *BasePool) LastTunerTs() time.Time {
+	return p.lastTuneTs.Load()
+}
+
+// SetLastTuneTs sets the last time when the pool was tuned.
+func (p *BasePool) SetLastTuneTs(t time.Time) {
+	p.lastTuneTs.Store(t)
+}
diff --git a/util/gpool/spmc/BUILD.bazel b/util/gpool/spmc/BUILD.bazel
@@ -11,7 +11,9 @@ go_library(
     importpath = "github.com/pingcap/tidb/util/gpool/spmc",
     visibility = ["//visibility:public"],
     deps = [
+        "//resourcemanager",
         "//resourcemanager/pooltask",
+        "//resourcemanager/util",
         "//util/gpool",
         "//util/logutil",
         "@com_github_pingcap_errors//:errors",
@@ -33,6 +35,7 @@ go_test(
     race = "on",
     deps = [
         "//resourcemanager/pooltask",
+        "//resourcemanager/util",
         "//testkit/testsetup",
         "//util",
         "//util/gpool",

diff --git a/util/gpool/spmc/spmcpool.go b/util/gpool/spmc/spmcpool.go
@@ -21,7 +21,9 @@ import (
 	"time"
 
 	"github.com/pingcap/log"
+	"github.com/pingcap/tidb/resourcemanager"
 	"github.com/pingcap/tidb/resourcemanager/pooltask"
+	"github.com/pingcap/tidb/resourcemanager/util"
 	"github.com/pingcap/tidb/util/gpool"
 	"github.com/pingcap/tidb/util/logutil"
 	atomicutil "go.uber.org/atomic"
@@ -55,7 +57,7 @@ type Pool[T any, U any, C any, CT any, TF pooltask.Context[CT]] struct {
 }
 
 // NewSPMCPool create a single producer, multiple consumer goroutine pool.
-func NewSPMCPool[T any, U any, C any, CT any, TF pooltask.Context[CT]](name string, size int32, options ...Option) (*Pool[T, U, C, CT, TF], error) {
+func NewSPMCPool[T any, U any, C any, CT any, TF pooltask.Context[CT]](name string, size int32, component util.Component, options ...Option) (*Pool[T, U, C, CT, TF], error) {
 	opts := loadOptions(options...)
 	if expiry := opts.ExpiryDuration; expiry <= 0 {
 		opts.ExpiryDuration = gpool.DefaultCleanIntervalTime
@@ -77,6 +79,10 @@ func NewSPMCPool[T any, U any, C any, CT any, TF pooltask.Context[CT]](name stri
 	result.capacity.Add(size)
 	result.workers = newWorkerLoopQueue[T, U, C, CT, TF](int(size))
 	result.cond = sync.NewCond(result.lock)
+	err := resourcemanager.GlobalResourceManager.Register(result, name, component)
+	if err != nil {
+		return nil, err
+	}
 	// Start a goroutine to clean up expired workers periodically.
 	go result.purgePeriodically()
 	return result, nil
@@ -129,6 +135,7 @@ func (p *Pool[T, U, C, CT, TF]) Tune(size int) {
 	if capacity == -1 || size <= 0 || size == capacity {
 		return
 	}
+	p.SetLastTuneTs(time.Now())
 	p.capacity.Store(int32(size))
 	if size > capacity {
 		// boost

diff --git a/util/gpool/spmc/spmcpool_benchmark_test.go b/util/gpool/spmc/spmcpool_benchmark_test.go
@@ -19,6 +19,7 @@ import (
 	"time"
 
 	"github.com/pingcap/tidb/resourcemanager/pooltask"
+	rmutil "github.com/pingcap/tidb/resourcemanager/util"
 	"github.com/pingcap/tidb/util"
 	"github.com/pingcap/tidb/util/gpool"
 )
@@ -29,7 +30,7 @@ const (
 )
 
 func BenchmarkGPool(b *testing.B) {
-	p, err := NewSPMCPool[struct{}, struct{}, int, any, pooltask.NilContext]("test", 10)
+	p, err := NewSPMCPool[struct{}, struct{}, int, any, pooltask.NilContext]("test", 10, rmutil.UNKNOWN)
 	if err != nil {
 		b.Fatal(err)
 	}

diff --git a/util/gpool/spmc/spmcpool_test.go b/util/gpool/spmc/spmcpool_test.go
@@ -20,6 +20,7 @@ import (
 	"testing"
 
 	"github.com/pingcap/tidb/resourcemanager/pooltask"
+	rmutil "github.com/pingcap/tidb/resourcemanager/util"
 	"github.com/pingcap/tidb/util"
 	"github.com/pingcap/tidb/util/gpool"
 	"github.com/stretchr/testify/require"
@@ -32,7 +33,7 @@ func TestPool(t *testing.T) {
 	myArgs := ConstArgs{a: 10}
 	// init the pool
 	// input type， output type, constArgs type
-	pool, err := NewSPMCPool[int, int, ConstArgs, any, pooltask.NilContext]("TestPool", 10)
+	pool, err := NewSPMCPool[int, int, ConstArgs, any, pooltask.NilContext]("TestPool", 10, rmutil.UNKNOWN)
 	require.NoError(t, err)
 	pool.SetConsumerFunc(func(task int, constArgs ConstArgs, ctx any) int {
 		return task + constArgs.a
@@ -76,7 +77,7 @@ func TestPoolWithEnoughCapacity(t *testing.T) {
 		poolsize    = 30
 		concurrency = 6
 	)
-	p, err := NewSPMCPool[struct{}, struct{}, int, any, pooltask.NilContext]("TestPoolWithEnoughCapa", poolsize, WithExpiryDuration(DefaultExpiredTime))
+	p, err := NewSPMCPool[struct{}, struct{}, int, any, pooltask.NilContext]("TestPoolWithEnoughCapa", poolsize, rmutil.UNKNOWN, WithExpiryDuration(DefaultExpiredTime))
 	require.NoError(t, err)
 	defer p.ReleaseAndWait()
 	p.SetConsumerFunc(func(a struct{}, b int, c any) struct{} {
@@ -128,7 +129,7 @@ func TestPoolWithoutEnoughCapacity(t *testing.T) {
 		concurrency = 2
 		poolsize    = 2
 	)
-	p, err := NewSPMCPool[struct{}, struct{}, int, any, pooltask.NilContext]("TestPoolWithoutEnoughCapa", poolsize,
+	p, err := NewSPMCPool[struct{}, struct{}, int, any, pooltask.NilContext]("TestPoolWithoutEnoughCapacity", poolsize, rmutil.UNKNOWN,
 		WithExpiryDuration(DefaultExpiredTime))
 	require.NoError(t, err)
 	defer p.ReleaseAndWait()
@@ -184,7 +185,7 @@ func TestPoolWithoutEnoughCapacityParallel(t *testing.T) {
 		concurrency = 2
 		poolsize    = 2
 	)
-	p, err := NewSPMCPool[struct{}, struct{}, int, any, pooltask.NilContext]("TestPoolWithoutEnoughCapa", poolsize,
+	p, err := NewSPMCPool[struct{}, struct{}, int, any, pooltask.NilContext]("TestPoolWithoutEnoughCapacityParallel", poolsize, rmutil.UNKNOWN,
 		WithExpiryDuration(DefaultExpiredTime), WithNonblocking(true))
 	require.NoError(t, err)
 	defer p.ReleaseAndWait()
@@ -236,7 +237,8 @@ func TestPoolWithoutEnoughCapacityParallel(t *testing.T) {
 }
 
 func TestBenchPool(t *testing.T) {
-	p, err := NewSPMCPool[struct{}, struct{}, int, any, pooltask.NilContext]("TestBenchPool", 10, WithExpiryDuration(DefaultExpiredTime))
+	p, err := NewSPMCPool[struct{}, struct{}, int, any, pooltask.NilContext]("TestBenchPool", 10,
+		rmutil.UNKNOWN, WithExpiryDuration(DefaultExpiredTime))
 	require.NoError(t, err)
 	defer p.ReleaseAndWait()
 	p.SetConsumerFunc(func(a struct{}, b int, c any) struct{} {