[3.4]Replace unnecessary Lock()/Unlock()s with RLock()/RUnlock()s

Signed-off-by: caojiamingalan <[email protected]>

auth/range_perm_cache.go

@@ -22,7 +22,7 @@ import (
 	"go.uber.org/zap"
 )
 
-func getMergedPerms(lg *zap.Logger, tx backend.BatchTx, userName string) *unifiedRangePermissions {
+func getMergedPerms(lg *zap.Logger, tx backend.ReadTx, userName string) *unifiedRangePermissions {
 	user := getUser(lg, tx, userName)
 	if user == nil {
 		return nil
@@ -135,7 +135,7 @@ func (as *authStore) isRangeOpPermitted(userName string, key, rangeEnd []byte, p
 	return checkKeyInterval(as.lg, rangePerm, key, rangeEnd, permtyp)
 }
 
-func (as *authStore) refreshRangePermCache(tx backend.BatchTx) {
+func (as *authStore) refreshRangePermCache(tx backend.ReadTx) {
 	// Note that every authentication configuration update calls this method and it invalidates the entire
 	// rangePermCache and reconstruct it based on information of users and roles stored in the backend.
 	// This can be a costly operation.

auth/store.go

@@ -318,9 +318,9 @@ func (as *authStore) Authenticate(ctx context.Context, username, password string
 		return nil, ErrAuthNotEnabled
 	}
 
-	tx := as.be.BatchTx()
-	tx.Lock()
-	defer tx.Unlock()
+	tx := as.be.ReadTx()
+	tx.RLock()
+	defer tx.RUnlock()
 
 	user := getUser(as.lg, tx, username)
 	if user == nil {
@@ -360,9 +360,9 @@ func (as *authStore) CheckPassword(username, password string) (uint64, error) {
 	// CompareHashAndPassword is very expensive, so we use closures
 	// to avoid putting it in the critical section of the tx lock.
 	revision, err := func() (uint64, error) {
-		tx := as.be.BatchTx()
-		tx.Lock()
-		defer tx.Unlock()
+		tx := as.be.ReadTx()
+		tx.RLock()
+		defer tx.RUnlock()
 
 		user = getUser(as.lg, tx, username)
 		if user == nil {
@@ -393,8 +393,8 @@ func (as *authStore) CheckPassword(username, password string) (uint64, error) {
 func (as *authStore) Recover(be backend.Backend) {
 	enabled := false
 	as.be = be
-	tx := be.BatchTx()
-	tx.Lock()
+	tx := be.ReadTx()
+	tx.RLock()
 	_, vs := tx.UnsafeRange(authBucketName, enableFlagKey, nil, 0)
 	if len(vs) == 1 {
 		if bytes.Equal(vs[0], authEnabled) {
@@ -405,7 +405,7 @@ func (as *authStore) Recover(be backend.Backend) {
 	as.setRevision(getRevision(tx))
 	as.refreshRangePermCache(tx)
 
-	tx.Unlock()
+	tx.RUnlock()
 
 	as.enabledMu.Lock()
 	as.enabled = enabled
@@ -623,10 +623,10 @@ func (as *authStore) UserGrantRole(r *pb.AuthUserGrantRoleRequest) (*pb.AuthUser
 }
 
 func (as *authStore) UserGet(r *pb.AuthUserGetRequest) (*pb.AuthUserGetResponse, error) {
-	tx := as.be.BatchTx()
-	tx.Lock()
+	tx := as.be.ReadTx()
+	tx.RLock()
 	user := getUser(as.lg, tx, r.Name)
-	tx.Unlock()
+	tx.RUnlock()
 
 	if user == nil {
 		return nil, ErrUserNotFound
@@ -638,10 +638,10 @@ func (as *authStore) UserGet(r *pb.AuthUserGetRequest) (*pb.AuthUserGetResponse,
 }
 
 func (as *authStore) UserList(r *pb.AuthUserListRequest) (*pb.AuthUserListResponse, error) {
-	tx := as.be.BatchTx()
-	tx.Lock()
+	tx := as.be.ReadTx()
+	tx.RLock()
 	users := getAllUsers(as.lg, tx)
-	tx.Unlock()
+	tx.RUnlock()
 
 	resp := &pb.AuthUserListResponse{Users: make([]string, len(users))}
 	for i := range users {
@@ -710,9 +710,9 @@ func (as *authStore) UserRevokeRole(r *pb.AuthUserRevokeRoleRequest) (*pb.AuthUs
 }
 
 func (as *authStore) RoleGet(r *pb.AuthRoleGetRequest) (*pb.AuthRoleGetResponse, error) {
-	tx := as.be.BatchTx()
-	tx.Lock()
-	defer tx.Unlock()
+	tx := as.be.ReadTx()
+	tx.RLock()
+	defer tx.RUnlock()
 
 	var resp pb.AuthRoleGetResponse
 
@@ -725,10 +725,10 @@ func (as *authStore) RoleGet(r *pb.AuthRoleGetRequest) (*pb.AuthRoleGetResponse,
 }
 
 func (as *authStore) RoleList(r *pb.AuthRoleListRequest) (*pb.AuthRoleListResponse, error) {
-	tx := as.be.BatchTx()
-	tx.Lock()
+	tx := as.be.ReadTx()
+	tx.RLock()
 	roles := getAllRoles(as.lg, tx)
-	tx.Unlock()
+	tx.RUnlock()
 
 	resp := &pb.AuthRoleListResponse{Roles: make([]string, len(roles))}
 	for i := range roles {
@@ -966,9 +966,9 @@ func (as *authStore) isOpPermitted(userName string, revision uint64, key, rangeE
 		return ErrAuthOldRevision
 	}
 
-	tx := as.be.BatchTx()
-	tx.Lock()
-	defer tx.Unlock()
+	tx := as.be.ReadTx()
+	tx.RLock()
+	defer tx.RUnlock()
 
 	user := getUser(as.lg, tx, userName)
 	if user == nil {
@@ -1012,10 +1012,10 @@ func (as *authStore) IsAdminPermitted(authInfo *AuthInfo) error {
 		return ErrUserEmpty
 	}
 
-	tx := as.be.BatchTx()
-	tx.Lock()
+	tx := as.be.ReadTx()
+	tx.RLock()
 	u := getUser(as.lg, tx, authInfo.Username)
-	tx.Unlock()
+	tx.RUnlock()
 
 	if u == nil {
 		return ErrUserNotFound
@@ -1028,7 +1028,7 @@ func (as *authStore) IsAdminPermitted(authInfo *AuthInfo) error {
 	return nil
 }
 
-func getUser(lg *zap.Logger, tx backend.BatchTx, username string) *authpb.User {
+func getUser(lg *zap.Logger, tx backend.ReadTx, username string) *authpb.User {
 	_, vs := tx.UnsafeRange(authUsersBucketName, []byte(username), nil, 0)
 	if len(vs) == 0 {
 		return nil
@@ -1050,7 +1050,7 @@ func getUser(lg *zap.Logger, tx backend.BatchTx, username string) *authpb.User {
 	return user
 }
 
-func getAllUsers(lg *zap.Logger, tx backend.BatchTx) []*authpb.User {
+func getAllUsers(lg *zap.Logger, tx backend.ReadTx) []*authpb.User {
 	var vs [][]byte
 	err := tx.UnsafeForEach(authUsersBucketName, func(k []byte, v []byte) error {
 		vs = append(vs, v)
@@ -1096,7 +1096,7 @@ func delUser(tx backend.BatchTx, username string) {
 	tx.UnsafeDelete(authUsersBucketName, []byte(username))
 }
 
-func getRole(tx backend.BatchTx, rolename string) *authpb.Role {
+func getRole(tx backend.ReadTx, rolename string) *authpb.Role {
 	_, vs := tx.UnsafeRange(authRolesBucketName, []byte(rolename), nil, 0)
 	if len(vs) == 0 {
 		return nil
@@ -1110,7 +1110,7 @@ func getRole(tx backend.BatchTx, rolename string) *authpb.Role {
 	return role
 }
 
-func getAllRoles(lg *zap.Logger, tx backend.BatchTx) []*authpb.Role {
+func getAllRoles(lg *zap.Logger, tx backend.ReadTx) []*authpb.Role {
 	_, vs := tx.UnsafeRange(authRolesBucketName, []byte{0}, []byte{0xff}, -1)
 	if len(vs) == 0 {
 		return nil
@@ -1233,7 +1233,7 @@ func (as *authStore) commitRevision(tx backend.BatchTx) {
 	tx.UnsafePut(authBucketName, revisionKey, revBytes)
 }
 
-func getRevision(tx backend.BatchTx) uint64 {
+func getRevision(tx backend.ReadTx) uint64 {
 	_, vs := tx.UnsafeRange(authBucketName, revisionKey, nil, 0)
 	if len(vs) != 1 {
 		// this can happen in the initialization phase
@@ -1463,10 +1463,10 @@ func (as *authStore) WithRoot(ctx context.Context) context.Context {
 }
 
 func (as *authStore) HasRole(user, role string) bool {
-	tx := as.be.BatchTx()
-	tx.Lock()
+	tx := as.be.ReadTx()
+	tx.RLock()
 	u := getUser(as.lg, tx, user)
-	tx.Unlock()
+	tx.RUnlock()
 
 	if u == nil {
 		if as.lg != nil {

mvcc/backend/read_tx.go

@@ -28,8 +28,13 @@ import (
 var safeRangeBucket = []byte("key")
 
 type ReadTx interface {
+	// Lock and Unlock should only be used in the following three cases:
+	// 1. When committing a transaction. Because it will reset the read tx, including the buffer;
+	// 2. When writing the pending data back to read buf, because obviously no reading is allowed when writing the read buffer;
+	// 3. When performing defragmentation, because again it will reset the read tx, including the read buffer.
 	Lock()
 	Unlock()
+	// RLock and RUnlock should be used for all other cases.
 	RLock()
 	RUnlock()
 

mvcc/kvstore.go

@@ -376,8 +376,8 @@ func (s *store) restore() error {
 	keyToLease := make(map[string]lease.LeaseID)
 
 	// restore index
-	tx := s.b.BatchTx()
-	tx.Lock()
+	tx := s.b.ReadTx()
+	tx.RLock()
 
 	_, finishedCompactBytes := tx.UnsafeRange(metaBucketName, finishedCompactKeyName, nil, 0)
 	if len(finishedCompactBytes) != 0 {
@@ -459,7 +459,7 @@ func (s *store) restore() error {
 		}
 	}
 
-	tx.Unlock()
+	tx.RUnlock()
 
 	if scheduledCompact != 0 {
 		s.compactLockfree(scheduledCompact)
@@ -575,9 +575,9 @@ func (s *store) ConsistentIndex() uint64 {
 	if ci := atomic.LoadUint64(&s.consistentIndex); ci > 0 {
 		return ci
 	}
-	tx := s.b.BatchTx()
-	tx.Lock()
-	defer tx.Unlock()
+	tx := s.b.ReadTx()
+	tx.RLock()
+	defer tx.RUnlock()
 	_, vs := tx.UnsafeRange(metaBucketName, consistentIndexKeyName, nil, 0)
 	if len(vs) == 0 {
 		return 0