[API change] add CheckObjectLock (advanced usage)

alex-aizman · alex-aizman · commit b3bc65da33a8 · 2025-05-22T11:46:58.000-04:00
* generally, this is a head-of-line (HOL) blocking or convoy effect
* specifically, we see:
  - multiple clients timing out waiting for slow cold GET(s);
  - they keep getting (one of the two possible flavors of) read timeout
  - and keep retrying
* client can now check lock status and maybe implement more intelligent
  retrying logic

Signed-off-by: Alex Aizman &lt;alex.aizman@gmail.com&gt;
diff --git a/ais/proxy.go b/ais/proxy.go
@@ -1663,7 +1663,7 @@ func (p *proxy) httpobjpost(w http.ResponseWriter, r *http.Request, apireq *apiR
 	if err != nil {
 		return
 	}
-	if msg.Action == apc.ActRenameObject {
+	if msg.Action == apc.ActRenameObject || msg.Action == apc.ActCheckLock {
 		apireq.after = 2
 	}
 	if err := p.parseReq(w, r, apireq); err != nil {
@@ -1737,6 +1737,11 @@ func (p *proxy) httpobjpost(w http.ResponseWriter, r *http.Request, apireq *apiR
 		}
 		objName := msg.Name
 		p.redirectAction(w, r, bck, objName, msg)
+	case apc.ActCheckLock:
+		if err := p.checkAccess(w, r, bck, apc.AccessRO); err != nil {
+			return
+		}
+		p.redirectAction(w, r, bck, apireq.items[1], msg)
 	default:
 		p.writeErrAct(w, r, msg.Action)
 	}
diff --git a/ais/target.go b/ais/target.go
@@ -1082,6 +1082,9 @@ func (t *target) httpobjpost(w http.ResponseWriter, r *http.Request, apireq *api
 
 			// lom is eventually freed by x-blob
 		}
+	case apc.ActCheckLock:
+		t._checkLocked(w, r, apireq.bck, apireq.items[1])
+		return
 	default:
 		t.writeErrAct(w, r, msg.Action)
 		return
@@ -1092,6 +1095,34 @@ func (t *target) httpobjpost(w http.ResponseWriter, r *http.Request, apireq *api
 	}
 }
 
+// return object's lock status (enum { apc.LockNone, ... }) via HTTP status:
+// - 200 OK:       unlocked
+// - 202 Accepted: read lock (NOTE: convention)
+// - 423 Locked:   write lock
+func (t *target) _checkLocked(w http.ResponseWriter, r *http.Request, bck *meta.Bck, objName string) {
+	var (
+		ecode  int
+		locked int
+		lom    = core.AllocLOM(objName)
+	)
+	defer core.FreeLOM(lom)
+	if err := lom.InitBck(bck.Bucket()); err != nil {
+		t.writeErr(w, r, err)
+		return
+	}
+	locked = lom.IsLocked()
+	switch locked {
+	case apc.LockWrite:
+		ecode = http.StatusLocked
+	case apc.LockRead:
+		ecode = http.StatusAccepted
+	default:
+		debug.Assert(locked == apc.LockNone)
+		ecode = http.StatusOK
+	}
+	w.WriteHeader(ecode)
+}
+
 // HEAD /v1/objects/<bucket-name>/<object-name>
 func (t *target) httpobjhead(w http.ResponseWriter, r *http.Request, apireq *apiRequest) {
 	if err := t.parseReq(w, r, apireq); err != nil {
diff --git a/api/apc/actmsg.go b/api/apc/actmsg.go
@@ -102,6 +102,9 @@ const (
 	ActListenToNotif     = "watch-xaction"
 	ActMergeOwnershipTbl = "ic-merge-own-tbl"
 	ActRegGlobalXaction  = "reg-global-xaction"
+
+	// advanced usage
+	ActCheckLock = "check-lock"
 )
 
 // internal use
diff --git a/api/apc/const.go b/api/apc/const.go
@@ -32,3 +32,10 @@ const (
 	DefaultTimeout = time.Duration(-1)
 	LongTimeout    = time.Duration(-2)
 )
+
+// locks
+const (
+	LockNone = iota
+	LockRead
+	LockWrite
+)
diff --git a/api/object.go b/api/object.go
@@ -636,6 +636,50 @@ func Promote(bp BaseParams, bck cmn.Bck, args *apc.PromoteArgs) (xid string, err
 	return xid, err
 }
 
+// Check if an object is currently locked by ongoing operations.
+// Handles HTTP status from AIStore:
+// - 200 OK:       object unlocked
+// - 202 Accepted: read lock (NOTE: internal convention)
+// - 423 Locked:   write lock
+// Returns {apc.LockNone, ...} enum or an error
+func CheckObjectLock(bp BaseParams, bck cmn.Bck, objName string) (int, error) {
+	var (
+		q      = qalloc()
+		actMsg = apc.ActMsg{Action: apc.ActCheckLock}
+	)
+
+	bp.Method = http.MethodPost
+	reqParams := AllocRp()
+	{
+		reqParams.BaseParams = bp
+		reqParams.Path = apc.URLPathObjects.Join(bck.Name, objName)
+		reqParams.Body = cos.MustMarshal(actMsg)
+		reqParams.Header = http.Header{cos.HdrContentType: []string{cos.ContentJSON}}
+		bck.SetQuery(q)
+		reqParams.Query = q
+	}
+
+	resp, err := reqParams.do()
+	FreeRp(reqParams)
+	qfree(q)
+	if err != nil {
+		return 0, err
+	}
+	resp.Body.Close()
+
+	switch status := resp.StatusCode; status {
+	case http.StatusAccepted: // NOTE convention
+		return apc.LockRead, nil
+	case http.StatusLocked:
+		return apc.LockWrite, nil
+	case http.StatusOK:
+		return apc.LockNone, nil
+	default:
+		err := &cmn.ErrHTTP{Message: http.StatusText(status), Status: status}
+		return 0, err
+	}
+}
+
 //
 // misc. helpers
 //
diff --git a/core/lcopy.go b/core/lcopy.go
@@ -35,7 +35,6 @@ func (lom *LOM) NumCopies() int  { return max(len(lom.md.copies), 1) } // metada
 // - copies include lom.FQN aka "main repl."
 // - caller must take a lock
 func (lom *LOM) GetCopies() fs.MPI {
-	debug.Assert(lom.isLockedRW(), lom.Cname())
 	return lom.md.copies
 }
 
@@ -143,9 +142,6 @@ func (lom *LOM) syncMetaWithCopies() (err error) {
 	if !lom.HasCopies() {
 		return nil
 	}
-	// caller is responsible for write-locking
-	debug.Assert(lom.isLockedExcl(), lom.Cname())
-
 	if !lom.WritePolicy().IsImmediate() {
 		lom.md.makeDirty()
 		return nil
diff --git a/core/ldp.go b/core/ldp.go
@@ -9,6 +9,7 @@ import (
 	"net/http"
 	"time"
 
+	"github.com/NVIDIA/aistore/api/apc"
 	"github.com/NVIDIA/aistore/cmn"
 	"github.com/NVIDIA/aistore/cmn/cos"
 	"github.com/NVIDIA/aistore/cmn/debug"
@@ -202,7 +203,7 @@ func (lom *LOM) CheckRemoteMD(locked, sync bool, origReq *http.Request) (res CRM
 
 // NOTE: Sync is false (ie., not deleting)
 func (lom *LOM) LoadLatest(latest bool) (oa *cmn.ObjAttrs, deleted bool, err error) {
-	debug.Assert(lom.isLockedRW(), lom.Cname()) // caller must take a lock
+	debug.Assert(lom.IsLocked() > apc.LockNone, "must be locked: ", lom.Cname())
 
 	err = lom.Load(true /*cache it*/, true /*locked*/)
 	if err != nil {
diff --git a/core/lfile.go b/core/lfile.go
@@ -10,6 +10,7 @@ import (
 	"path/filepath"
 	"syscall"
 
+	"github.com/NVIDIA/aistore/api/apc"
 	"github.com/NVIDIA/aistore/cmn"
 	"github.com/NVIDIA/aistore/cmn/cos"
 	"github.com/NVIDIA/aistore/cmn/debug"
@@ -71,7 +72,7 @@ func (lom *LOM) Open() (fh cos.LomReader, err error) {
 //
 
 func (lom *LOM) Create() (cos.LomWriter, error) {
-	debug.Assert(lom.isLockedExcl(), lom.Cname()) // caller must wlock
+	debug.Assert(lom.IsLocked() == apc.LockWrite, "must be wlocked: ", lom.Cname())
 	return lom._cf(lom.FQN)
 }
 
@@ -156,11 +157,12 @@ func (lom *LOM) RemoveObj(force ...bool) (err error) {
 	lom.UncacheDel()
 
 	debug.AssertFunc(func() bool {
-		if lom.isLockedExcl() {
+		locked := lom.IsLocked()
+		if locked == apc.LockWrite {
 			return true
 		}
 		// NOTE: making "rlock" exception to be able to forcefully rm corrupted object in the GET path
-		return len(force) > 0 && force[0] && lom.isLockedRW()
+		return len(force) > 0 && force[0] && locked == apc.LockRead
 	})
 	err = lom.RemoveMain()
 	for copyFQN := range lom.md.copies {
diff --git a/core/lom.go b/core/lom.go
@@ -633,16 +633,10 @@ func (lom *LOM) whingeSize(size int64) error {
 
 func (lom *LOM) getLocker() *nlc { return &g.locker[lom.CacheIdx()] } // (lif.getLocker())
 
-func (lom *LOM) isLockedExcl() (exclusive bool) {
+// returns {apc.LockNone, ...} enum
+func (lom *LOM) IsLocked() int {
 	nlc := lom.getLocker()
-	_, exclusive = nlc.IsLocked(lom.Uname())
-	return exclusive
-}
-
-func (lom *LOM) isLockedRW() bool {
-	nlc := lom.getLocker()
-	rc, exclusive := nlc.IsLocked(lom.Uname())
-	return exclusive || rc > 0
+	return nlc.IsLocked(lom.Uname())
 }
 
 func (lom *LOM) TryLock(exclusive bool) bool {
diff --git a/core/namelocker.go b/core/namelocker.go
@@ -8,6 +8,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/NVIDIA/aistore/api/apc"
 	"github.com/NVIDIA/aistore/cmn/cos"
 	"github.com/NVIDIA/aistore/cmn/debug"
 
@@ -78,15 +79,21 @@ func (nlc *nlc) init() {
 	nlc.m = make(map[string]*lockInfo, initCapacity)
 }
 
-func (nlc *nlc) IsLocked(uname string) (rc int, exclusive bool) {
+// returns {apc.LockNone, ...} enum
+func (nlc *nlc) IsLocked(uname string) int {
 	nlc.mu.Lock()
 	defer nlc.mu.Unlock()
 
 	lockInfo, found := nlc.m[uname]
-	if !found {
-		return
-	}
-	return int(lockInfo.rc), lockInfo.exclusive
+	switch {
+	case !found:
+		return apc.LockNone
+	case lockInfo.exclusive:
+		return apc.LockWrite
+	case lockInfo.rc > 0:
+		return apc.LockRead
+	}
+	return apc.LockNone
 }
 
 func (nlc *nlc) TryLock(uname string, exclusive bool) (locked bool) {

Original file line number	Diff line number	Diff line change
`@@ -1663,7 +1663,7 @@ func (p proxy) httpobjpost(w http.ResponseWriter, r http.Request, apireq *apiR`
`1663`	`1663`	`if err != nil {`
`1664`	`1664`	`return`
`1665`	`1665`	`}`
`1666`		`- if msg.Action == apc.ActRenameObject {`
	`1666`	`+ if msg.Action == apc.ActRenameObject \|\| msg.Action == apc.ActCheckLock {`
`1667`	`1667`	`apireq.after = 2`
`1668`	`1668`	`}`
`1669`	`1669`	`if err := p.parseReq(w, r, apireq); err != nil {`
`@@ -1737,6 +1737,11 @@ func (p proxy) httpobjpost(w http.ResponseWriter, r http.Request, apireq *apiR`
`1737`	`1737`	`}`
`1738`	`1738`	`objName := msg.Name`
`1739`	`1739`	`p.redirectAction(w, r, bck, objName, msg)`
	`1740`	`+ case apc.ActCheckLock:`
	`1741`	`+ if err := p.checkAccess(w, r, bck, apc.AccessRO); err != nil {`
	`1742`	`+ return`
	`1743`	`+ }`
	`1744`	`+ p.redirectAction(w, r, bck, apireq.items[1], msg)`
`1740`	`1745`	`default:`
`1741`	`1746`	`p.writeErrAct(w, r, msg.Action)`
`1742`	`1747`	`}`
Original file line number	Diff line number	Diff line change
`@@ -102,6 +102,9 @@ const (`
`102`	`102`	`ActListenToNotif = "watch-xaction"`
`103`	`103`	`ActMergeOwnershipTbl = "ic-merge-own-tbl"`
`104`	`104`	`ActRegGlobalXaction = "reg-global-xaction"`
	`105`	`+`
	`106`	`+ // advanced usage`
	`107`	`+ ActCheckLock = "check-lock"`
`105`	`108`	`)`
`106`	`109`
`107`	`110`	`// internal use`