txn: record resolving locks

txnkv/transaction/prewrite.go

@@ -365,6 +365,7 @@ func (action actionPrewrite) handleSingleBatch(c *twoPhaseCommitter, bo *retry.B
 		}
 		start := time.Now()
 		msBeforeExpired, err := c.store.GetLockResolver().ResolveLocks(bo, c.startTS, locks)
+		defer c.store.GetLockResolver().ResolveLocksDone(c.startTS)
 		if err != nil {
 			return err
 		}

txnkv/txnlock/lock_resolver.go

@@ -62,6 +62,8 @@ type LockResolver struct {
 	resolveLockLiteThreshold uint64
 	mu                       struct {
 		sync.RWMutex
+		// currentStartTS resolving locks
+		resolving map[uint64][]Lock
 		// resolved caches resolved txns (FIFO, txn id -> txnStatus).
 		resolved       map[uint64]TxnStatus
 		recentResolved *list.List
@@ -76,13 +78,22 @@ type LockResolver struct {
 	asyncResolveCancel func()
 }
 
+// ResolvingLock stands for current resolving locks' information
+type ResolvingLock struct {
+	TxnID     uint64
+	LockTxnID uint64
+	Key       []byte
+	Primary   []byte
+}
+
 // NewLockResolver creates a new LockResolver instance.
 func NewLockResolver(store storage) *LockResolver {
 	r := &LockResolver{
 		store:                    store,
 		resolveLockLiteThreshold: config.GetGlobalConfig().TiKVClient.ResolveLockLiteThreshold,
 	}
 	r.mu.resolved = make(map[uint64]TxnStatus)
+	r.mu.resolving = make(map[uint64][]Lock)
 	r.mu.recentResolved = list.New()
 	r.asyncResolveCtx, r.asyncResolveCancel = context.WithCancel(context.Background())
 	return r
@@ -311,6 +322,7 @@ func (lr *LockResolver) BatchResolveLocks(bo *retry.Backoffer, locks []*Lock, lo
 // 3) Send `ResolveLock` cmd to the lock's region to resolve all locks belong to
 //    the same transaction.
 func (lr *LockResolver) ResolveLocks(bo *retry.Backoffer, callerStartTS uint64, locks []*Lock) (int64, error) {
+	lr.saveResolvingLocks(locks, callerStartTS)
 	ttl, _, _, err := lr.resolveLocks(bo, callerStartTS, locks, false, false)
 	return ttl, err
 }
@@ -319,9 +331,20 @@ func (lr *LockResolver) ResolveLocks(bo *retry.Backoffer, callerStartTS uint64,
 // Read operations needn't wait for resolve secondary locks and can read through(the lock's transaction is committed
 // and its commitTS is less than or equal to callerStartTS) or ignore(the lock's transaction is rolled back or its minCommitTS is pushed) the lock .
 func (lr *LockResolver) ResolveLocksForRead(bo *retry.Backoffer, callerStartTS uint64, locks []*Lock, lite bool) (int64, []uint64 /* canIgnore */, []uint64 /* canAccess */, error) {
+	lr.saveResolvingLocks(locks, callerStartTS)
 	return lr.resolveLocks(bo, callerStartTS, locks, true, lite)
 }
 
+func (lr *LockResolver) saveResolvingLocks(locks []*Lock, callerStartTS uint64) {
+	resolving := []Lock{}
+	for _, lock := range locks {
+		resolving = append(resolving, *lock)
+	}
+	lr.mu.Lock()
+	lr.mu.resolving[callerStartTS] = resolving
+	lr.mu.Unlock()
+}
+
 func (lr *LockResolver) resolveLocks(bo *retry.Backoffer, callerStartTS uint64, locks []*Lock, forRead bool, lite bool) (int64, []uint64 /* canIgnore */, []uint64 /* canAccess */, error) {
 	if lr.testingKnobs.meetLock != nil {
 		lr.testingKnobs.meetLock(locks)
@@ -425,6 +448,31 @@ func (lr *LockResolver) resolveLocks(bo *retry.Backoffer, callerStartTS uint64,
 	return msBeforeTxnExpired.value(), canIgnore, canAccess, nil
 }
 
+// ResolveLocksDone will remove resolving locks information related with callerStartTS
+func (lr *LockResolver) ResolveLocksDone(callerStartTS uint64) {
+	lr.mu.Lock()
+	delete(lr.mu.resolving, callerStartTS)
+	lr.mu.Unlock()
+}
+
+// Resolving returns the locks' information we are resolving currently.
+func (lr *LockResolver) Resolving() []ResolvingLock {
+	result := []ResolvingLock{}
+	lr.mu.RLock()
+	defer lr.mu.RUnlock()
+	for txnID, item := range lr.mu.resolving {
+		for _, lock := range item {
+			result = append(result, ResolvingLock{
+				TxnID:     txnID,
+				LockTxnID: lock.TxnID,
+				Key:       lock.Key,
+				Primary:   lock.Primary,
+			})
+		}
+	}
+	return result
+}
+
 type txnExpireTime struct {
 	initialized bool
 	txnExpire   int64

txnkv/txnlock/test_probe.go

@@ -100,3 +100,10 @@ func (l LockResolverProbe) IsNonAsyncCommitLock(err error) bool {
 	_, ok := errors.Cause(err).(*nonAsyncCommitLock)
 	return ok
 }
+
+// SetResolving set the resolving lock status for LockResolver
+func (l LockResolverProbe) SetResolving(currentStartTS uint64, locks []Lock) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	l.mu.resolving[currentStartTS] = append(l.mu.resolving[currentStartTS], locks...)
+}