[wip]: scheduling mcs support gctuner

pkg/gctuner/tuner.go

@@ -25,6 +25,7 @@ import (
 	"github.com/pingcap/log"
 
 	util "github.com/tikv/pd/pkg/gogc"
+	"github.com/tikv/pd/pkg/memory"
 )
 
 var (
@@ -191,3 +192,83 @@ func calcGCPercent(inuse, threshold uint64) uint32 {
 	}
 	return gcPercent
 }
+
+// Cfg is the configuration for the GCTuner.
+type Cfg struct {
+	EnableGOGCTuner            bool
+	GCTunerThreshold           float64
+	ServerMemoryLimit          float64
+	ServerMemoryLimitGCTrigger float64
+	MemoryLimitBytes           uint64
+	MemoryLimitGCTriggerBytes  uint64
+	MemoryLimitGCTriggerRatio  float64
+	ThresholdBytes             uint64
+}
+
+// NewCfg creates a new cfg with the given configuration.
+func NewCfg(enableGOGCTuner bool, gcTunerThreshold float64, serverMemoryLimit float64, serverMemoryLimitGCTrigger float64) *Cfg {
+	return &Cfg{
+		EnableGOGCTuner:            enableGOGCTuner,
+		GCTunerThreshold:           gcTunerThreshold,
+		ServerMemoryLimit:          serverMemoryLimit,
+		ServerMemoryLimitGCTrigger: serverMemoryLimitGCTrigger,
+	}
+}
+
+// Checker is the checker for the GCTuner.
+type Checker struct {
+	cfg *Cfg
+}
+
+// NewChecker creates a new GCTunerChecker with default configuration.
+func NewChecker() *Checker {
+	cfg := NewCfg(false, 0, 0, 0)
+	return &Checker{cfg: cfg}
+}
+
+// SetCfg sets the configuration for the GCTuner.
+func (g *Checker) SetCfg(cfg *Cfg) {
+	totalMem, err := memory.MemTotal()
+	if err != nil {
+		log.Warn("fail to get total memory", zap.Error(err))
+		return
+	}
+	log.Info("memory info", zap.Uint64("total-mem", totalMem))
+	if g.cfg.EnableGOGCTuner != cfg.EnableGOGCTuner || g.cfg.GCTunerThreshold != cfg.GCTunerThreshold {
+		g.cfg.EnableGOGCTuner = cfg.EnableGOGCTuner
+		g.cfg.GCTunerThreshold = cfg.GCTunerThreshold
+		g.updateGCTuner()
+	}
+
+	newMemoryLimitGCTriggerRatio := cfg.ServerMemoryLimitGCTrigger
+	newMemoryLimitBytes := uint64(float64(totalMem) * cfg.ServerMemoryLimit)
+	newMemoryLimitGCTriggerBytes := uint64(float64(newMemoryLimitBytes) * newMemoryLimitGCTriggerRatio)
+	newThresholdBytes := uint64(float64(newMemoryLimitBytes) * cfg.GCTunerThreshold)
+	if newMemoryLimitBytes == 0 {
+		newThresholdBytes = uint64(float64(totalMem) * cfg.GCTunerThreshold)
+	}
+	g.cfg.ThresholdBytes = newThresholdBytes
+	if g.cfg.MemoryLimitBytes != newMemoryLimitBytes || g.cfg.MemoryLimitGCTriggerBytes != newMemoryLimitGCTriggerBytes {
+		g.cfg.MemoryLimitBytes = newMemoryLimitBytes
+		g.cfg.MemoryLimitGCTriggerBytes = newMemoryLimitGCTriggerBytes
+		g.cfg.MemoryLimitGCTriggerRatio = newMemoryLimitGCTriggerRatio
+		g.updateGCMemLimit()
+	}
+}
+
+func (g *Checker) updateGCTuner() {
+	Tuning(g.cfg.ThresholdBytes)
+	EnableGOGCTuner.Store(g.cfg.EnableGOGCTuner)
+	log.Info("update gc tuner",
+		zap.Bool("enable-gc-tuner", g.cfg.EnableGOGCTuner),
+		zap.Uint64("gc-threshold-bytes", g.cfg.ThresholdBytes))
+}
+
+func (g *Checker) updateGCMemLimit() {
+	memory.ServerMemoryLimit.Store(g.cfg.MemoryLimitBytes)
+	GlobalMemoryLimitTuner.SetPercentage(g.cfg.MemoryLimitGCTriggerRatio)
+	GlobalMemoryLimitTuner.UpdateMemoryLimit()
+	log.Info("update gc memory limit",
+		zap.Uint64("memory-limit-bytes", g.cfg.MemoryLimitBytes),
+		zap.Float64("memory-limit-gc-trigger-ratio", g.cfg.MemoryLimitGCTriggerRatio))
+}

pkg/mcs/scheduling/server/config/config.go

