sysvar, config: introduce tidb_connection_concurrency_limit as an instance scope sysvar.

config/config.go

@@ -82,6 +82,9 @@ const (
 	DefExpensiveQueryTimeThreshold = 60
 	// DefMemoryUsageAlarmRatio is the threshold triggering an alarm which the memory usage of tidb-server instance exceeds.
 	DefMemoryUsageAlarmRatio = 0.8
+	// DefConnectionConcurrencyLimit is the number of sessions that can execute requests concurrently.
+	DefConnectionConcurrencyLimit = 1000
+	MaxConnectionConcurrencyLimit = 4294967295
 )
 
 // Valid config maps
@@ -117,6 +120,7 @@ var (
 				"enable-collect-execution-info": "tidb_enable_collect_execution_info",
 				"plugin.load":                   "plugin_load",
 				"plugin.dir":                    "plugin_dir",
+				"token-limit":                   "tidb_connection_concurrency_limit",
 			},
 		},
 		{
@@ -159,7 +163,7 @@ type Config struct {
 	Lease            string `toml:"lease" json:"lease"`
 	RunDDL           bool   `toml:"run-ddl" json:"run-ddl"`
 	SplitTable       bool   `toml:"split-table" json:"split-table"`
-	TokenLimit       uint   `toml:"token-limit" json:"token-limit"`
+	TokenLimit       uint32 `toml:"token-limit" json:"token-limit"`
 	OOMUseTmpStorage bool   `toml:"oom-use-tmp-storage" json:"oom-use-tmp-storage"`
 	TempStoragePath  string `toml:"tmp-storage-path" json:"tmp-storage-path"`
 	// TempStorageQuota describe the temporary storage Quota during query exector when OOMUseTmpStorage is enabled
@@ -464,6 +468,7 @@ type Instance struct {
 	EnableCollectExecutionInfo bool   `toml:"tidb_enable_collect_execution_info" json:"tidb_enable_collect_execution_info"`
 	PluginDir                  string `toml:"plugin_dir" json:"plugin_dir"`
 	PluginLoad                 string `toml:"plugin_load" json:"plugin_load"`
+	ConnectionConcurrencyLimit uint32 `toml:"tidb_connection_concurrency_limit" json:"tidb_connection_concurrency_limit"`
 }
 
 func (l *Log) getDisableTimestamp() bool {
@@ -808,6 +813,7 @@ var defaultConf = Config{
 		EnableCollectExecutionInfo:  true,
 		PluginDir:                   "/data/deploy/plugin",
 		PluginLoad:                  "",
+		ConnectionConcurrencyLimit:  DefConnectionConcurrencyLimit,
 	},
 	Status: Status{
 		ReportStatus:          true,
@@ -1101,6 +1107,9 @@ func (c *Config) Valid() error {
 	if c.IndexLimit < DefIndexLimit || c.IndexLimit > DefMaxOfIndexLimit {
 		return fmt.Errorf("index-limit should be [%d, %d]", DefIndexLimit, DefMaxOfIndexLimit)
 	}
+	if c.TokenLimit < 1 || c.TokenLimit > MaxConnectionConcurrencyLimit {
+		return fmt.Errorf("token-limit must be greater than or equal to 1 and less than or equal to %d", MaxConnectionConcurrencyLimit)
+	}
 	if c.Log.File.MaxSize > MaxLogFileSize {
 		return fmt.Errorf("invalid max log file size=%v which is larger than max=%v", c.Log.File.MaxSize, MaxLogFileSize)
 	}
@@ -1126,6 +1135,9 @@ func (c *Config) Valid() error {
 		return fmt.Errorf("tidb_memory_usage_alarm_ratio in [Instance] must be greater than or equal to 0 and less than or equal to 1")
 	}
 
+	if c.Instance.ConnectionConcurrencyLimit < 1 || c.Instance.ConnectionConcurrencyLimit > MaxConnectionConcurrencyLimit {
+		return fmt.Errorf("tidb_connection_concurrency_limit in [Instance] must be greater than or equal to 1 and less than or equal to %d", MaxConnectionConcurrencyLimit)
+	}
 	if len(c.IsolationRead.Engines) < 1 {
 		return fmt.Errorf("the number of [isolation-read]engines for isolation read should be at least 1")
 	}

config/config.toml.example

@@ -28,9 +28,6 @@ lease = "45s"
 # turn off this option if there will be a large number of tables created.
 split-table = true
 
-# The limit of concurrent executed sessions.
-token-limit = 1000
-
 # Controls whether to enable the temporary storage for some operators when a single SQL statement exceeds the memory quota specified by the memory quota.
 oom-use-tmp-storage = true
 
@@ -468,3 +465,6 @@ tidb_slow_log_threshold = 300
 # tidb_record_plan_in_slow_log is used to enable record query plan in slow log.
 # 0 is disable. 1 is enable.
 tidb_record_plan_in_slow_log = 1
+
+# The limit of concurrent executed connections.
+tidb_connection_concurrency_limit = 1000

config/config_test.go

@@ -204,7 +204,6 @@ unrecognized-option-test = true
 	require.NoError(t, err)
 
 	_, err = f.WriteString(`
-token-limit = 0
 enable-table-lock = true
 alter-primary-key = true
 delay-clean-table-lock = 5
@@ -279,7 +278,6 @@ grpc-max-send-msg-size = 40960
 	require.Equal(t, uint(6000), conf.TiKVClient.RegionCacheTTL)
 	require.Equal(t, int64(0), conf.TiKVClient.StoreLimit)
 	require.Equal(t, int64(8192), conf.TiKVClient.TTLRefreshedTxnSize)
-	require.Equal(t, uint(1000), conf.TokenLimit)
 	require.True(t, conf.EnableTableLock)
 	require.Equal(t, uint64(5), conf.DelayCleanTableLock)
 	require.Equal(t, uint64(10000), conf.SplitRegionMaxNum)
@@ -509,7 +507,10 @@ func TestConflictInstanceConfig(t *testing.T) {
 	// Just receive a warning and keep their respective values.
 	expectedConflictOptions := map[string]InstanceConfigSection{
 		"": {
-			"", map[string]string{"check-mb4-value-in-utf8": "tidb_check_mb4_value_in_utf8"},
+			"", map[string]string{
+				"check-mb4-value-in-utf8": "tidb_check_mb4_value_in_utf8",
+				"token-limit":             "tidb_connection_concurrency_limit",
+			},
 		},
 		"log": {
 			"log", map[string]string{"enable-slow-log": "tidb_enable_slow_log"},
@@ -518,17 +519,19 @@ func TestConflictInstanceConfig(t *testing.T) {
 			"performance", map[string]string{"force-priority": "tidb_force_priority"},
 		},
 	}
-	_, err = f.WriteString("check-mb4-value-in-utf8 = true \n" +
+	_, err = f.WriteString("check-mb4-value-in-utf8 = true \ntoken-limit = 10 \n" +
 		"[log] \nenable-slow-log = true \n" +
 		"[performance] \nforce-priority = \"NO_PRIORITY\"\n" +
-		"[instance] \ntidb_check_mb4_value_in_utf8 = false \ntidb_enable_slow_log = false \ntidb_force_priority = \"LOW_PRIORITY\"")
+		"[instance] \ntidb_check_mb4_value_in_utf8 = false \ntidb_connection_concurrency_limit = 100 \ntidb_enable_slow_log = false \ntidb_force_priority = \"LOW_PRIORITY\"")
 	require.NoError(t, err)
 	require.NoError(t, f.Sync())
 	err = conf.Load(configFile)
 	require.Error(t, err)
 	require.True(t, strings.Contains(err.Error(), "Conflict configuration options exists on both [instance] section and some other sections."))
 	require.False(t, conf.Instance.CheckMb4ValueInUTF8.Load())
 	require.True(t, conf.CheckMb4ValueInUTF8.Load())
+	require.Equal(t, uint32(10), conf.TokenLimit)
+	require.Equal(t, uint32(100), conf.Instance.ConnectionConcurrencyLimit)
 	require.Equal(t, true, conf.Log.EnableSlowLog.Load())
 	require.Equal(t, false, conf.Instance.EnableSlowLog.Load())
 	require.Equal(t, "NO_PRIORITY", conf.Performance.ForcePriority)

server/server.go

@@ -161,6 +161,11 @@ func (s *Server) releaseToken(token *Token) {
 	metrics.TokenGauge.Dec()
 }
 
+// SetConnectionConcurrencyLimit updates the size of concurrentLimiter.
+func (s *Server) SetConnectionConcurrencyLimit(count uint32) {
+	s.concurrentLimiter.Resize(uint(count))
+}
+
 // SetDomain use to set the server domain.
 func (s *Server) SetDomain(dom *domain.Domain) {
 	s.dom = dom
@@ -193,7 +198,7 @@ func NewServer(cfg *config.Config, driver IDriver) (*Server, error) {
 	s := &Server{
 		cfg:               cfg,
 		driver:            driver,
-		concurrentLimiter: NewTokenLimiter(cfg.TokenLimit),
+		concurrentLimiter: NewTokenLimiter(uint(cfg.Instance.ConnectionConcurrencyLimit)),
 		clients:           make(map[uint64]*clientConn),
 		globalConnID:      util.NewGlobalConnID(0, true),
 		internalSessions:  make(map[interface{}]struct{}, 100),
@@ -302,6 +307,7 @@ func NewServer(cfg *config.Config, driver IDriver) (*Server, error) {
 	rand.Seed(time.Now().UTC().UnixNano())
 
 	variable.RegisterStatistics(s)
+	variable.SetConnectionConcurrencyLimit = s.SetConnectionConcurrencyLimit
 
 	return s, nil
 }
@@ -351,7 +357,7 @@ func setTxnScope() {
 
 // Export config-related metrics
 func (s *Server) reportConfig() {
-	metrics.ConfigStatus.WithLabelValues("token-limit").Set(float64(s.cfg.TokenLimit))
+	metrics.ConfigStatus.WithLabelValues("tidb_connection_concurrency_limit").Set(float64(s.cfg.Instance.ConnectionConcurrencyLimit))
 	metrics.ConfigStatus.WithLabelValues("max-server-connections").Set(float64(s.cfg.MaxServerConnections))
 }
 

server/tokenlimiter.go

@@ -14,6 +14,10 @@
 
 package server
 
+import (
+	"sync"
+)
+
 // Token is used as a permission to keep on running.
 type Token struct {
 }
@@ -22,18 +26,38 @@ type Token struct {
 type TokenLimiter struct {
 	count uint
 	ch    chan *Token
+	mutex sync.Mutex // For synchronization.
 }
 
 // Put releases the token.
 func (tl *TokenLimiter) Put(tk *Token) {
+	tl.mutex.Lock()
+	defer tl.mutex.Unlock()
 	tl.ch <- tk
 }
 
 // Get obtains a token.
 func (tl *TokenLimiter) Get() *Token {
+	tl.mutex.Lock()
+	defer tl.mutex.Unlock()
 	return <-tl.ch
 }
 
+// Resize adjusts the size of tl.ch
+func (tl *TokenLimiter) Resize(count uint) {
+	if tl.count < count || tl.count > count && tl.count-uint(len(tl.ch)) <= count {
+		tl.mutex.Lock()
+		defer tl.mutex.Unlock()
+
+		newCh := make(chan *Token, count)
+		for i := uint(0); i < count-tl.count+uint(len(tl.ch)) && i < count; i++ {
+			newCh <- &Token{}
+		}
+		tl.count = count
+		tl.ch = newCh
+	}
+}
+
 // NewTokenLimiter creates a TokenLimiter with count tokens.
 func NewTokenLimiter(count uint) *TokenLimiter {
 	tl := &TokenLimiter{count: count, ch: make(chan *Token, count)}

sessionctx/variable/sysvar.go

@@ -419,6 +419,17 @@ var defaultSysVars = []*SysVar{
 	{Scope: ScopeInstance, Name: PluginDir, Value: "/data/deploy/plugin", ReadOnly: true, GetGlobal: func(s *SessionVars) (string, error) {
 		return config.GetGlobalConfig().Instance.PluginDir, nil
 	}},
+	{Scope: ScopeInstance, Name: TiDBConnectionConcurrencyLimit, Value: strconv.Itoa(int(config.GetGlobalConfig().Instance.ConnectionConcurrencyLimit)), Type: TypeInt, MinValue: 1, MaxValue: config.MaxConnectionConcurrencyLimit, SetGlobal: func(s *SessionVars, val string) error {
+		oldVal := atomic.LoadUint32(&config.GetGlobalConfig().Instance.ConnectionConcurrencyLimit)
+		ival := uint32(TidbOptInt64(val, int64(config.GetGlobalConfig().Instance.ConnectionConcurrencyLimit)))
+		if oldVal != ival {
+			SetConnectionConcurrencyLimit(ival)
+		}
+		atomic.StoreUint32(&config.GetGlobalConfig().Instance.ConnectionConcurrencyLimit, ival)
+		return nil
+	}, GetGlobal: func(s *SessionVars) (string, error) {
+		return strconv.FormatInt(int64(atomic.LoadUint32(&config.GetGlobalConfig().Instance.ConnectionConcurrencyLimit)), 10), nil
+	}},
 
 	/* The system variables below have GLOBAL scope  */
 	{Scope: ScopeGlobal, Name: MaxPreparedStmtCount, Value: strconv.FormatInt(DefMaxPreparedStmtCount, 10), Type: TypeInt, MinValue: -1, MaxValue: 1048576},

sessionctx/variable/tidb_vars.go

@@ -437,6 +437,9 @@ const (
 	// It can be "NO_PRIORITY", "LOW_PRIORITY", "HIGH_PRIORITY", "DELAYED"
 	TiDBForcePriority = "tidb_force_priority"
 
+	// TiDBConnectionConcurrencyLimit indicates the number of sessions that can execute requests concurrently.
+	TiDBConnectionConcurrencyLimit = "tidb_connection_concurrency_limit"
+
 	// TiDBConstraintCheckInPlace indicates to check the constraint when the SQL executing.
 	// It could hurt the performance of bulking insert when it is ON.
 	TiDBConstraintCheckInPlace = "tidb_constraint_check_in_place"
@@ -941,4 +944,6 @@ var (
 	GetMemQuotaAnalyze func() int64 = nil
 	// SetStatsCacheCapacity is the func registered by domain to set statsCache memory quota.
 	SetStatsCacheCapacity atomic.Value
+	// SetConnectionConcurrencyLimit is the func registered by server to set the limit of connection number.
+	SetConnectionConcurrencyLimit func(c uint32) = nil
 )

tidb-server/main.go

@@ -448,7 +448,7 @@ func overrideConfig(cfg *config.Config) {
 		cfg.Lease = *ddlLease
 	}
 	if actualFlags[nmTokenLimit] {
-		cfg.TokenLimit = uint(*tokenLimit)
+		cfg.Instance.ConnectionConcurrencyLimit = uint32(*tokenLimit)
 	}
 	if actualFlags[nmPluginLoad] {
 		cfg.Instance.PluginLoad = *pluginLoad
@@ -558,6 +558,8 @@ func setGlobalVars() {
 					cfg.Instance.PluginLoad = cfg.Plugin.Load
 				case "plugin.dir":
 					cfg.Instance.PluginDir = cfg.Plugin.Dir
+				case "token-limit":
+					cfg.Instance.ConnectionConcurrencyLimit = cfg.TokenLimit
 				}
 			case "log":
 				switch oldName {