*: add resource manage to schedule pool by static

build/nogo_config.json

@@ -404,7 +404,8 @@
       "util/": "util code",
       "parser/": "parser code",
       "meta/": "parser code",
-      "extension/": "extension code"
+      "extension/": "extension code",
+      "resourcemanager/": "resourcemanager code"
     }
   },
   "shift": {
@@ -765,7 +766,8 @@
       "server/conn.go": "server/conn.go",
       "server/conn_stmt.go": "server/conn_stmt.go",
       "server/conn_test.go": "server/conn_test.go",
-      "extension/": "extension code"
+      "extension/": "extension code",
+      "resourcemanager/": "resourcemanager code"
     }
   },
   "SA2000": {

resourcemanager/BUILD.bazel

@@ -0,0 +1,19 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "resourcemanage",
+    srcs = [
+        "rm.go",
+        "schedule.go",
+    ],
+    importpath = "github.com/pingcap/tidb/resourcemanager",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//resourcemanager/scheduler",
+        "//resourcemanager/util",
+        "//util",
+        "//util/cpu",
+        "@com_github_pingcap_log//:log",
+        "@org_uber_go_zap//:zap",
+    ],
+)

resourcemanager/rm.go

@@ -0,0 +1,82 @@
+// Copyright 2022 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package resourcemanager
+
+import (
+	"time"
+
+	"github.com/pingcap/tidb/resourcemanager/scheduler"
+	"github.com/pingcap/tidb/resourcemanager/util"
+	tidbutil "github.com/pingcap/tidb/util"
+	"github.com/pingcap/tidb/util/cpu"
+)
+
+// GlobalResourceManager is a global resource manager
+var GlobalResourceManager = NewResourceManger()
+
+// ResourceManager is a resource manager
+type ResourceManager struct {
+	poolMap     *util.ShardPoolMap
+	scheduler   []scheduler.Scheduler
+	cpuObserver *cpu.Observer
+	exitCh      chan struct{}
+	wg          tidbutil.WaitGroupWrapper
+}
+
+// NewResourceManger is to create a new resource manager
+func NewResourceManger() *ResourceManager {
+	sc := make([]scheduler.Scheduler, 0, 1)
+	sc = append(sc, scheduler.NewGradient2Scheduler())
+	return &ResourceManager{
+		poolMap: util.NewShardPoolMap(),
+
+		exitCh:    make(chan struct{}),
+		scheduler: sc,
+	}
+}
+
+// Start is to start resource manager
+func (r *ResourceManager) Start() {
+	r.wg.Run(r.cpuObserver.Start)
+	r.wg.Run(func() {
+		tick := time.NewTicker(100 * time.Millisecond)
+		defer tick.Stop()
+		for {
+			select {
+			case <-tick.C:
+				r.schedule()
+			case <-r.exitCh:
+				return
+			}
+		}
+	})
+}
+
+// Stop is to stop resource manager
+func (r *ResourceManager) Stop() {
+	r.cpuObserver.Stop()
+	close(r.exitCh)
+	r.wg.Done()
+}
+
+// Register is to register pool into resource manager
+func (r *ResourceManager) Register(pool util.GorotinuePool, name string, component util.Component) error {
+	p := util.PoolContainer{Pool: pool, Component: component}
+	return r.registerPool(name, &p)
+}
+
+func (r *ResourceManager) registerPool(name string, pool *util.PoolContainer) error {
+	return r.poolMap.Add(name, pool)
+}

resourcemanager/schedule.go

@@ -0,0 +1,69 @@
+// Copyright 2022 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package resourcemanager
+
+import (
+	"time"
+
+	"github.com/pingcap/log"
+	"github.com/pingcap/tidb/resourcemanager/scheduler"
+	"github.com/pingcap/tidb/resourcemanager/util"
+	"go.uber.org/zap"
+)
+
+func (r *ResourceManager) schedule() {
+	r.poolMap.Iter(func(pool *util.PoolContainer) {
+		cmd := r.schedulePool(pool)
+		r.exec(pool, cmd)
+	})
+}
+
+func (r *ResourceManager) schedulePool(pool *util.PoolContainer) scheduler.Command {
+	for _, sch := range r.scheduler {
+		cmd := sch.Tune(pool.Component, pool.Pool)
+		switch cmd {
+		case scheduler.Hold:
+			continue
+		default:
+			return cmd
+		}
+	}
+	return scheduler.Hold
+}
+
+func (*ResourceManager) exec(pool *util.PoolContainer, cmd scheduler.Command) {
+	if cmd == scheduler.Hold {
+		return
+	}
+	if time.Since(pool.Pool.LastTunerTs()) > 200*time.Millisecond {
+		con := pool.Pool.Cap()
+		switch cmd {
+		case scheduler.Downclock:
+			concurrency := con - 1
+			log.Info("downclock goroutine pool",
+				zap.Int("origin concurrency", con),
+				zap.Int("concurrency", concurrency),
+				zap.String("name", pool.Pool.Name()))
+			pool.Pool.Tune(concurrency)
+		case scheduler.Overclock:
+			concurrency := con + 1
+			log.Info("overclock goroutine pool",
+				zap.Int("origin concurrency", con),
+				zap.Int("concurrency", concurrency),
+				zap.String("name", pool.Pool.Name()))
+			pool.Pool.Tune(concurrency)
+		}
+	}
+}