@@ -196,6 +196,7 @@ type PersistConfig struct {
 	schedule       atomic.Value
 	replication    atomic.Value
 	storeConfig    atomic.Value
+	pdServerCfg    atomic.Value
 	// schedulersUpdatingNotifier is used to notify that the schedulers have been updated.
 	// Store as `chan<- struct{}`.
 	schedulersUpdatingNotifier atomic.Value
@@ -291,6 +292,16 @@ func (o *PersistConfig) SetStoreConfig(cfg *sc.StoreConfig) {
 	o.storeConfig.Store(cfg)
 }
 
+// SetPDServerCfg sets the PD server configuration.
+func (o *PersistConfig) SetPDServerCfg(cfg *sc.PDServerCfg) {
+	o.pdServerCfg.Store(cfg)
+}
+
+// GetPDServerCfg returns the PD server configuration.
+func (o *PersistConfig) GetPDServerCfg() *sc.PDServerCfg {
+	return o.pdServerCfg.Load().(*sc.PDServerCfg)
+}
+
 // GetStoreConfig returns the TiKV store configuration.
 func (o *PersistConfig) GetStoreConfig() *sc.StoreConfig {
 	return o.storeConfig.Load().(*sc.StoreConfig)

pkg/mcs/scheduling/server/config/watcher.go

@@ -31,6 +31,7 @@ import (
 	"github.com/pingcap/log"
 
 	"github.com/tikv/pd/pkg/errs"
+	"github.com/tikv/pd/pkg/gctuner"
 	sc "github.com/tikv/pd/pkg/schedule/config"
 	"github.com/tikv/pd/pkg/schedule/schedulers"
 	"github.com/tikv/pd/pkg/storage"
@@ -65,13 +66,17 @@ type Watcher struct {
 	// schedulersController is used to trigger the scheduler's config reloading.
 	// Store as `*schedulers.Controller`.
 	schedulersController atomic.Value
+
+	// gcChecker is used to check and update the GCTuner configuration.
+	gcChecker *gctuner.Checker
 }
 
 type persistedConfig struct {
 	ClusterVersion semver.Version       `json:"cluster-version"`
 	Schedule       sc.ScheduleConfig    `json:"schedule"`
 	Replication    sc.ReplicationConfig `json:"replication"`
 	Store          sc.StoreConfig       `json:"store"`
+	PDServerCfg    sc.PDServerCfg       `json:"pd-server"`
 }
 
 // NewWatcher creates a new watcher to watch the config meta change from PD.
@@ -90,6 +95,7 @@ func NewWatcher(
 		etcdClient:                etcdClient,
 		PersistConfig:             persistConfig,
 		storage:                   storage,
+		gcChecker:                 gctuner.NewChecker(),
 	}
 	err := cw.initializeConfigWatcher()
 	if err != nil {
@@ -136,6 +142,10 @@ func (cw *Watcher) initializeConfigWatcher() error {
 		cw.SetScheduleConfig(&cfg.Schedule)
 		cw.SetReplicationConfig(&cfg.Replication)
 		cw.SetStoreConfig(&cfg.Store)
+		cw.SetPDServerCfg(&cfg.PDServerCfg)
+		serverCfg := cw.GetPDServerCfg()
+		cw.gcChecker.SetCfg(gctuner.NewCfg(serverCfg.EnableGOGCTuner, serverCfg.GCTunerThreshold,
+			serverCfg.ServerMemoryLimit, serverCfg.ServerMemoryLimitGCTrigger))
 		return nil
 	}
 	deleteFn := func(*mvccpb.KeyValue) error {

pkg/schedule/config/store_config.go

@@ -41,6 +41,18 @@ var (
 	RaftstoreV2 = "raft-kv2"
 )
 
+// PDServerCfg is the config of pd server.
+type PDServerCfg struct {
+	// ServerMemoryLimit indicates the memory limit of current process.
+	ServerMemoryLimit float64 `toml:"server-memory-limit" json:"server-memory-limit"`
+	// ServerMemoryLimitGCTrigger indicates the gc percentage of the ServerMemoryLimit.
+	ServerMemoryLimitGCTrigger float64 `toml:"server-memory-limit-gc-trigger" json:"server-memory-limit-gc-trigger"`
+	// EnableGOGCTuner is to enable GOGC tuner. it can tuner GOGC.
+	EnableGOGCTuner bool `toml:"enable-gogc-tuner" json:"enable-gogc-tuner,string"`
+	// GCTunerThreshold is the threshold of GC tuner.
+	GCTunerThreshold float64 `toml:"gc-tuner-threshold" json:"gc-tuner-threshold"`
+}
+
 // StoreConfig is the config of store like TiKV.
 // generated by https://mholt.github.io/json-to-go/.
 type StoreConfig struct {

server/cluster/cluster.go

@@ -49,7 +49,6 @@ import (
 	"github.com/tikv/pd/pkg/mcs/discovery"
 	"github.com/tikv/pd/pkg/mcs/utils/constant"
 	"github.com/tikv/pd/pkg/member"
-	"github.com/tikv/pd/pkg/memory"
 	"github.com/tikv/pd/pkg/metering"
 	"github.com/tikv/pd/pkg/progress"
 	"github.com/tikv/pd/pkg/ratelimit"
@@ -568,64 +567,18 @@ func (c *RaftCluster) startGCTuner() {
 
 	tick := time.NewTicker(gcTunerCheckCfgInterval)
 	defer tick.Stop()
-	totalMem, err := memory.MemTotal()
-	if err != nil {
-		log.Fatal("fail to get total memory", zap.Error(err))
-	}
-	log.Info("memory info", zap.Uint64("total-mem", totalMem))
 	cfg := c.opt.GetPDServerConfig()
-	enableGCTuner := cfg.EnableGOGCTuner
-	memoryLimitBytes := uint64(float64(totalMem) * cfg.ServerMemoryLimit)
-	gcThresholdBytes := uint64(float64(memoryLimitBytes) * cfg.GCTunerThreshold)
-	if memoryLimitBytes == 0 {
-		gcThresholdBytes = uint64(float64(totalMem) * cfg.GCTunerThreshold)
-	}
-	memoryLimitGCTriggerRatio := cfg.ServerMemoryLimitGCTrigger
-	memoryLimitGCTriggerBytes := uint64(float64(memoryLimitBytes) * memoryLimitGCTriggerRatio)
-	updateGCTuner := func() {
-		gctuner.Tuning(gcThresholdBytes)
-		gctuner.EnableGOGCTuner.Store(enableGCTuner)
-		log.Info("update gc tuner", zap.Bool("enable-gc-tuner", enableGCTuner),
-			zap.Uint64("gc-threshold-bytes", gcThresholdBytes))
-	}
-	updateGCMemLimit := func() {
-		memory.ServerMemoryLimit.Store(memoryLimitBytes)
-		gctuner.GlobalMemoryLimitTuner.SetPercentage(memoryLimitGCTriggerRatio)
-		gctuner.GlobalMemoryLimitTuner.UpdateMemoryLimit()
-		log.Info("update gc memory limit", zap.Uint64("memory-limit-bytes", memoryLimitBytes),
-			zap.Float64("memory-limit-gc-trigger-ratio", memoryLimitGCTriggerRatio))
-	}
-	updateGCTuner()
-	updateGCMemLimit()
-	checkAndUpdateIfCfgChange := func() {
-		cfg := c.opt.GetPDServerConfig()
-		newEnableGCTuner := cfg.EnableGOGCTuner
-		newMemoryLimitBytes := uint64(float64(totalMem) * cfg.ServerMemoryLimit)
-		newGCThresholdBytes := uint64(float64(newMemoryLimitBytes) * cfg.GCTunerThreshold)
-		if newMemoryLimitBytes == 0 {
-			newGCThresholdBytes = uint64(float64(totalMem) * cfg.GCTunerThreshold)
-		}
-		newMemoryLimitGCTriggerRatio := cfg.ServerMemoryLimitGCTrigger
-		newMemoryLimitGCTriggerBytes := uint64(float64(newMemoryLimitBytes) * newMemoryLimitGCTriggerRatio)
-		if newEnableGCTuner != enableGCTuner || newGCThresholdBytes != gcThresholdBytes {
-			enableGCTuner = newEnableGCTuner
-			gcThresholdBytes = newGCThresholdBytes
-			updateGCTuner()
-		}
-		if newMemoryLimitBytes != memoryLimitBytes || newMemoryLimitGCTriggerBytes != memoryLimitGCTriggerBytes {
-			memoryLimitBytes = newMemoryLimitBytes
-			memoryLimitGCTriggerBytes = newMemoryLimitGCTriggerBytes
-			memoryLimitGCTriggerRatio = newMemoryLimitGCTriggerRatio
-			updateGCMemLimit()
-		}
-	}
+	tunerCfg := gctuner.NewCfg(cfg.EnableGOGCTuner, cfg.GCTunerThreshold, cfg.ServerMemoryLimit, cfg.ServerMemoryLimitGCTrigger)
+	gctunerChecker := gctuner.NewChecker()
+	gctunerChecker.SetCfg(tunerCfg)
 	for {
 		select {
 		case <-c.ctx.Done():
 			log.Info("gc tuner is stopped")
 			return
 		case <-tick.C:
-			checkAndUpdateIfCfgChange()
+			tunerCfg := gctuner.NewCfg(cfg.EnableGOGCTuner, cfg.GCTunerThreshold, cfg.ServerMemoryLimit, cfg.ServerMemoryLimitGCTrigger)
+			gctunerChecker.SetCfg(tunerCfg)
 		}
 	}
 }