config,txnkv: make txnCommitBatchSize adjustable by config

Signed-off-by: Jack Yu <jackysp@gmail.com>

config/client.go

@@ -44,6 +44,8 @@ import (
 const (
 	// DefStoreLivenessTimeout is the default value for store liveness timeout.
 	DefStoreLivenessTimeout = "1s"
+	// DefTxnCommitBatchSize recommends each RPC packet should be less than ~1MB
+	DefTxnCommitBatchSize = 16 * 1024
 )
 
 // TiKVClient is the config for tikv client.
@@ -84,6 +86,7 @@ type TiKVClient struct {
 	// TTLRefreshedTxnSize controls whether a transaction should update its TTL or not.
 	TTLRefreshedTxnSize      int64  `toml:"ttl-refreshed-txn-size" json:"ttl-refreshed-txn-size"`
 	ResolveLockLiteThreshold uint64 `toml:"resolve-lock-lite-threshold" json:"resolve-lock-lite-threshold"`
+	TxnCommitBatchSize       uint   `toml:"txn-commit-batch-size" json:"txn-commit-batch-size"`
 }
 
 // AsyncCommit is the config for the async commit feature. The switch to enable it is a system variable.
@@ -152,6 +155,7 @@ func DefaultTiKVClient() TiKVClient {
 		},
 
 		ResolveLockLiteThreshold: 16,
+		TxnCommitBatchSize:       DefTxnCommitBatchSize,
 	}
 }
 

txnkv/transaction/2pc.go

@@ -700,7 +700,7 @@ func txnLockTTL(startTime time.Time, txnSize int) uint64 {
 	// When writeSize is less than 256KB, the base ttl is defaultTTL (3s);
 	// When writeSize is 1MiB, 4MiB, or 10MiB, ttl is 6s, 12s, 20s correspondingly;
 	lockTTL := defaultLockTTL
-	if txnSize >= txnCommitBatchSize {
+	if txnSize >= int(config.GetGlobalConfig().TiKVClient.TxnCommitBatchSize) {
 		sizeMiB := float64(txnSize) / bytesPerMiB
 		lockTTL = uint64(float64(ttlFactor) * math.Sqrt(sizeMiB))
 		if lockTTL < defaultLockTTL {
@@ -874,7 +874,8 @@ func (c *twoPhaseCommitter) doActionOnGroupMutations(bo *retry.Backoffer, action
 
 	batchBuilder := newBatched(c.primary())
 	for _, group := range groups {
-		batchBuilder.appendBatchMutationsBySize(group.region, group.mutations, sizeFunc, txnCommitBatchSize)
+		batchBuilder.appendBatchMutationsBySize(group.region, group.mutations, sizeFunc,
+			int(config.GetGlobalConfig().TiKVClient.TxnCommitBatchSize))
 	}
 	firstIsPrimary := batchBuilder.setPrimary()
 
@@ -1851,10 +1852,6 @@ func (c *twoPhaseCommitter) shouldWriteBinlog() bool {
 	return c.binlog != nil
 }
 
-// TiKV recommends each RPC packet should be less than ~1MB. We keep each packet's
-// Key+Value size below 16KB.
-const txnCommitBatchSize = 16 * 1024
-
 type batchMutations struct {
 	region    locate.RegionVerID
 	mutations CommitterMutations

txnkv/transaction/test_probe.go

@@ -20,6 +20,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/tikv/client-go/v2/config"
 	"github.com/tikv/client-go/v2/internal/locate"
 	"github.com/tikv/client-go/v2/internal/retry"
 	"github.com/tikv/client-go/v2/internal/unionstore"
@@ -329,7 +330,7 @@ type ConfigProbe struct{}
 
 // GetTxnCommitBatchSize returns the batch size to commit txn.
 func (c ConfigProbe) GetTxnCommitBatchSize() uint64 {
-	return txnCommitBatchSize
+	return uint64(config.GetGlobalConfig().TiKVClient.TxnCommitBatchSize)
 }
 
 // GetPessimisticLockMaxBackoff returns pessimisticLockMaxBackoff