config: Relax limitation of tso-update-physical-interval (#5437)

ref #5440, close #5440

Minimal valid value of config item `tso-update-physical-interval` is changed to 1ms.

Signed-off-by: pingyu <[email protected]>

Author: pingyu

server/config/config.go

@@ -91,8 +91,9 @@ type Config struct {
 	TSOSaveInterval typeutil.Duration `toml:"tso-save-interval" json:"tso-save-interval"`
 
 	// The interval to update physical part of timestamp. Usually, this config should not be set.
-	// It's only useful for test purposes.
-	// This config is only valid in 50ms to 10s. If it's configured too long or too short, it will
+	// At most 1<<18 (262144) TSOs can be generated in the interval. The smaller the value, the
+	// more TSOs provided, and at the same time consuming more CPU time.
+	// This config is only valid in 1ms to 10s. If it's configured too long or too short, it will
 	// be automatically clamped to the range.
 	TSOUpdatePhysicalInterval typeutil.Duration `toml:"tso-update-physical-interval" json:"tso-update-physical-interval"`
 
@@ -252,7 +253,7 @@ const (
 	// DefaultTSOUpdatePhysicalInterval is the default value of the config `TSOUpdatePhysicalInterval`.
 	DefaultTSOUpdatePhysicalInterval = 50 * time.Millisecond
 	maxTSOUpdatePhysicalInterval     = 10 * time.Second
-	minTSOUpdatePhysicalInterval     = 50 * time.Millisecond
+	minTSOUpdatePhysicalInterval     = 1 * time.Millisecond
 
 	defaultLogFormat = "text"
 

server/config/config_test.go

@@ -261,7 +261,7 @@ address = "localhost:9090"
 
 	// Test clamping TSOUpdatePhysicalInterval value
 	cfgData = `
-tso-update-physical-interval = "10ms"
+tso-update-physical-interval = "500ns"
 `
 	cfg = NewConfig()
 	meta, err = toml.Decode(cfgData, &cfg)

server/tso/tso.go

@@ -44,6 +44,11 @@ const (
 	maxLogical = int64(1 << 18)
 	// MaxSuffixBits indicates the max number of suffix bits.
 	MaxSuffixBits = 4
+	// jetLagWarningThreshold is the warning threshold of jetLag in `timestampOracle.UpdateTimestamp`.
+	// In case of small `updatePhysicalInterval`, the `3 * updatePhysicalInterval` would also is small,
+	// and trigger unnecessary warnings about clock offset.
+	// It's an empirical value.
+	jetLagWarningThreshold = 150 * time.Millisecond
 )
 
 // tsoObject is used to store the current TSO in memory with a RWMutex lock.
@@ -321,7 +326,7 @@ func (t *timestampOracle) UpdateTimestamp(leadership *election.Leadership) error
 	tsoCounter.WithLabelValues("save", t.dcLocation).Inc()
 
 	jetLag := typeutil.SubRealTimeByWallClock(now, prevPhysical)
-	if jetLag > 3*t.updatePhysicalInterval {
+	if jetLag > 3*t.updatePhysicalInterval && jetLag > jetLagWarningThreshold {
 		log.Warn("clock offset", zap.Duration("jet-lag", jetLag), zap.Time("prev-physical", prevPhysical), zap.Time("now", now), zap.Duration("update-physical-interval", t.updatePhysicalInterval))
 		tsoCounter.WithLabelValues("slow_save", t.dcLocation).Inc()
 	}
@@ -385,7 +390,7 @@ func (t *timestampOracle) getTS(leadership *election.Leadership, count uint32, s
 			return pdpb.Timestamp{}, errs.ErrGenerateTimestamp.FastGenByArgs("timestamp in memory has been reset")
 		}
 		if resp.GetLogical() >= maxLogical {
-			log.Error("logical part outside of max logical interval, please check ntp time",
+			log.Warn("logical part outside of max logical interval, please check ntp time, or adjust config item `tso-update-physical-interval`",
 				zap.Reflect("response", resp),
 				zap.Int("retry-count", i), errs.ZapError(errs.ErrLogicOverflow))
 			tsoCounter.WithLabelValues("logical_overflow", t.dcLocation).Inc()

tests/server/tso/global_tso_test.go

@@ -28,6 +28,8 @@ import (
 	"github.com/stretchr/testify/require"
 	"github.com/tikv/pd/pkg/grpcutil"
 	"github.com/tikv/pd/pkg/testutil"
+	"github.com/tikv/pd/pkg/typeutil"
+	"github.com/tikv/pd/server/config"
 	"github.com/tikv/pd/server/tso"
 	"github.com/tikv/pd/tests"
 )
@@ -185,3 +187,48 @@ func TestDelaySyncTimestamp(t *testing.T) {
 	re.NotNil(checkAndReturnTimestampResponse(re, req, resp))
 	re.NoError(failpoint.Disable("github.com/tikv/pd/server/tso/delaySyncTimestamp"))
 }
+
+func TestLogicalOverflow(t *testing.T) {
+	re := require.New(t)
+
+	runCase := func(updateInterval time.Duration) {
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+		cluster, err := tests.NewTestCluster(ctx, 1, func(conf *config.Config, serverName string) {
+			conf.TSOUpdatePhysicalInterval = typeutil.Duration{Duration: updateInterval}
+		})
+		defer cluster.Destroy()
+		re.NoError(err)
+		re.NoError(cluster.RunInitialServers())
+		cluster.WaitLeader()
+
+		leaderServer := cluster.GetServer(cluster.GetLeader())
+		grpcPDClient := testutil.MustNewGrpcClient(re, leaderServer.GetAddr())
+		clusterID := leaderServer.GetClusterID()
+
+		tsoClient, err := grpcPDClient.Tso(ctx)
+		re.NoError(err)
+		defer tsoClient.CloseSend()
+
+		begin := time.Now()
+		for i := 0; i < 2; i += 1 { // the 2nd request may (but not must) overflow, as max logical interval is 262144
+			req := &pdpb.TsoRequest{
+				Header:     testutil.NewRequestHeader(clusterID),
+				Count:      150000,
+				DcLocation: tso.GlobalDCLocation,
+			}
+			re.NoError(tsoClient.Send(req))
+			_, err = tsoClient.Recv()
+			re.NoError(err)
+		}
+		elapse := time.Since(begin)
+		if updateInterval >= 20*time.Millisecond { // on small interval, the physical may update before overflow
+			re.GreaterOrEqual(elapse, updateInterval)
+		}
+		re.Less(elapse, updateInterval+20*time.Millisecond) // additional 20ms for gRPC latency
+	}
+
+	for _, updateInterval := range []int{1, 5, 30, 50} {
+		runCase(time.Duration(updateInterval) * time.Millisecond)
+	}
+}