mcs: pick some tso prs (tikv#145)

* mcs: use patch method in keyspace group (tikv#6713) ref tikv#6233 Signed-off-by: lhy1024 <admin@liudos.us> Co-authored-by: ti-chi-bot[bot] <108142056+ti-chi-bot[bot]@users.noreply.github.com> * *: fix memory leak introduced by timer.After (tikv#6720) close tikv#6719 Signed-off-by: lhy1024 <admin@liudos.us> Co-authored-by: ti-chi-bot[bot] <108142056+ti-chi-bot[bot]@users.noreply.github.com> * tso: implement groupSplitPatroller to speed up the split process (tikv#6736) ref tikv#5895, close tikv#6696 Implement `groupSplitPatroller` to speed up the split process. Signed-off-by: JmPotato <ghzpotato@gmail.com> Co-authored-by: ti-chi-bot[bot] <108142056+ti-chi-bot[bot]@users.noreply.github.com> * client: fix tso service discovery at the first time for NewClientWithAPIContext (tikv#6749) close tikv#6748 After NewClientWithAPIContextV2 returns, the keyspace group should be discovered by the passed keyspace name immediately Signed-off-by: Bin Shi <binshi.bing@gmail.com> * *: add test for misusing keyspace ID when creating the client (tikv#6754) ref tikv#6747, ref tikv#6748, ref tikv#6749 Signed-off-by: Ryan Leung <rleungx@gmail.com> * tso: support multi-keyspace, fault injection and keyspace-name in pd-tso-bench (tikv#6608) ref tikv#5895 support multi-keyspace, fault injection and keyspace-name in pd-tso-bench Signed-off-by: Bin Shi <binshi.bing@gmail.com> Co-authored-by: ti-chi-bot[bot] <108142056+ti-chi-bot[bot]@users.noreply.github.com> * *: make test great again (tikv#6767) close tikv#6761 Signed-off-by: Ryan Leung <rleungx@gmail.com> * tso: implement deletedGroupCleaner to clean up the legacy TSO key (tikv#6745) close tikv#6589 - Implement `deletedGroupCleaner` to clean up the legacy TSO key. - Extract the timestamp key path constructor. Signed-off-by: JmPotato <ghzpotato@gmail.com> * Add TestUpgradingAPIandTSOClusters (tikv#6534) ref tikv#5895 Add TestUpgradingAPIandTSOClusters to test the scenario that after we restart the API cluster then restart the TSO cluster, the TSO service can still serve TSO requests normally. Signed-off-by: Bin Shi <binshi.bing@gmail.com> * client, tests: allow TSO fallback happens in TestMixedTSODeployment (tikv#6740) close tikv#6634 Introduce `WithAllowTSOFallback` client option to bypass the panic in `TestMixedTSODeployment`. Signed-off-by: JmPotato <ghzpotato@gmail.com> Co-authored-by: ti-chi-bot[bot] <108142056+ti-chi-bot[bot]@users.noreply.github.com> Signed-off-by: lhy1024 <admin@liudos.us> * tso: allow mergedTS to be zero in mergingChecker (tikv#6758) ref tikv#6589 Since it's possible that a keyspace group is to be deleted and merged before its TSO is initialized, we should allow `mergedTS` to be zero in `mergingChecker`. This PR allows this case and only block the merging when loading the TSO meets the error. Signed-off-by: JmPotato <ghzpotato@gmail.com> Co-authored-by: ti-chi-bot[bot] <108142056+ti-chi-bot[bot]@users.noreply.github.com> * *: move keyspace group primary path code to key_path.go (tikv#6755) ref tikv#5895 Signed-off-by: lhy1024 <admin@liudos.us> Co-authored-by: ti-chi-bot[bot] <108142056+ti-chi-bot[bot]@users.noreply.github.com> Signed-off-by: lhy1024 <admin@liudos.us> * mcs: add log flags (tikv#6777) ref tikv#5766 Signed-off-by: Ryan Leung <rleungx@gmail.com> Signed-off-by: lhy1024 <admin@liudos.us> * keyspace, tso, apiv2: impl the interface to merge all keyspace groups into the default (tikv#6757) ref tikv#6756 Impl the interface to merge all keyspace groups into the default. Signed-off-by: JmPotato <ghzpotato@gmail.com> Co-authored-by: ti-chi-bot[bot] <108142056+ti-chi-bot[bot]@users.noreply.github.com> * pdctl: support show keyspace group primary (tikv#6747) close tikv#6746 Signed-off-by: lhy1024 <admin@liudos.us> Co-authored-by: ti-chi-bot[bot] <108142056+ti-chi-bot[bot]@users.noreply.github.com> * pd-ctl, tests: impl the merge all keyspace groups command (tikv#6782) close tikv#6756 - Impl the merge all keyspace groups command. - Further reuse of TSO cluster-related test code. Signed-off-by: JmPotato <ghzpotato@gmail.com> Signed-off-by: lhy1024 <admin@liudos.us> * pd-ctl, tests: impl the merge all keyspace groups command (tikv#6782) close tikv#6756 - Impl the merge all keyspace groups command. - Further reuse of TSO cluster-related test code. Signed-off-by: JmPotato <ghzpotato@gmail.com> Signed-off-by: lhy1024 <admin@liudos.us> * *: fix test suite race (tikv#6784) close tikv#6772 Signed-off-by: Ryan Leung <rleungx@gmail.com> * keyspace: fix data race (tikv#6797) Signed-off-by: lhy1024 <admin@liudos.us> Co-authored-by: ti-chi-bot[bot] <108142056+ti-chi-bot[bot]@users.noreply.github.com> * pdctl: support show keyspace meta with refresh group id (tikv#6751) close tikv#6746 Signed-off-by: lhy1024 <admin@liudos.us> Co-authored-by: ti-chi-bot[bot] <108142056+ti-chi-bot[bot]@users.noreply.github.com> * mcs: Refactor ServicePath to make caller's life easier (tikv#6799) close tikv#6800 Signed-off-by: Xiaoguang Sun <sunxiaoguang@gmail.com> Co-authored-by: ti-chi-bot[bot] <108142056+ti-chi-bot[bot]@users.noreply.github.com> * Remove the lastPhysical check in dispatchClient (tikv#6812) * *: fix `TestGetTSOImmediately` (tikv#6811) close tikv#6795 Signed-off-by: Ryan Leung <rleungx@gmail.com> * fix test Signed-off-by: lhy1024 <admin@liudos.us> --------- Signed-off-by: Bin Shi <binshi.bing@gmail.com> Signed-off-by: Ryan Leung <rleungx@gmail.com> Signed-off-by: JmPotato <ghzpotato@gmail.com> Signed-off-by: lhy1024 <admin@liudos.us> Co-authored-by: ti-chi-bot[bot] <108142056+ti-chi-bot[bot]@users.noreply.github.com> Co-authored-by: JmPotato <ghzpotato@gmail.com> Co-authored-by: Bin Shi <39923490+binshi-bing@users.noreply.github.com> Co-authored-by: Ryan Leung <rleungx@gmail.com> Co-authored-by: Xiaoguang Sun <sunxiaoguang@users.noreply.github.com>
JmPotato · Jul 24, 2023 · a962d84 · a962d84
1 parent a6b64c3
commit a962d84
Show file tree

Hide file tree

Showing 67 changed files with 2,150 additions and 713 deletions.
diff --git a/.gitignore b/.gitignore
@@ -24,3 +24,4 @@ report.xml
 coverage.xml
 coverage
 *.txt
+go.work*
diff --git a/client/client.go b/client/client.go
@@ -251,6 +251,14 @@ func WithMaxErrorRetry(count int) ClientOption {
 	}
 }
 
+// WithAllowTSOFallback configures the client with `allowTSOFallback` option.
+// NOTICE: This should only be used for testing.
+func WithAllowTSOFallback() ClientOption {
+	return func(c *client) {
+		c.option.allowTSOFallback = true
+	}
+}
+
 var _ Client = (*client)(nil)
 
 // serviceModeKeeper is for service mode switching.
@@ -263,14 +271,6 @@ type serviceModeKeeper struct {
 	tsoSvcDiscovery ServiceDiscovery
 }
 
-func (k *serviceModeKeeper) SetKeyspaceID(keyspaceID uint32) {
-	k.Lock()
-	defer k.Unlock()
-	if k.serviceMode == pdpb.ServiceMode_API_SVC_MODE {
-		k.tsoSvcDiscovery.SetKeyspaceID(keyspaceID)
-	}
-}
-
 func (k *serviceModeKeeper) close() {
 	k.Lock()
 	defer k.Unlock()
@@ -378,7 +378,7 @@ func createClientWithKeyspace(
 
 	c.pdSvcDiscovery = newPDServiceDiscovery(
 		clientCtx, clientCancel, &c.wg, c.setServiceMode,
-		keyspaceID, c.svrUrls, c.tlsCfg, c.option)
+		nil, keyspaceID, c.svrUrls, c.tlsCfg, c.option)
 	if err := c.setup(); err != nil {
 		c.cancel()
 		return nil, err
@@ -490,36 +490,42 @@ func newClientWithKeyspaceName(
 		opt(c)
 	}
 
+	updateKeyspaceIDCb := func() error {
+		if err := c.initRetry(c.loadKeyspaceMeta, keyspaceName); err != nil {
+			return err
+		}
+		// c.keyspaceID is the source of truth for keyspace id.
+		c.pdSvcDiscovery.(*pdServiceDiscovery).SetKeyspaceID(c.keyspaceID)
+		return nil
+	}
+
 	// Create a PD service discovery with null keyspace id, then query the real id wth the keyspace name,
 	// finally update the keyspace id to the PD service discovery for the following interactions.
 	c.pdSvcDiscovery = newPDServiceDiscovery(
-		clientCtx, clientCancel, &c.wg, c.setServiceMode, nullKeyspaceID, c.svrUrls, c.tlsCfg, c.option)
+		clientCtx, clientCancel, &c.wg, c.setServiceMode, updateKeyspaceIDCb, nullKeyspaceID, c.svrUrls, c.tlsCfg, c.option)
 	if err := c.setup(); err != nil {
 		c.cancel()
 		return nil, err
 	}
-	if err := c.initRetry(c.loadKeyspaceMeta, keyspaceName); err != nil {
-		return nil, err
-	}
-	// We call "c.pdSvcDiscovery.SetKeyspaceID(c.keyspaceID)" after service mode already switching to API mode
-	// and tso service discovery already initialized, so here we need to set the tso_service_discovery's keyspace id too.
-	c.pdSvcDiscovery.SetKeyspaceID(c.keyspaceID)
-	c.serviceModeKeeper.SetKeyspaceID(c.keyspaceID)
 	log.Info("[pd] create pd client with endpoints and keyspace",
-		zap.Strings("pd-address", svrAddrs), zap.String("keyspace-name", keyspaceName), zap.Uint32("keyspace-id", c.keyspaceID))
+		zap.Strings("pd-address", svrAddrs),
+		zap.String("keyspace-name", keyspaceName),
+		zap.Uint32("keyspace-id", c.keyspaceID))
 	return c, nil
 }
 
 func (c *client) initRetry(f func(s string) error, str string) error {
 	var err error
+	ticker := time.NewTicker(time.Second)
+	defer ticker.Stop()
 	for i := 0; i < c.option.maxRetryTimes; i++ {
 		if err = f(str); err == nil {
 			return nil
 		}
 		select {
 		case <-c.ctx.Done():
 			return err
-		case <-time.After(time.Second):
+		case <-ticker.C:
 		}
 	}
 	return errors.WithStack(err)

diff --git a/client/option.go b/client/option.go
@@ -51,6 +51,7 @@ type option struct {
 	timeout          time.Duration
 	maxRetryTimes    int
 	enableForwarding bool
+	allowTSOFallback bool
 
 	// Dynamic options.
 	dynamicOptions [dynamicOptionCount]atomic.Value

diff --git a/client/pd_service_discovery.go b/client/pd_service_discovery.go
@@ -59,8 +59,6 @@ type ServiceDiscovery interface {
 	GetClusterID() uint64
 	// GetKeyspaceID returns the ID of the keyspace
 	GetKeyspaceID() uint32
-	// SetKeyspaceID sets the ID of the keyspace
-	SetKeyspaceID(keyspaceID uint32)
 	// GetKeyspaceGroupID returns the ID of the keyspace group
 	GetKeyspaceGroupID() uint32
 	// DiscoverServiceURLs discovers the microservice with the specified type and returns the server urls.
@@ -99,6 +97,7 @@ type ServiceDiscovery interface {
 	AddServiceAddrsSwitchedCallback(callbacks ...func())
 }
 
+type updateKeyspaceIDFunc func() error
 type tsoLocalServAddrsUpdatedFunc func(map[string]string) error
 type tsoGlobalServAddrUpdatedFunc func(string) error
 
@@ -149,8 +148,9 @@ type pdServiceDiscovery struct {
 	cancel    context.CancelFunc
 	closeOnce sync.Once
 
-	keyspaceID uint32
-	tlsCfg     *tlsutil.TLSConfig
+	updateKeyspaceIDCb updateKeyspaceIDFunc
+	keyspaceID         uint32
+	tlsCfg             *tlsutil.TLSConfig
 	// Client option.
 	option *option
 }
@@ -160,6 +160,7 @@ func newPDServiceDiscovery(
 	ctx context.Context, cancel context.CancelFunc,
 	wg *sync.WaitGroup,
 	serviceModeUpdateCb func(pdpb.ServiceMode),
+	updateKeyspaceIDCb updateKeyspaceIDFunc,
 	keyspaceID uint32,
 	urls []string, tlsCfg *tlsutil.TLSConfig, option *option,
 ) *pdServiceDiscovery {
@@ -169,6 +170,7 @@ func newPDServiceDiscovery(
 		cancel:              cancel,
 		wg:                  wg,
 		serviceModeUpdateCb: serviceModeUpdateCb,
+		updateKeyspaceIDCb:  updateKeyspaceIDCb,
 		keyspaceID:          keyspaceID,
 		tlsCfg:              tlsCfg,
 		option:              option,
@@ -192,6 +194,14 @@ func (c *pdServiceDiscovery) Init() error {
 	}
 	log.Info("[pd] init cluster id", zap.Uint64("cluster-id", c.clusterID))
 
+	// We need to update the keyspace ID before we discover and update the service mode
+	// so that TSO in API mode can be initialized with the correct keyspace ID.
+	if c.updateKeyspaceIDCb != nil {
+		if err := c.updateKeyspaceIDCb(); err != nil {
+			return err
+		}
+	}
+
 	if err := c.checkServiceModeChanged(); err != nil {
 		log.Warn("[pd] failed to check service mode and will check later", zap.Error(err))
 	}
@@ -206,14 +216,16 @@ func (c *pdServiceDiscovery) Init() error {
 
 func (c *pdServiceDiscovery) initRetry(f func() error) error {
 	var err error
+	ticker := time.NewTicker(time.Second)
+	defer ticker.Stop()
 	for i := 0; i < c.option.maxRetryTimes; i++ {
 		if err = f(); err == nil {
 			return nil
 		}
 		select {
 		case <-c.ctx.Done():
 			return err
-		case <-time.After(time.Second):
+		case <-ticker.C:
 		}
 	}
 	return errors.WithStack(err)

diff --git a/client/resource_manager_client.go b/client/resource_manager_client.go
@@ -349,6 +349,8 @@ func (c *client) tryResourceManagerConnect(ctx context.Context, connection *reso
 		err    error
 		stream rmpb.ResourceManager_AcquireTokenBucketsClient
 	)
+	ticker := time.NewTicker(retryInterval)
+	defer ticker.Stop()
 	for i := 0; i < maxRetryTimes; i++ {
 		cctx, cancel := context.WithCancel(ctx)
 		stream, err = c.resourceManagerClient().AcquireTokenBuckets(cctx)
@@ -362,7 +364,7 @@ func (c *client) tryResourceManagerConnect(ctx context.Context, connection *reso
 		select {
 		case <-ctx.Done():
 			return err
-		case <-time.After(retryInterval):
+		case <-ticker.C:
 		}
 	}
 	return err

diff --git a/client/timerpool/pool.go b/client/timerpool/pool.go
@@ -0,0 +1,43 @@
+// Copyright 2020 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+// Note: This file is copied from https://go-review.googlesource.com/c/go/+/276133
+
+package timerpool
+
+import (
+	"sync"
+	"time"
+)
+
+// GlobalTimerPool is a global pool for reusing *time.Timer.
+var GlobalTimerPool TimerPool
+
+// TimerPool is a wrapper of sync.Pool which caches *time.Timer for reuse.
+type TimerPool struct {
+	pool sync.Pool
+}
+
+// Get returns a timer with a given duration.
+func (tp *TimerPool) Get(d time.Duration) *time.Timer {
+	if v := tp.pool.Get(); v != nil {
+		timer := v.(*time.Timer)
+		timer.Reset(d)
+		return timer
+	}
+	return time.NewTimer(d)
+}
+
+// Put tries to call timer.Stop() before putting it back into pool,
+// if the timer.Stop() returns false (it has either already expired or been stopped),
+// have a shot at draining the channel with residual time if there is one.
+func (tp *TimerPool) Put(timer *time.Timer) {
+	if !timer.Stop() {
+		select {
+		case <-timer.C:
+		default:
+		}
+	}
+	tp.pool.Put(timer)
+}
diff --git a/client/timerpool/pool_test.go b/client/timerpool/pool_test.go
@@ -0,0 +1,70 @@
+// Copyright 2020 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+// Note: This file is copied from https://go-review.googlesource.com/c/go/+/276133
+
+package timerpool
+
+import (
+	"testing"
+	"time"
+)
+
+func TestTimerPool(t *testing.T) {
+	var tp TimerPool
+
+	for i := 0; i < 100; i++ {
+		timer := tp.Get(20 * time.Millisecond)
+
+		select {
+		case <-timer.C:
+			t.Errorf("timer expired too early")
+			continue
+		default:
+		}
+
+		select {
+		case <-time.After(100 * time.Millisecond):
+			t.Errorf("timer didn't expire on time")
+		case <-timer.C:
+		}
+
+		tp.Put(timer)
+	}
+}
+
+const timeout = 10 * time.Millisecond
+
+func BenchmarkTimerUtilization(b *testing.B) {
+	b.Run("TimerWithPool", func(b *testing.B) {
+		for i := 0; i < b.N; i++ {
+			t := GlobalTimerPool.Get(timeout)
+			GlobalTimerPool.Put(t)
+		}
+	})
+	b.Run("TimerWithoutPool", func(b *testing.B) {
+		for i := 0; i < b.N; i++ {
+			t := time.NewTimer(timeout)
+			t.Stop()
+		}
+	})
+}
+
+func BenchmarkTimerPoolParallel(b *testing.B) {
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			t := GlobalTimerPool.Get(timeout)
+			GlobalTimerPool.Put(t)
+		}
+	})
+}
+
+func BenchmarkTimerNativeParallel(b *testing.B) {
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			t := time.NewTimer(timeout)
+			t.Stop()
+		}
+	})
+}