tikv/txn: support local latch in transaction

config/config.go

@@ -40,18 +40,19 @@ var (
 
 // Config contains configuration options.
 type Config struct {
-	Host            string `toml:"host" json:"host"`
-	Port            uint   `toml:"port" json:"port"`
-	Store           string `toml:"store" json:"store"`
-	Path            string `toml:"path" json:"path"`
-	Socket          string `toml:"socket" json:"socket"`
-	BinlogSocket    string `toml:"binlog-socket" json:"binlog-socket"`
-	Lease           string `toml:"lease" json:"lease"`
-	RunDDL          bool   `toml:"run-ddl" json:"run-ddl"`
-	SplitTable      bool   `toml:"split-table" json:"split-table"`
-	TokenLimit      uint   `toml:"token-limit" json:"token-limit"`
-	OOMAction       string `toml:"oom-action" json:"oom-action"`
-	EnableStreaming bool   `toml:"enable-streaming" json:"enable-streaming"`
+	Host                  string `toml:"host" json:"host"`
+	Port                  uint   `toml:"port" json:"port"`
+	Store                 string `toml:"store" json:"store"`
+	Path                  string `toml:"path" json:"path"`
+	Socket                string `toml:"socket" json:"socket"`
+	BinlogSocket          string `toml:"binlog-socket" json:"binlog-socket"`
+	Lease                 string `toml:"lease" json:"lease"`
+	RunDDL                bool   `toml:"run-ddl" json:"run-ddl"`
+	SplitTable            bool   `toml:"split-table" json:"split-table"`
+	TokenLimit            uint   `toml:"token-limit" json:"token-limit"`
+	OOMAction             string `toml:"oom-action" json:"oom-action"`
+	EnableStreaming       bool   `toml:"enable-streaming" json:"enable-streaming"`
+	EnableTxnLocalLatches bool   `toml:"enable-txn-local-latches" json:"enable-txn-local-latches"`
 	// Set sys variable lower-case-table-names, ref: https://dev.mysql.com/doc/refman/5.7/en/identifier-case-sensitivity.html.
 	// TODO: We actually only support mode 2, which keeps the original case, but the comparison is case-insensitive.
 	LowerCaseTableNames int `toml:"lower-case-table-names" json:"lower-case-table-names"`
@@ -222,17 +223,18 @@ type TiKVClient struct {
 }
 
 var defaultConf = Config{
-	Host:                "0.0.0.0",
-	Port:                4000,
-	Store:               "mocktikv",
-	Path:                "/tmp/tidb",
-	RunDDL:              true,
-	SplitTable:          true,
-	Lease:               "45s",
-	TokenLimit:          1000,
-	OOMAction:           "log",
-	EnableStreaming:     false,
-	LowerCaseTableNames: 2,
+	Host:                  "0.0.0.0",
+	Port:                  4000,
+	Store:                 "mocktikv",
+	Path:                  "/tmp/tidb",
+	RunDDL:                true,
+	SplitTable:            true,
+	Lease:                 "45s",
+	TokenLimit:            1000,
+	OOMAction:             "log",
+	EnableStreaming:       false,
+	EnableTxnLocalLatches: false,
+	LowerCaseTableNames:   2,
 	Log: Log{
 		Level:  "info",
 		Format: "text",

config/config.toml.example

@@ -38,6 +38,10 @@ oom-action = "log"
 # Enable coprocessor streaming.
 enable-streaming = false
 
+# Enable local latches for transactions. Enable it when
+# there are lots of conflicts between transactions.
+enable-txn-local-latches = false
+
 # Set system variable 'lower_case_table_names'
 lower-case-table-names = 2
 

executor/simple.go

@@ -206,7 +206,7 @@ func (e *SimpleExec) executeAlterUser(s *ast.AlterUserStmt) error {
 	}
 	if len(failedUsers) > 0 {
 		// Commit the transaction even if we returns error
-		err := e.ctx.Txn().Commit(sessionctx.SetConnID2Ctx(context.Background(), e.ctx))
+		err := e.ctx.Txn().Commit(sessionctx.SetCommitCtx(context.Background(), e.ctx))
 		if err != nil {
 			return errors.Trace(err)
 		}
@@ -238,7 +238,7 @@ func (e *SimpleExec) executeDropUser(s *ast.DropUserStmt) error {
 	}
 	if len(failedUsers) > 0 {
 		// Commit the transaction even if we returns error
-		err := e.ctx.Txn().Commit(sessionctx.SetConnID2Ctx(context.Background(), e.ctx))
+		err := e.ctx.Txn().Commit(sessionctx.SetCommitCtx(context.Background(), e.ctx))
 		if err != nil {
 			return errors.Trace(err)
 		}

server/conn.go

@@ -806,7 +806,7 @@ func (cc *clientConn) handleLoadData(ctx context.Context, loadDataInfo *executor
 		}
 		return errors.Trace(err)
 	}
-	return errors.Trace(txn.Commit(sessionctx.SetConnID2Ctx(ctx, loadDataInfo.Ctx)))
+	return errors.Trace(txn.Commit(sessionctx.SetCommitCtx(ctx, loadDataInfo.Ctx)))
 }
 
 // handleLoadStats does the additional work after processing the 'load stats' query.

session/session.go

@@ -320,7 +320,7 @@ func (s *session) doCommit(ctx context.Context) error {
 		schemaVer:       s.sessionVars.TxnCtx.SchemaVersion,
 		relatedTableIDs: tableIDs,
 	})
-	if err := s.txn.Commit(sessionctx.SetConnID2Ctx(ctx, s)); err != nil {
+	if err := s.txn.Commit(sessionctx.SetCommitCtx(ctx, s)); err != nil {
 		return errors.Trace(err)
 	}
 	return nil

sessionctx/context.go

@@ -112,7 +112,22 @@ type contextKey string
 // ConnID is the key in context.
 const ConnID contextKey = "conn ID"
 
-// SetConnID2Ctx sets the connection ID to context.
-func SetConnID2Ctx(ctx context.Context, sessCtx Context) context.Context {
-	return context.WithValue(ctx, ConnID, sessCtx.GetSessionVars().ConnectionID)
+// Retryable is the key in context
+const Retryable contextKey = "Retryable"
+
+// SetCommitCtx sets the variables for context before commit a transaction.
+func SetCommitCtx(ctx context.Context, sessCtx Context) context.Context {
+	ctx = context.WithValue(ctx, ConnID, sessCtx.GetSessionVars().ConnectionID)
+	retryAble := !sessCtx.GetSessionVars().TxnCtx.ForUpdate
+	return context.WithValue(ctx, Retryable, retryAble)
+}
+
+// GetRetryable returns the value of GetRetryable from the ctx.
+func GetRetryable(ctx context.Context) bool {
+	var retryAble bool
+	val := ctx.Value(Retryable)
+	if val != nil {
+		retryAble = val.(bool)
+	}
+	return retryAble
 }

store/tikv/2pc.go

@@ -563,6 +563,17 @@ func (c *twoPhaseCommitter) cleanupKeys(bo *Backoffer, keys [][]byte) error {
 // should be less than `gcRunInterval`.
 const maxTxnTimeUse = 590000
 
+func (c *twoPhaseCommitter) executeAndWriteFinishBinlog(ctx context.Context) error {
+	err := c.execute(ctx)
+	if err != nil {
+		c.writeFinishBinlog(binlog.BinlogType_Rollback, 0)
+
+	} else {
+		c.writeFinishBinlog(binlog.BinlogType_Commit, int64(c.commitTS))
+	}
+	return errors.Trace(err)
+}
+
 // execute executes the two-phase commit protocol.
 func (c *twoPhaseCommitter) execute(ctx context.Context) error {
 	defer func() {

store/tikv/2pc_test.go

@@ -43,7 +43,7 @@ func (s *testCommitterSuite) SetUpTest(c *C) {
 	client := mocktikv.NewRPCClient(s.cluster, mvccStore)
 	pdCli := &codecPDClient{mocktikv.NewPDClient(s.cluster)}
 	spkv := NewMockSafePointKV()
-	store, err := newTikvStore("mocktikv-store", pdCli, spkv, client, false)
+	store, err := newTikvStore("mocktikv-store", pdCli, spkv, client, false, true)
 	c.Assert(err, IsNil)
 	s.store = store
 	CommitMaxBackoff = 2000

store/tikv/kv.go

@@ -29,6 +29,7 @@ import (
 	"github.com/pingcap/tidb/config"
 	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/metrics"
+	"github.com/pingcap/tidb/store/tikv/latch"
 	"github.com/pingcap/tidb/store/tikv/oracle"
 	"github.com/pingcap/tidb/store/tikv/oracle/oracles"
 	"github.com/pingcap/tidb/store/tikv/tikvrpc"
@@ -71,6 +72,7 @@ func (d Driver) Open(path string) (kv.Storage, error) {
 	defer mc.Unlock()
 
 	security := config.GetGlobalConfig().Security
+	enableTxnLocalLatches := config.GetGlobalConfig().EnableTxnLocalLatches
 	etcdAddrs, disableGC, err := parsePath(path)
 	if err != nil {
 		return nil, errors.Trace(err)
@@ -105,7 +107,7 @@ func (d Driver) Open(path string) (kv.Storage, error) {
 		return nil, errors.Trace(err)
 	}
 
-	s, err := newTikvStore(uuid, &codecPDClient{pdCli}, spkv, newRPCClient(security), !disableGC)
+	s, err := newTikvStore(uuid, &codecPDClient{pdCli}, spkv, newRPCClient(security), !disableGC, enableTxnLocalLatches)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
@@ -127,6 +129,7 @@ type tikvStore struct {
 	pdClient     pd.Client
 	regionCache  *RegionCache
 	lockResolver *LockResolver
+	txnLatches   *latch.LatchesScheduler
 	gcWorker     GCHandler
 	etcdAddrs    []string
 	tlsConfig    *tls.Config
@@ -165,7 +168,7 @@ func (s *tikvStore) CheckVisibility(startTime uint64) error {
 	return nil
 }
 
-func newTikvStore(uuid string, pdClient pd.Client, spkv SafePointKV, client Client, enableGC bool) (*tikvStore, error) {
+func newTikvStore(uuid string, pdClient pd.Client, spkv SafePointKV, client Client, enableGC, enableTxnLocalLatches bool) (*tikvStore, error) {
 	o, err := oracles.NewPdOracle(pdClient, time.Duration(oracleUpdateInterval)*time.Millisecond)
 	if err != nil {
 		return nil, errors.Trace(err)
@@ -183,6 +186,11 @@ func newTikvStore(uuid string, pdClient pd.Client, spkv SafePointKV, client Clie
 		closed:      make(chan struct{}),
 	}
 	store.lockResolver = newLockResolver(store)
+	if enableTxnLocalLatches {
+		store.txnLatches = latch.NewScheduler(102400)
+	} else {
+		store.txnLatches = nil
+	}
 	store.enableGC = enableGC
 
 	go store.runSafePointChecker()
@@ -274,6 +282,10 @@ func (s *tikvStore) Close() error {
 	if err := s.client.Close(); err != nil {
 		return errors.Trace(err)
 	}
+
+	if s.txnLatches != nil {
+		s.txnLatches.Close()
+	}
 	return nil
 }
 

store/tikv/latch/scheduler.go

@@ -85,6 +85,7 @@ func (scheduler *LatchesScheduler) Lock(startTS uint64, keys [][]byte) *Lock {
 
 // UnLock unlocks a lock with commitTS.
 func (scheduler *LatchesScheduler) UnLock(lock *Lock, commitTS uint64) {
+	lock.commitTS = commitTS
 	scheduler.RLock()
 	defer scheduler.RUnlock()
 	if !scheduler.closed {

store/tikv/lock_resolver.go

@@ -76,7 +76,7 @@ func NewLockResolver(etcdAddrs []string, security config.Security) (*LockResolve
 		return nil, errors.Trace(err)
 	}
 
-	s, err := newTikvStore(uuid, &codecPDClient{pdCli}, spkv, newRPCClient(security), false)
+	s, err := newTikvStore(uuid, &codecPDClient{pdCli}, spkv, newRPCClient(security), false, false)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}

store/tikv/test_util.go

@@ -34,7 +34,7 @@ func NewTestTiKVStore(client Client, pdClient pd.Client, clientHijack func(Clien
 	// Make sure the uuid is unique.
 	uid := uuid.NewV4().String()
 	spkv := NewMockSafePointKV()
-	tikvStore, err := newTikvStore(uid, pdCli, spkv, client, false)
+	tikvStore, err := newTikvStore(uid, pdCli, spkv, client, false, true)
 	tikvStore.mock = true
 	return tikvStore, errors.Trace(err)
 }

store/tikv/txn.go

@@ -22,7 +22,6 @@ import (
 	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/metrics"
 	"github.com/pingcap/tidb/sessionctx"
-	binlog "github.com/pingcap/tipb/go-binlog"
 	log "github.com/sirupsen/logrus"
 	"golang.org/x/net/context"
 )
@@ -185,20 +184,44 @@ func (txn *tikvTxn) Commit(ctx context.Context) error {
 		connID = val.(uint64)
 	}
 	committer, err := newTwoPhaseCommitter(txn, connID)
-	if err != nil {
+	if err != nil || committer == nil {
 		return errors.Trace(err)
 	}
-	if committer == nil {
-		return nil
+
+	defer func() {
+		if err == nil {
+			txn.commitTS = committer.commitTS
+		}
+	}()
+	// latches disabled
+	if txn.store.txnLatches == nil {
+		err = committer.executeAndWriteFinishBinlog(ctx)
+		return errors.Trace(err)
 	}
-	err = committer.execute(ctx)
-	if err != nil {
-		committer.writeFinishBinlog(binlog.BinlogType_Rollback, 0)
+
+	// latches enabled
+	// for transactions not retryable, commit directly.
+	if !sessionctx.GetRetryable(ctx) {
+		err = committer.executeAndWriteFinishBinlog(ctx)
+		txn.store.txnLatches.RefreshCommitTS(committer.keys, committer.startTS)
 		return errors.Trace(err)
 	}
-	committer.writeFinishBinlog(binlog.BinlogType_Commit, int64(committer.commitTS))
-	txn.commitTS = committer.commitTS
-	return nil
+
+	// for transactions which need to acquire latches
+	lock := txn.store.txnLatches.Lock(committer.startTS, committer.keys)
+	defer func() {
+		commitTS := uint64(0)
+		if err == nil {
+			commitTS = committer.commitTS
+		}
+		txn.store.txnLatches.UnLock(lock, commitTS)
+	}()
+	if lock.IsStale() {
+		err = errors.Errorf("startTS %d is stale", txn.startTS)
+		return errors.Annotate(err, txnRetryableMark)
+	}
+	err = committer.executeAndWriteFinishBinlog(ctx)
+	return errors.Trace(err)
 }
 
 func (txn *tikvTxn) close() {