Merge branch '2.0-dev' into moc4704-2.0

pkg/bootstrap/service_test.go

@@ -152,11 +152,6 @@ func (tTxnOp *testTxnOperator) RemoveWaitLock(key uint64) {
 	panic("implement me")
 }
 
-func (tTxnOp *testTxnOperator) LockTableCount() int32 {
-	//TODO implement me
-	panic("implement me")
-}
-
 func (tTxnOp *testTxnOperator) LockSkipped(tableID uint64, mode lock.LockMode) bool {
 	//TODO implement me
 	panic("implement me")

pkg/frontend/mysql_protocol.go

@@ -249,14 +249,14 @@ type MysqlProtocolImpl struct {
 	maxClientPacketSize uint32
 
 	//the user of the client
-	username string
+	username atomic.Value
 
 	// opaque authentication response data generated by Authentication Method
 	// indicated by the plugin name field.
 	authResponse []byte
 
 	//the default database for the client
-	database string
+	database atomic.Value
 
 	// Connection attributes are key-value pairs that application programs
 	// can pass to the server at connect time.
@@ -523,15 +523,15 @@ func (mp *MysqlProtocolImpl) SetSequenceID(value uint8) {
 }
 
 func (mp *MysqlProtocolImpl) GetDatabaseName() string {
-	mp.m.Lock()
-	defer mp.m.Unlock()
-	return mp.database
+	val := mp.database.Load()
+	if val == nil {
+		return ""
+	}
+	return val.(string)
 }
 
 func (mp *MysqlProtocolImpl) SetDatabaseName(s string) {
-	mp.m.Lock()
-	defer mp.m.Unlock()
-	mp.database = s
+	mp.database.Store(s)
 }
 
 func (mp *MysqlProtocolImpl) GetAuthString() []byte {
@@ -547,15 +547,15 @@ func (mp *MysqlProtocolImpl) GetAuthResponse() []byte {
 }
 
 func (mp *MysqlProtocolImpl) GetUserName() string {
-	mp.m.Lock()
-	defer mp.m.Unlock()
-	return mp.username
+	val := mp.username.Load()
+	if val == nil {
+		return ""
+	}
+	return val.(string)
 }
 
 func (mp *MysqlProtocolImpl) SetUserName(s string) {
-	mp.m.Lock()
-	defer mp.m.Unlock()
-	mp.username = s
+	mp.username.Store(s)
 }
 
 const defaultTcp4PackageSize = 1<<14 - 66
@@ -1521,8 +1521,8 @@ func (mp *MysqlProtocolImpl) HandleHandshake(ctx context.Context, payload []byte
 		}
 
 		mp.maxClientPacketSize = resp41.maxPacketSize
-		mp.username = resp41.username
-		mp.database = resp41.database
+		mp.username.Store(resp41.username)
+		mp.database.Store(resp41.database)
 		mp.connectAttrs = resp41.connectAttrs
 	} else {
 		var resp320 response320
@@ -1543,8 +1543,8 @@ func (mp *MysqlProtocolImpl) HandleHandshake(ctx context.Context, payload []byte
 		mp.charset = "utf8mb4"
 
 		mp.maxClientPacketSize = resp320.maxPacketSize
-		mp.username = resp320.username
-		mp.database = resp320.database
+		mp.username.Store(resp320.username)
+		mp.database.Store(resp320.database)
 	}
 	return false, nil
 }
@@ -1560,7 +1560,7 @@ func (mp *MysqlProtocolImpl) Authenticate(ctx context.Context) error {
 	ses.Debugf(ctx, "authenticate user")
 	if err := mp.authenticateUser(ctx, mp.authResponse); err != nil {
 		ses.Errorf(ctx, "authenticate user failed.error:%v", err)
-		errorCode, sqlState, msg := RewriteError(err, mp.username)
+		errorCode, sqlState, msg := RewriteError(err, mp.GetUserName())
 		ses.timestampMap[TSSendErrPacketStart] = time.Now()
 		err2 := mp.sendErrPacket(errorCode, sqlState, msg)
 		ses.timestampMap[TSSendErrPacketEnd] = time.Now()

pkg/frontend/txn_test.go

@@ -910,11 +910,6 @@ func (txnop *testTxnOp) RemoveWaitLock(key uint64) {
 	panic("implement me")
 }
 
-func (txnop *testTxnOp) LockTableCount() int32 {
-	//TODO implement me
-	panic("implement me")
-}
-
 func (txnop *testTxnOp) LockSkipped(tableID uint64, mode lock.LockMode) bool {
 	//TODO implement me
 	panic("implement me")

pkg/incrservice/store_sql_test.go

@@ -265,11 +265,6 @@ func (tTxnOp *testTxnOperator) RemoveWaitLock(key uint64) {
 	panic("implement me")
 }
 
-func (tTxnOp *testTxnOperator) LockTableCount() int32 {
-	//TODO implement me
-	panic("implement me")
-}
-
 func (tTxnOp *testTxnOperator) LockSkipped(tableID uint64, mode lock.LockMode) bool {
 	//TODO implement me
 	panic("implement me")

pkg/sql/colexec/lockop/lock_op.go

@@ -674,8 +674,7 @@ func canRetryLock(table uint64, txn client.TxnOperator, err error) bool {
 		time.Sleep(defaultWaitTimeOnRetryLock)
 		return true
 	}
-	if txn.LockTableCount() > 0 &&
-		txn.HasLockTable(table) {
+	if txn.HasLockTable(table) {
 		return false
 	}
 	if moerr.IsMoErrCode(err, moerr.ErrLockTableBindChanged) ||

pkg/txn/client/operator.go

@@ -1306,15 +1306,6 @@ func (tc *txnOperator) RemoveWaitLock(key uint64) {
 	delete(tc.mu.waitLocks, key)
 }
 
-func (tc *txnOperator) LockTableCount() int32 {
-	tc.mu.RLock()
-	defer tc.mu.RUnlock()
-	if tc.mu.txn.Mode != txn.TxnMode_Pessimistic {
-		panic("lock in optimistic mode")
-	}
-	return int32(len(tc.mu.lockTables))
-}
-
 func (tc *txnOperator) GetOverview() TxnOverview {
 	tc.mu.RLock()
 	defer tc.mu.RUnlock()

pkg/txn/client/types.go

@@ -182,8 +182,6 @@ type TxnOperator interface {
 	AddWaitLock(tableID uint64, rows [][]byte, opt lock.LockOptions) uint64
 	// RemoveWaitLock remove wait lock for current txn
 	RemoveWaitLock(key uint64)
-	// LockTableCount get quality of lock table
-	LockTableCount() int32
 	// LockSkipped return true if lock need skipped.
 	LockSkipped(tableID uint64, mode lock.LockMode) bool
 

pkg/txn/storage/memorystorage/storage_txn_client.go

@@ -366,10 +366,6 @@ func (s *StorageTxnOperator) RemoveWaitLock(key uint64) {
 	panic("should not call")
 }
 
-func (s *StorageTxnOperator) LockTableCount() int32 {
-	panic("should not call")
-}
-
 func (s *StorageTxnOperator) GetOverview() client.TxnOverview {
 	panic("should not call")
 }

pkg/vm/engine/entire_engine_test.go

@@ -483,10 +483,6 @@ func (o *testOperator) RemoveWaitLock(key uint64) {
 	panic("should not call")
 }
 
-func (o *testOperator) LockTableCount() int32 {
-	panic("should not call")
-}
-
 func (o *testOperator) GetOverview() client.TxnOverview {
 	panic("should not call")
 }