store/tikv: fix context cancelation for 2PC commit (#3330)

store/tikv/2pc.go

@@ -262,9 +262,21 @@ func (c *twoPhaseCommitter) doActionOnBatches(bo *Backoffer, action twoPhaseComm
 	ch := make(chan error, len(batches))
 	for _, batch := range batches {
 		go func(batch batchKeys) {
-			singleBatchBackoffer, singleBatchCancel := backoffer.Fork()
-			defer singleBatchCancel()
-			ch <- singleBatchActionFunc(singleBatchBackoffer, batch)
+			if action == actionCommit {
+				// Because the secondary batches of the commit actions are implemented to be
+				// committed asynchronously in backgroud goroutines, we should not
+				// fork a child context and call cancel() while the foreground goroutine exits.
+				// Otherwise the backgroud goroutines will be canceled execeptionally.
+				// Here we makes a new clone of the original backoffer for this goroutine
+				// exclusively to avoid the data race when using the same backoffer
+				// in concurrent goroutines.
+				singleBatchBackoffer := backoffer.Clone()
+				ch <- singleBatchActionFunc(singleBatchBackoffer, batch)
+			} else {
+				singleBatchBackoffer, singleBatchCancel := backoffer.Fork()
+				defer singleBatchCancel()
+				ch <- singleBatchActionFunc(singleBatchBackoffer, batch)
+			}
 		}(batch)
 	}
 	var err error

store/tikv/backoff.go

@@ -193,6 +193,17 @@ func (b *Backoffer) String() string {
 	return fmt.Sprintf(" backoff(%dms %s)", b.totalSleep, b.types)
 }
 
+// Clone creates a new Backoffer which keeps current Backoffer's sleep time and errors, and shares
+// current Backoffer's context.
+func (b *Backoffer) Clone() *Backoffer {
+	return &Backoffer{
+		maxSleep:   b.maxSleep,
+		totalSleep: b.totalSleep,
+		errors:     b.errors,
+		ctx:        b.ctx,
+	}
+}
+
 // Fork creates a new Backoffer which keeps current Backoffer's sleep time and errors, and holds
 // a child context of current Backoffer's context.
 func (b *Backoffer) Fork() (*Backoffer, goctx.CancelFunc) {