*: calculate GC safe point based on global min start timestamp

config/config.go

@@ -276,9 +276,6 @@ type TiKVClient struct {
 	// CommitTimeout is the max time which command 'commit' will wait.
 	CommitTimeout string `toml:"commit-timeout" json:"commit-timeout"`
 
-	// MaxTxnTimeUse is the max time a Txn may use (in seconds) from its startTS to commitTS.
-	MaxTxnTimeUse uint `toml:"max-txn-time-use" json:"max-txn-time-use"`
-
 	// MaxBatchSize is the max batch size when calling batch commands API.
 	MaxBatchSize uint `toml:"max-batch-size" json:"max-batch-size"`
 	// If TiKV load is greater than this, TiDB will wait for a while to avoid little batch.
@@ -407,8 +404,6 @@ var defaultConf = Config{
 		GrpcKeepAliveTimeout: 3,
 		CommitTimeout:        "41s",
 
-		MaxTxnTimeUse: 590,
-
 		MaxBatchSize:      128,
 		OverloadThreshold: 200,
 		MaxBatchWaitTime:  0,
@@ -595,9 +590,6 @@ func (c *Config) Valid() error {
 	if c.TiKVClient.GrpcConnectionCount == 0 {
 		return fmt.Errorf("grpc-connection-count should be greater than 0")
 	}
-	if c.TiKVClient.MaxTxnTimeUse == 0 {
-		return fmt.Errorf("max-txn-time-use should be greater than 0")
-	}
 	if c.PessimisticTxn.TTL != "" {
 		dur, err := time.ParseDuration(c.PessimisticTxn.TTL)
 		if err != nil {

config/config.toml.example

@@ -271,11 +271,6 @@ grpc-keepalive-timeout = 3
 # Max time for commit command, must be twice bigger than raft election timeout.
 commit-timeout = "41s"
 
-# The max time a Txn may use (in seconds) from its startTS to commitTS.
-# We use it to guarantee GC worker will not influence any active txn. Please make sure that this
-# value is less than gc_life_time - 10s.
-max-txn-time-use = 590
-
 # Max batch size in gRPC.
 max-batch-size = 128
 # Overload threshold of TiKV.

domain/domain.go

@@ -425,7 +425,7 @@ func (do *Domain) infoSyncerKeeper() {
 	for {
 		select {
 		case <-ticker.C:
-			do.info.ReportMinStartTS()
+			do.info.ReportMinStartTS(do.Store())
 		case <-do.info.Done():
 			logutil.BgLogger().Info("server info syncer need to restart")
 			if err := do.info.Restart(context.Background()); err != nil {

domain/domain_test.go

@@ -16,6 +16,7 @@ package domain
 import (
 	"context"
 	"crypto/tls"
+	"math"
 	"testing"
 	"time"
 
@@ -34,6 +35,7 @@ import (
 	"github.com/pingcap/tidb/sessionctx/variable"
 	"github.com/pingcap/tidb/store/mockstore"
 	"github.com/pingcap/tidb/store/tikv/oracle"
+	"github.com/pingcap/tidb/util"
 	"github.com/pingcap/tidb/util/mock"
 	"github.com/pingcap/tidb/util/testleak"
 	dto "github.com/prometheus/client_model/go"
@@ -184,6 +186,29 @@ func TestInfo(t *testing.T) {
 	}
 }
 
+type mockSessionManager struct {
+	PS []*util.ProcessInfo
+}
+
+func (msm *mockSessionManager) ShowProcessList() map[uint64]*util.ProcessInfo {
+	ret := make(map[uint64]*util.ProcessInfo)
+	for _, item := range msm.PS {
+		ret[item.ID] = item
+	}
+	return ret
+}
+
+func (msm *mockSessionManager) GetProcessInfo(id uint64) (*util.ProcessInfo, bool) {
+	for _, item := range msm.PS {
+		if item.ID == id {
+			return item, true
+		}
+	}
+	return &util.ProcessInfo{}, false
+}
+
+func (msm *mockSessionManager) Kill(cid uint64, query bool) {}
+
 func (*testSuite) TestT(c *C) {
 	defer testleak.AfterTest(c)()
 	store, err := mockstore.NewMockTikvStore()
@@ -338,6 +363,28 @@ func (*testSuite) TestT(c *C) {
 	c.Assert(err.Error(), Equals, ErrInfoSchemaExpired.Error())
 	dom.SchemaValidator.Reset()
 
+	// Test for reporting min start timestamp.
+	infoSyncer := dom.InfoSyncer()
+	sm := &mockSessionManager{
+		PS: make([]*util.ProcessInfo, 0),
+	}
+	infoSyncer.SetSessionManager(sm)
+	beforeTS := variable.GoTimeToTS(time.Now())
+	infoSyncer.ReportMinStartTS(dom.Store())
+	afterTS := variable.GoTimeToTS(time.Now())
+	c.Assert(infoSyncer.minStartTS > beforeTS && infoSyncer.minStartTS < afterTS, IsFalse)
+	lowerLimit := time.Now().Add(-time.Duration(kv.MaxTxnTimeUse) * time.Millisecond)
+	validTS := variable.GoTimeToTS(lowerLimit.Add(time.Minute))
+	sm.PS = []*util.ProcessInfo{
+		{CurTxnStartTS: 0},
+		{CurTxnStartTS: math.MaxUint64},
+		{CurTxnStartTS: variable.GoTimeToTS(lowerLimit)},
+		{CurTxnStartTS: validTS},
+	}
+	infoSyncer.SetSessionManager(sm)
+	infoSyncer.ReportMinStartTS(dom.Store())
+	c.Assert(infoSyncer.minStartTS == validTS, IsTrue)
+
 	err = store.Close()
 	c.Assert(err, IsNil)
 	isClose := dom.isClose()

domain/info.go

@@ -17,7 +17,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"math"
 	"strconv"
 	"time"
 
@@ -27,7 +26,10 @@ import (
 	"github.com/pingcap/parser/mysql"
 	"github.com/pingcap/tidb/config"
 	"github.com/pingcap/tidb/ddl/util"
+	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/owner"
+	"github.com/pingcap/tidb/sessionctx/variable"
+	"github.com/pingcap/tidb/store/tikv/oracle"
 	util2 "github.com/pingcap/tidb/util"
 	"github.com/pingcap/tidb/util/hack"
 	"github.com/pingcap/tidb/util/logutil"
@@ -180,27 +182,38 @@ func (is *InfoSyncer) RemoveMinStartTS() {
 }
 
 // ReportMinStartTS reports self server min start timestamp to ETCD.
-func (is *InfoSyncer) ReportMinStartTS() {
+func (is *InfoSyncer) ReportMinStartTS(store kv.Storage) {
 	if is.manager == nil {
 		// Server may not start in time.
 		return
 	}
 	pl := is.manager.ShowProcessList()
-	var minStartTS uint64 = math.MaxUint64
+
+	// Calculate the lower limit of the start timestamp to avoid extremely old transaction delaying GC.
+	currentVer, err := store.CurrentVersion()
+	if err != nil {
+		logutil.BgLogger().Error("update minStartTS failed", zap.Error(err))
+		return
+	}
+	now := time.Unix(0, oracle.ExtractPhysical(currentVer.Ver)*1e6)
+	startTSLowerLimit := variable.GoTimeToTS(now.Add(-time.Duration(kv.MaxTxnTimeUse) * time.Millisecond))
+
+	minStartTS := variable.GoTimeToTS(now)
 	for _, info := range pl {
-		if info.CurTxnStartTS < minStartTS {
+		if info.CurTxnStartTS > startTSLowerLimit && info.CurTxnStartTS < minStartTS {
 			minStartTS = info.CurTxnStartTS
 		}
 	}
+
 	is.minStartTS = minStartTS
-	err := is.storeMinStartTS(context.Background())
+	err = is.storeMinStartTS(context.Background())
 	if err != nil {
 		logutil.BgLogger().Error("update minStartTS failed", zap.Error(err))
 	}
 }
 
 // Done returns a channel that closes when the info syncer is no longer being refreshed.
-func (is InfoSyncer) Done() <-chan struct{} {
+func (is *InfoSyncer) Done() <-chan struct{} {
 	if is.etcdCli == nil {
 		return make(chan struct{}, 1)
 	}

kv/kv.go

@@ -71,6 +71,10 @@ const (
 // Append UnCommitIndexKVFlag to the value indicate the index key/value is no need to commit.
 const UnCommitIndexKVFlag byte = '1'
 
+// MaxTxnTimeUse is the max time a Txn may use (in ms) from its begin to commit.
+// We use it to abort the transaction to guarantee GC worker will not influence it.
+const MaxTxnTimeUse = 24 * 60 * 60 * 1000
+
 // IsoLevel is the transaction's isolation level.
 type IsoLevel int
 

store/tikv/2pc.go

@@ -24,10 +24,8 @@ import (
 
 	"github.com/opentracing/opentracing-go"
 	"github.com/pingcap/errors"
-	"github.com/pingcap/failpoint"
 	pb "github.com/pingcap/kvproto/pkg/kvrpcpb"
 	"github.com/pingcap/parser/terror"
-	"github.com/pingcap/tidb/config"
 	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/metrics"
 	"github.com/pingcap/tidb/sessionctx/binloginfo"
@@ -95,11 +93,7 @@ type twoPhaseCommitter struct {
 	priority  pb.CommandPri
 	connID    uint64 // connID is used for log.
 	cleanWg   sync.WaitGroup
-	// maxTxnTimeUse represents max time a Txn may use (in ms) from its startTS to commitTS.
-	// We use it to guarantee GC worker will not influence any active txn. The value
-	// should be less than GC life time.
-	maxTxnTimeUse uint64
-	detail        unsafe.Pointer
+	detail    unsafe.Pointer
 
 	primaryKey     []byte
 	forUpdateTS    uint64
@@ -314,9 +308,6 @@ func (c *twoPhaseCommitter) initKeysAndMutations() error {
 			zap.Uint64("txnStartTS", txn.startTS))
 	}
 
-	// Convert from sec to ms
-	maxTxnTimeUse := uint64(config.GetGlobalConfig().TiKVClient.MaxTxnTimeUse) * 1000
-
 	// Sanity check for startTS.
 	if txn.StartTS() == math.MaxUint64 {
 		err = errors.Errorf("try to commit with invalid txnStartTS: %d", txn.StartTS())
@@ -329,7 +320,6 @@ func (c *twoPhaseCommitter) initKeysAndMutations() error {
 	commitDetail := &execdetails.CommitDetails{WriteSize: size, WriteKeys: len(keys)}
 	metrics.TiKVTxnWriteKVCountHistogram.Observe(float64(commitDetail.WriteKeys))
 	metrics.TiKVTxnWriteSizeHistogram.Observe(float64(commitDetail.WriteSize))
-	c.maxTxnTimeUse = maxTxnTimeUse
 	c.keys = keys
 	c.mutations = mutations
 	c.lockTTL = txnLockTTL(txn.startTime, size)
@@ -1034,13 +1024,7 @@ func (c *twoPhaseCommitter) execute(ctx context.Context) error {
 		return errors.Trace(err)
 	}
 
-	failpoint.Inject("tmpMaxTxnTime", func(val failpoint.Value) {
-		if tmpMaxTxnTime := uint64(val.(int)); tmpMaxTxnTime > 0 {
-			c.maxTxnTimeUse = tmpMaxTxnTime
-		}
-	})
-
-	if c.store.oracle.IsExpired(c.startTS, c.maxTxnTimeUse) {
+	if c.store.oracle.IsExpired(c.startTS, kv.MaxTxnTimeUse) {
 		err = errors.Errorf("conn %d txn takes too much time, txnStartTS: %d, comm: %d",
 			c.connID, c.startTS, c.commitTS)
 		return err

store/tikv/gcworker/gc_worker.go

@@ -29,12 +29,13 @@ import (
 	"github.com/pingcap/kvproto/pkg/metapb"
 	"github.com/pingcap/parser/terror"
 	pd "github.com/pingcap/pd/client"
-	"github.com/pingcap/tidb/config"
 	"github.com/pingcap/tidb/ddl/util"
+	"github.com/pingcap/tidb/domain"
 	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/metrics"
 	"github.com/pingcap/tidb/privilege"
 	"github.com/pingcap/tidb/session"
+	"github.com/pingcap/tidb/sessionctx/variable"
 	"github.com/pingcap/tidb/store/tikv"
 	"github.com/pingcap/tidb/store/tikv/oracle"
 	"github.com/pingcap/tidb/store/tikv/tikvrpc"
@@ -320,6 +321,30 @@ func (w *GCWorker) checkPrepare(ctx context.Context) (bool, uint64, error) {
 	return true, oracle.ComposeTS(oracle.GetPhysical(*newSafePoint), 0), nil
 }
 
+// calculateNewSafePoint uses the current global transaction min start timestamp to calculate the new safe point.
+func (w *GCWorker) calSafePointByMinStartTS(safePoint time.Time) time.Time {
+	kvs, err := w.store.GetSafePointKV().GetWithPrefix(domain.ServerMinStartTSPath)
+	if err != nil {
+		logutil.BgLogger().Warn("get all minStartTS failed", zap.Error(err))
+		return safePoint
+	}
+
+	safePointTS := variable.GoTimeToTS(safePoint)
+	for _, v := range kvs {
+		minStartTS, err := strconv.ParseUint(string(v.Value), 10, 64)
+		if err != nil {
+			logutil.BgLogger().Warn("parse minStartTS failed", zap.Error(err))
+			continue
+		}
+		if minStartTS < safePointTS {
+			safePointTS = minStartTS
+		}
+	}
+	safePoint = time.Unix(0, oracle.ExtractPhysical(safePointTS)*1e6)
+	logutil.BgLogger().Debug("calSafePointByMinStartTS", zap.Time("safePoint", safePoint))
+	return safePoint
+}
+
 func (w *GCWorker) getOracleTime() (time.Time, error) {
 	currentVer, err := w.store.CurrentVersion()
 	if err != nil {
@@ -419,23 +444,16 @@ func (w *GCWorker) checkGCInterval(now time.Time) (bool, error) {
 
 // validateGCLiftTime checks whether life time is small than min gc life time.
 func (w *GCWorker) validateGCLiftTime(lifeTime time.Duration) (time.Duration, error) {
-	minLifeTime := gcMinLifeTime
-	// max-txn-time-use value is less than gc_life_time - 10s.
-	maxTxnTime := time.Duration(config.GetGlobalConfig().TiKVClient.MaxTxnTimeUse+10) * time.Second
-	if minLifeTime < maxTxnTime {
-		minLifeTime = maxTxnTime
-	}
-
-	if lifeTime >= minLifeTime {
+	if lifeTime >= gcMinLifeTime {
 		return lifeTime, nil
 	}
 
 	logutil.BgLogger().Info("[gc worker] invalid gc life time",
 		zap.Duration("get gc life time", lifeTime),
-		zap.Duration("min gc life time", minLifeTime))
+		zap.Duration("min gc life time", gcMinLifeTime))
 
-	err := w.saveDuration(gcLifeTimeKey, minLifeTime)
-	return minLifeTime, err
+	err := w.saveDuration(gcLifeTimeKey, gcMinLifeTime)
+	return gcMinLifeTime, err
 }
 
 func (w *GCWorker) calculateNewSafePoint(now time.Time) (*time.Time, error) {
@@ -452,7 +470,7 @@ func (w *GCWorker) calculateNewSafePoint(now time.Time) (*time.Time, error) {
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
-	safePoint := now.Add(-*lifeTime)
+	safePoint := w.calSafePointByMinStartTS(now.Add(-*lifeTime))
 	// We should never decrease safePoint.
 	if lastSafePoint != nil && safePoint.Before(*lastSafePoint) {
 		logutil.BgLogger().Info("[gc worker] last safe point is later than current one."+