resourcemanager/scheduler/BUILD.bazel

@@ -0,0 +1,30 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "scheduler",
+    srcs = [
+        "gradient2_scheduler.go",
+        "scheduler.go",
+    ],
+    importpath = "github.com/pingcap/tidb/resourcemanager/scheduler",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//resourcemanager/util",
+        "//util/mathutil",
+        "@org_uber_go_atomic//:atomic",
+    ],
+)
+
+go_test(
+    name = "scheduler_test",
+    srcs = [
+        "gradient2_scheduler_test.go",
+        "mock_rms_test.go",
+    ],
+    embed = [":scheduler"],
+    flaky = True,
+    deps = [
+        "//resourcemanager/util",
+        "@com_github_stretchr_testify//require",
+    ],
+)

resourcemanager/scheduler/gradient2_scheduler.go

@@ -0,0 +1,78 @@
+// Copyright 2022 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package scheduler
+
+import (
+	"time"
+
+	"github.com/pingcap/tidb/resourcemanager/util"
+	"github.com/pingcap/tidb/util/mathutil"
+)
+
+const (
+	// defaultSmoothing is a factor, it will be used to smooth the old concurrency limit and new one.
+	defaultSmoothing float64 = 0.2
+	// defaultMinConcurrency is the minimum concurrency limit for a component.
+	defaultMinConcurrency int = 1
+	// max concurrency = user setting concurrency + delta
+	maxOverloadConcurrencyDelta int = 2
+)
+
+// Gradient2Scheduler is a scheduler using Gradient2 algorithm.
+// Gradient2 attempts to reduce the delay by obeserving the long and short window RTT and queue length to predict
+// reasonable concurrency. Using average algorithm to smooth the result for making scheduler more robust.
+type Gradient2Scheduler struct {
+	smoothing float64
+}
+
+// NewGradient2Scheduler is to create a new Gradient2Scheduler
+func NewGradient2Scheduler() *Gradient2Scheduler {
+	return &Gradient2Scheduler{
+		smoothing: defaultSmoothing,
+	}
+}
+
+// Tune is to tune the concurrency of the component
+func (b *Gradient2Scheduler) Tune(c util.Component, p util.GorotinuePool) Command {
+	newLimit := b.tune(c, p)
+	if newLimit > float64(p.Running()) {
+		return Overclock
+	} else if newLimit < float64(p.Running()) {
+		return Downclock
+	}
+	return Hold
+}
+
+func (b *Gradient2Scheduler) tune(_ util.Component, p util.GorotinuePool) float64 {
+	if time.Since(p.LastTunerTs()) < minCPUSchedulerInterval.Load() {
+		return float64(p.Running())
+	}
+
+	if p.InFlight() < int64(p.Cap())/2 {
+		return float64(p.Running())
+	}
+
+	if (p.LongRTT() / float64(p.ShortRTT())) > 2 {
+		p.UpdateLongRTT(func(old float64) float64 {
+			return old * 0.9
+		})
+	}
+
+	gradient := mathutil.Max(0.5, mathutil.Min(1.0, p.LongRTT()/float64(p.ShortRTT())))
+	newLimit := float64(p.Running())*gradient + float64(p.GetQueueSize())
+	newLimit = float64(p.Running())*(1-b.smoothing) + newLimit*b.smoothing
+	newLimit = mathutil.Max(float64(defaultMinConcurrency), mathutil.Min(float64(maxOverloadConcurrencyDelta+p.Cap()), newLimit))
+	return newLimit
+}