store/tikv, config: increase pessimistic ttl (#11499)

config/config.go

@@ -37,7 +37,7 @@ import (
 const (
 	MaxLogFileSize    = 4096 // MB
 	MinPessimisticTTL = time.Second * 15
-	MaxPessimisticTTL = time.Second * 60
+	MaxPessimisticTTL = time.Second * 120
 )
 
 // Valid config maps
@@ -395,7 +395,7 @@ var defaultConf = Config{
 		Enable:        false,
 		Default:       false,
 		MaxRetryCount: 256,
-		TTL:           "30s",
+		TTL:           "40s",
 	},
 }
 

config/config.toml.example

@@ -298,5 +298,5 @@ default = false
 max-retry-count = 256
 
 # default TTL in milliseconds for pessimistic lock.
-# The value must between "15s" and "60s".
-ttl = "30s"
+# The value must between "15s" and "120s".
+ttl = "40s"

config/config_test.go

@@ -203,8 +203,8 @@ func (s *testConfigSuite) TestValid(c *C) {
 	}{
 		{"14s", false},
 		{"15s", true},
-		{"60s", true},
-		{"61s", false},
+		{"120s", true},
+		{"121s", false},
 	}
 	for _, tt := range tests {
 		c1.PessimisticTxn.TTL = tt.ttl

store/tikv/2pc.go

@@ -103,10 +103,11 @@ type twoPhaseCommitter struct {
 	maxTxnTimeUse uint64
 	detail        *execdetails.CommitDetails
 	// For pessimistic transaction
-	isPessimistic bool
-	primaryKey    []byte
-	forUpdateTS   uint64
-	isFirstLock   bool
+	isPessimistic  bool
+	primaryKey     []byte
+	forUpdateTS    uint64
+	isFirstLock    bool
+	pessimisticTTL uint64
 }
 
 type mutationEx struct {
@@ -579,7 +580,7 @@ func (c *twoPhaseCommitter) pessimisticLockSingleBatch(bo *Backoffer, batch batc
 			PrimaryLock:  c.primary(),
 			StartVersion: c.startTS,
 			ForUpdateTs:  c.forUpdateTS,
-			LockTtl:      PessimisticLockTTL,
+			LockTtl:      c.pessimisticTTL,
 			IsFirstLock:  c.isFirstLock,
 		},
 		Context: pb.Context{

store/tikv/2pc_test.go

@@ -490,3 +490,43 @@ func (s *testCommitterSuite) TestPessimisticLockedKeysDedup(c *C) {
 	c.Assert(err, IsNil)
 	c.Assert(txn.lockKeys, HasLen, 2)
 }
+
+func (s *testCommitterSuite) TestPessimisticTTL(c *C) {
+	key := kv.Key("key")
+	txn := s.begin(c)
+	txn.SetOption(kv.Pessimistic, true)
+	time.Sleep(time.Millisecond * 100)
+	err := txn.LockKeys(context.Background(), txn.startTS, key)
+	c.Assert(err, IsNil)
+	time.Sleep(time.Millisecond * 100)
+	key2 := kv.Key("key2")
+	err = txn.LockKeys(context.Background(), txn.startTS, key2)
+	c.Assert(err, IsNil)
+	lockInfo := s.getLockInfo(c, key)
+	elapsedTTL := lockInfo.LockTtl - PessimisticLockTTL
+	c.Assert(elapsedTTL, GreaterEqual, uint64(100))
+	c.Assert(elapsedTTL, Less, uint64(200))
+	lockInfo2 := s.getLockInfo(c, key2)
+	c.Assert(lockInfo2.LockTtl, Equals, lockInfo.LockTtl)
+}
+
+func (s *testCommitterSuite) getLockInfo(c *C, key []byte) *kvrpcpb.LockInfo {
+	txn := s.begin(c)
+	err := txn.Set(key, key)
+	c.Assert(err, IsNil)
+	commiter, err := newTwoPhaseCommitterWithInit(txn, 1)
+	c.Assert(err, IsNil)
+	bo := NewBackoffer(context.Background(), getMaxBackoff)
+	loc, err := s.store.regionCache.LocateKey(bo, key)
+	c.Assert(err, IsNil)
+	batch := batchKeys{region: loc.Region, keys: [][]byte{key}}
+	req := commiter.buildPrewriteRequest(batch)
+	resp, err := s.store.SendReq(bo, req, loc.Region, readTimeoutShort)
+	c.Assert(err, IsNil)
+	c.Assert(resp.Prewrite, NotNil)
+	keyErrs := resp.Prewrite.Errors
+	c.Assert(keyErrs, HasLen, 1)
+	locked := keyErrs[0].Locked
+	c.Assert(locked, NotNil)
+	return locked
+}

store/tikv/txn.go

@@ -376,6 +376,11 @@ func (txn *tikvTxn) LockKeys(ctx context.Context, forUpdateTS uint64, keysInput
 				return err
 			}
 		}
+		if txn.committer.pessimisticTTL == 0 {
+			// add elapsed time to pessimistic TTL on the first LockKeys request.
+			elapsed := uint64(time.Since(txn.startTime) / time.Millisecond)
+			txn.committer.pessimisticTTL = PessimisticLockTTL + elapsed
+		}
 		var assignedPrimaryKey bool
 		if txn.committer.primaryKey == nil {
 			txn.committer.primaryKey = keys[0]