refactor(meta): refactor dir lock and unlock logic. #22659556

Signed-off-by: Victor1319 <zengxuewei@oppo.com>
2026-08-02 02:00:56 +00:00 · 2024-11-11 14:44:13 +08:00 · 2024-11-11 14:44:13 +08:00 · 6e9caa9131
commit 6e9caa9131
parent 8a7a949f96
6 changed files with 139 additions and 81 deletions
--- a/libsdk/libsdk.go
+++ b/libsdk/libsdk.go
@ -475,7 +475,7 @@ func cfs_unlock_dir(id C.int64_t, path *C.char) C.int {
 	}

 	ino := info.Inode
-	if err = c.unlockDir(ino); err != nil {
+	if err = c.unlockDir(ino, 0); err != nil {
 		log.LogErrorf("unlockDir failed, ino(%v) err(%v)", ino, err)
 		return errorToStatus(err)
 	}
@ -1643,8 +1643,9 @@ func (c *client) lockDir(ino uint64, lease uint64, lockId int64) (retLockId int6
 	return c.mw.LockDir(ino, lease, lockId)
 }

-func (c *client) unlockDir(ino uint64) error {
-	return c.mw.XAttrDel_ll(ino, "dir_lock")
+func (c *client) unlockDir(ino uint64, lockId int64) error {
+	_, err := c.mw.LockDir(ino, 0, lockId)
+	return err
 }

 func (c *client) getDirLock(ino uint64) ([]byte, error) {
--- a/metanode/manager_op.go
+++ b/metanode/manager_op.go
@ -31,6 +31,7 @@ import (
 	raftProto "github.com/cubefs/cubefs/depends/tiglabs/raft/proto"
 	"github.com/cubefs/cubefs/proto"
 	"github.com/cubefs/cubefs/util"
+	"github.com/cubefs/cubefs/util/auditlog"
 	"github.com/cubefs/cubefs/util/errors"
 	"github.com/cubefs/cubefs/util/log"
 )
@ -2034,6 +2035,15 @@ func (m *metadataManager) opMetaLockDir(conn net.Conn, p *Packet, remoteAddr str
 	if !m.serveProxy(conn, mp, p) {
 		return
 	}
+
+	start := time.Now()
+	if mp.IsEnableAuditLog() {
+		defer func() {
+			err1 := fmt.Errorf("data(%s)_err(%v)_status(%s)", p.Data, err, p.GetResultMsg())
+			auditlog.LogInodeOp(remoteAddr, "", p.GetOpMsg(), req.String(), err1, time.Since(start).Milliseconds(), req.Inode, 0)
+		}()
+	}
+
 	err = mp.LockDir(req, p)
 	_ = m.respondToClient(conn, p)
 	log.LogDebugf("%s [opMetaLockDir] req: %d - %v, resp: %v, body: %s",
--- a/metanode/partition_fsmop_extend.go
+++ b/metanode/partition_fsmop_extend.go
@ -17,99 +17,130 @@ package metanode
 import (
 	"fmt"
 	"math"
-	"strconv"
-	"strings"
-	"time"

 	"github.com/cubefs/cubefs/proto"
 	"github.com/cubefs/cubefs/util/log"
-	"github.com/google/uuid"
 )

+const innerDirLockKey = "cfs_inner_xattr_dir_lock_key"
+
 type ExtendOpResult struct {
 	Status uint8
 	Extend *Extend
 }

 func (mp *metaPartition) fsmLockDir(req *proto.LockDirRequest) (resp *proto.LockDirResponse) {
+	if req.Lease == 0 {
+		return mp.fsmUnlockDir(req)
+	}
+
 	mp.xattrLock.Lock()
 	defer mp.xattrLock.Unlock()
-	resp = &proto.LockDirResponse{}

-	ino := req.Inode
-	lockId := req.LockId
-	submitTime := req.SubmitTime
-	lease := req.Lease
+	resp = &proto.LockDirResponse{
+		Status: proto.OpOk,
+		LockId: req.LockId,
+	}

-	log.LogDebugf("fsmLockDir ino=%v, lockId=%d, submitTime=%v, lease=%d\n", ino, lockId, submitTime, lease)
+	newExpire := req.SubmitTime.Unix() + int64(req.Lease)
+	newVal := fmt.Sprintf("%d|%d", req.LockId, newExpire)

-	newExtend := NewExtend(ino)
+	log.LogDebugf("fsmLockDir: req info %s, val %s", req.String(), newVal)
+
+	var newExtend = NewExtend(req.Inode)
 	treeItem := mp.extendTree.CopyGet(newExtend)

-	var firstLock bool = false
 	var oldValue []byte
 	var existExtend *Extend

 	if treeItem == nil {
-		firstLock = true
-	} else {
-		existExtend = treeItem.(*Extend)
-		oldValue, _ = existExtend.Get([]byte("dir_lock"))
-		if oldValue == nil {
-			firstLock = true
-		}
-	}
-
-	lockIdStr := strconv.Itoa(int(lockId))
-	validTime := submitTime.Add(time.Duration(int(lease)) * time.Second)
-	validTimeStr := validTime.Format("2006-01-02 15:04:05")
-	value := lockIdStr + "|" + validTimeStr
-
-	if firstLock {
-		// first time lock dir
-		log.LogDebugf("fsmLockDir first time\n")
-		uu_id, _ := uuid.NewRandom()
-		lockId = int64(uu_id.ID())
-		lockIdStr = strconv.Itoa(int(lockId))
-		value = lockIdStr + "|" + validTimeStr
-		newExtend.Put([]byte("dir_lock"), []byte(value), 0)
+		newExtend.Put([]byte(innerDirLockKey), []byte(newVal), 0)
 		mp.extendTree.ReplaceOrInsert(newExtend, true)
-	} else {
-		log.LogDebugf("fsmLockDir oldValue=%s\n", oldValue)
-		renewDirLock := false
-		lockExpired := false
-		parts := strings.Split(string(oldValue), "|")
-		oldLockIdStr := parts[0]
-		oldValidTimeStr := parts[1]
-		oldValidTime, _ := time.Parse("2006-01-02 15:04:05", oldValidTimeStr)
-		renewDirLock = (oldLockIdStr == lockIdStr)
-
-		// convert time before compare (CST/UTC)
-		submit_time, _ := time.Parse("2006-01-02 15:04:05", submitTime.Format("2006-01-02 15:04:05"))
-		lockExpired = submit_time.After(oldValidTime)
-
-		if !renewDirLock && !lockExpired {
-			resp.Status = proto.OpExistErr
-			log.LogDebugf("fsmLockDir failed, dir has been locked by others and in lease\n")
-			return
-		}
-
-		if lockExpired {
-			// if lock expired, use new lockId
-			uu_id, _ := uuid.NewRandom()
-			lockId = int64(uu_id.ID())
-			lockIdStr = strconv.Itoa(int(lockId))
-			value = lockIdStr + "|" + validTimeStr
-		}
-
-		existExtend.Remove([]byte("dir_lock"))
-		newExtend.Put([]byte("dir_lock"), []byte(value), 0)
-		existExtend.Merge(newExtend, true)
+		return
 	}

-	resp.Status = proto.OpOk
-	resp.LockId = lockId
-	log.LogDebugf("fsmLockDir success lockId=%d\n", lockId)
+	existExtend = treeItem.(*Extend)
+	oldValue, _ = existExtend.Get([]byte(innerDirLockKey))
+	if oldValue == nil {
+		newExtend.Put([]byte(innerDirLockKey), []byte(newVal), 0)
+		existExtend.Merge(newExtend, true)
+		return
+	}
+
+	var oldLkId, oldExpire int64
+	_, err := fmt.Sscanf(string(oldValue), "%d|%d", &oldLkId, &oldExpire)
+	if err != nil {
+		log.LogErrorf("fsmLockDir: parse req failed, req %s, old %s, err %s", req.String(), string(oldValue), err.Error())
+		resp.Status = proto.OpExistErr
+		return 
+	}
+
+	log.LogDebugf("fsmLockDir: get old lock dir info, req %v, old %d, expire %d", req, oldLkId, oldExpire)
+
+	if req.LockId == oldLkId {
+		newExtend.Put([]byte(innerDirLockKey), []byte(newVal), 0)
+		existExtend.Merge(newExtend, true)
+		return
+	}
+
+	if req.SubmitTime.Unix() < oldExpire {
+		resp.Status = proto.OpExistErr
+		return
+	}
+
+	newExtend.Put([]byte(innerDirLockKey), []byte(newVal), 0)
+	existExtend.Merge(newExtend, true)
+	return
+}
+
+func (mp *metaPartition) fsmUnlockDir(req *proto.LockDirRequest) (resp *proto.LockDirResponse) {
+	mp.xattrLock.Lock()
+	defer mp.xattrLock.Unlock()
+
+	resp = &proto.LockDirResponse{
+		Status: proto.OpOk,
+		LockId: req.LockId,
+	}
+
+	newExpire := req.SubmitTime.Unix() + int64(req.Lease)
+	newVal := fmt.Sprintf("%d|%d", req.LockId, newExpire)
+	log.LogDebugf("fsmUnlockDir: req info %s, val %s", req, newVal)
+
+	var newExtend = NewExtend(req.Inode)
+	treeItem := mp.extendTree.CopyGet(newExtend)
+
+	var oldValue []byte
+	var existExtend *Extend
+
+	if treeItem == nil {
+		log.LogWarnf("fsmUnlockDir: lock not exist, no need to unlock, req %s", req.String())
+		return
+	}
+
+	existExtend = treeItem.(*Extend)
+	oldValue, _ = existExtend.Get([]byte(innerDirLockKey))
+	if oldValue == nil {
+		log.LogWarnf("fsmUnlockDir: target lock val not exist, no need to unlock, req %s", req.String())
+		return
+	}
+
+	var oldLkId, oldExpire int64
+	_, err := fmt.Sscanf(string(oldValue), "%d|%d", &oldLkId, &oldExpire)
+	if err != nil {
+		log.LogErrorf("fsmUnlockDir: parse req failed, req %s, old %s, err %s", req.String(), string(oldValue), err.Error())
+		resp.Status = proto.OpExistErr
+		return
+	}
+
+	log.LogDebugf("fsmUnlockDir: get old lock dir info, req %v, old %d, expire %d", req, oldLkId, oldExpire)
+
+	if req.LockId != oldLkId {
+		log.LogWarnf("fsmUnlockDir: already been locked by other, req %v, old %d", req.String(), oldLkId)
+		resp.Status = proto.OpExistErr
+		return
+	}
+
+	existExtend.Remove([]byte(innerDirLockKey))
 	return
 }

--- a/metanode/partition_op_extend.go
+++ b/metanode/partition_op_extend.go
@ -21,6 +21,7 @@ import (
 	"time"

 	"github.com/cubefs/cubefs/proto"
+	"github.com/cubefs/cubefs/util/log"
 )

 func (mp *metaPartition) UpdateXAttr(req *proto.UpdateXAttrRequest, p *Packet) (err error) {
@ -250,6 +251,11 @@ func (mp *metaPartition) putExtend(op uint32, extend *Extend) (resp interface{},

 func (mp *metaPartition) LockDir(req *proto.LockDirRequest, p *Packet) (err error) {
 	req.SubmitTime = time.Now()
+	if req.LockId == 0 && req.Lease != 0 {
+		req.LockId = proto.GenerateRequestID()
+		log.LogWarnf("LockDir: req %s has empyt lkId from old verion.", req.String())
+	}
+
 	val, err := json.Marshal(req)
 	if err != nil {
 		p.PacketErrorWithBody(proto.OpErr, []byte(err.Error()))
--- a/proto/fs_proto.go
+++ b/proto/fs_proto.go
@ -15,6 +15,7 @@
 package proto

 import (
+	"encoding/json"
 	"fmt"
 	"os"
 	"path/filepath"
@ -1062,6 +1063,11 @@ type LockDirRequest struct {
 	SubmitTime  time.Time `json:"submitTime"`
 }

+func (lr *LockDirRequest) String() string {
+	data, _ := json.Marshal(lr)
+	return string(data)
+}
+
 type LockDirResponse struct {
 	LockId int64 `json:"lockId"`
 	Status uint8 `json:"status"`
--- a/sdk/meta/operation.go
+++ b/sdk/meta/operation.go
@ -2904,6 +2904,17 @@ func (mw *MetaWrapper) lockDir(mp *MetaPartition, inode uint64, lease uint64, lo
 		Lease:       lease,
 	}

+	if lockId == 0 && lease != 0 {
+		status, uniqId, err1 := mw.consumeUniqID(mp)
+		if err1 != nil || status != statusOK {
+			log.LogErrorf("lockDir: get uniq lockId failed, ino %d, status %d, err %v", inode, status, err1)
+			err = statusErrToErrno(status, err1)
+			return
+		}
+		lockId = int64(uniqId)
+		log.LogDebugf("lockDir: get lockId success, id %d, ino %d", lockId, inode)
+	}
+
 	packet := proto.NewPacketReqID()
 	packet.Opcode = proto.OpMetaLockDir
 	packet.PartitionID = mp.PartitionID
@ -2928,19 +2939,12 @@ func (mw *MetaWrapper) lockDir(mp *MetaPartition, inode uint64, lease uint64, lo
 	status := parseStatus(packet.ResultCode)
 	if status != statusOK {
 		err = statusToErrno(status)
-		log.LogErrorf("lockDir: received fail status, packet(%v) mp(%v) req(%v) result(%v) err(%v)", packet, mp, *req, packet.GetResultMsg(), err)
+		log.LogWarnf("lockDir: received fail status, packet(%v) mp(%v) req(%v) result(%v) err(%v)", packet, mp, *req, packet.GetResultMsg(), err)
 		return
 	}

-	resp := new(proto.LockDirResponse)
-	err = packet.UnmarshalData(resp)
-	if err != nil {
-		log.LogErrorf("lockDir: packet(%v) mp(%v) err(%v) PacketData(%v)", packet, mp, err, string(packet.Data))
-		return
-	}
-	retLockId = resp.LockId
-
-	log.LogDebugf("lockDir: packet(%v) mp(%v) req(%v) result(%v)", packet, mp, *req, packet.GetResultMsg())
+	retLockId = lockId
+	log.LogDebugf("lockDir: packet(%v) mp(%v) req(%v) lockId(%v)", packet, mp, *req, lockId)
 	return
 }