8000 [API change] add CheckObjectLock (advanced usage) · NVIDIA/aistore@b3bc65d · GitHub
[go: up one dir, main page]
More Web Proxy on the site http://driver.im/
Skip to content

Commit b3bc65d

Browse files
alex-aizmanalex-aizman
committed
[API change] add CheckObjectLock (advanced usage)
* 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 <alex.aizman@gmail.com>
1 parent e25c8b9 commit b3bc65d

File tree

10 files changed

+113
-23
lines changed

10 files changed

+113
-23
lines changed

‎ais/proxy.go

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1663,7 +1663,7 @@ func (p *proxy) httpobjpost(w http.ResponseWriter, r *http.Request, apireq *apiR
16631663
if err != nil {
16641664
return
16651665
}
1666-
if msg.Action == apc.ActRenameObject {
1666+
if msg.Action == apc.ActRenameObject || msg.Action == apc.ActCheckLock {
16671667
apireq.after = 2
16681668
}
16691669
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
17371737
}
17381738
objName := msg.Name
17391739
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)
17401745
default:
17411746
p.writeErrAct(w, r, msg.Action)
17421747
}

‎ais/target.go

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1082,6 +1082,9 @@ func (t *target) httpobjpost(w http.ResponseWriter, r *http.Request, apireq *api
10821082

10831083
// lom is eventually freed by x-blob
10841084
}
1085+
case apc.ActCheckLock:
1086+
t._checkLocked(w, r, apireq.bck, apireq.items[1])
1087+
return
10851088
default:
10861089
t.writeErrAct(w, r, msg.Action)
10871090
return
@@ -1092,6 +1095,34 @@ func (t *target) httpobjpost(w http.ResponseWriter, r *http.Request, apireq *api
10921095
}
10931096
}
10941097

1098+
// return object's lock status (enum { apc.LockNone, ... }) via HTTP status:
1099+
// - 200 OK: unlocked
1100+
// - 202 Accepted: read lock (NOTE: convention)
1101+
// - 423 Locked: write lock
1102+
func (t *target) _checkLocked(w http.ResponseWriter, r *http.Request, bck *meta.Bck, objName string) {
1103+
var (
1104+
ecode int
1105+
locked int
1106+
lom = core.AllocLOM(objName)
1107+
)
1108+
defer core.FreeLOM(lom)
1109+
if err := lom.InitBck(bck.Bucket()); err != nil {
1110+
t.writeErr(w, r, err)
1111+
return
1112+
}
1113+
locked = lom.IsLocked()
1114+
switch locked {
1115+
case apc.LockWrite:
1116+
ecode = http.StatusLocked
1117+
case apc.LockRead:
1118+
ecode = http.StatusAccepted
1119+
default:
1120+
debug.Assert(locked == apc.LockNone)
1121+
ecode = http.StatusOK
1122+
}
1123+
w.WriteHeader(ecode)
1124+
}
1125+
10951126
// HEAD /v1/objects/<bucket-name>/<object-name>
10961127
func (t *target) httpobjhead(w http.ResponseWriter, r *http.Request, apireq *apiRequest) {
10971128
if err := t.parseReq(w, r, apireq); err != nil {

‎api/apc/actmsg.go

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -102,6 +102,9 @@ const (
102102
ActListenToNotif = "watch-xaction"
103103
ActMergeOwnershipTbl = "ic-merge-own-tbl"
104104
ActRegGlobalXaction = "reg-global-xaction"
105+
106+
// advanced usage
107+
ActCheckLock = "check-lock"
105108
)
106109

107110
// internal use

‎api/apc/const.go

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,3 +32,10 @@ const (
3232
DefaultTimeout = time.Duration(-1)
3333
LongTimeout = time.Duration(-2)
3434
)
35+
36+
// locks
37+
const (
38+
LockNone = iota
39+
LockRead
40+
LockWrite
41+
)

‎api/object.go

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -636,6 +636,50 @@ func Promote(bp BaseParams, bck cmn.Bck, args *apc.PromoteArgs) (xid string, err
636636
return xid, err
637637
}
638638

639+
// Check if an object is currently locked by ongoing operations.
640+
// Handles HTTP status from AIStore:
641+
// - 200 OK: object unlocked
642+
// - 202 Accepted: read lock (NOTE: internal convention)
643+
// - 423 Locked: write lock
644+
// Returns {apc.LockNone, ...} enum or an error
645+
func CheckObjectLock(bp BaseParams, bck cmn.Bck, objName string) (int, error) {
646+
var (
647+
q = qalloc()
648+
actMsg = apc.ActMsg{Action: apc.ActCheckLock}
649+
)
650+
651+
bp.Method = http.MethodPost
652+
reqParams := AllocRp()
653+
{
654+
reqParams.BaseParams = bp
655+
reqParams.Path = apc.URLPathObjects.Join(bck.Name, objName)
656+
reqParams.Body = cos.MustMarshal(actMsg)
657+
reqParams.Header = http.Header{cos.HdrContentType: []string{cos.ContentJSON}}
658+
bck.SetQuery(q)
659+
reqParams.Query = q
660+
}
661+
662+
resp, err := reqParams.do()
663+
FreeRp(reqParams)
664+
qfree(q)
665+
if err != nil {
666+
return 0, err
667+
}
668+
resp.Body.Close()
669+
670+
switch status := resp.StatusCode; status {
671+
case http.StatusAccepted: // NOTE convention
672+
return apc.LockRead, nil
673+
case http.StatusLocked:
674+
return apc.LockWrite, nil
675+
case http.StatusOK:
676+
return apc.LockNone, nil
677+
default:
678+
err := &cmn.ErrHTTP{Message: http.StatusText(status), Status: status}
679+
return 0, err
680+
}
681+
}
682+
639683
//
640684
// misc. helpers
641685
//

‎core/lcopy.go

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,6 @@ func (lom *LOM) NumCopies() int { return max(len(lom.md.copies), 1) } // metada
3535
// - copies include lom.FQN aka "main repl."
3636
// - caller must take a lock
3737
func (lom *LOM) GetCopies() fs.MPI {
38-
debug.Assert(lom.isLockedRW(), lom.Cname())
3938
return lom.md.copies
4039
}
4140

@@ -143,9 +142,6 @@ func (lom *LOM) syncMetaWithCopies() (err error) {
143142
if !lom.HasCopies() {
144143
return nil
145144
}
146-
// caller is responsible for write-locking
147-
debug.Assert(lom.isLockedExcl(), lom.Cname())
148-
149145
if !lom.WritePolicy().IsImmediate() {
150146
lom.md.makeDirty()
151147
return nil

‎core/ldp.go

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import (
99
"net/http"
1010
"time"
1111

12+
"github.com/NVIDIA/aistore/api/apc"
1213
"github.com/NVIDIA/aistore/cmn"
1314
"github.com/NVIDIA/aistore/cmn/cos"
1415
"github.com/NVIDIA/aistore/cmn/debug"
@@ -202,7 +203,7 @@ func (lom *LOM) CheckRemoteMD(locked, sync bool, origReq *http.Request) (res CRM
202203

203204
// NOTE: Sync is false (ie., not deleting)
204205
func (lom *LOM) LoadLatest(latest bool) (oa *cmn.ObjAttrs, deleted bool, err error) {
205-
debug.Assert(lom.isLockedRW(), lom.Cname()) // caller must take a lock
206+
debug.Assert(lom.IsLocked() > apc.LockNone, "must be locked: ", lom.Cname())
206207

207208
err = lom.Load(true /*cache it*/, true /*locked*/)
208209
if err != nil {

‎core/lfile.go

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ import (
1010
"path/filepath"
1111
"syscall"
1212

13+
"github.com/NVIDIA/aistore/api/apc"
1314
"github.com/NVIDIA/aistore/cmn"
1415
"github.com/NVIDIA/aistore/cmn/cos"
1516
"github.com/NVIDIA/aistore/cmn/debug"
@@ -71,7 +72,7 @@ func (lom *LOM) Open() (fh cos.LomReader, err error) {
7172
//
7273

7374
func (lom *LOM) Create() (cos.LomWriter, error) {
74-
debug.Assert(lom.isLockedExcl(), lom.Cname()) // caller must wlock
75+
debug.Assert(lom.IsLocked() == apc.LockWrite, "must be wlocked: ", lom.Cname())
7576
return lom._cf(lom.FQN)
7677
}
7778

@@ -156,11 +157,12 @@ func (lom *LOM) RemoveObj(force ...bool) (err error) {
156157
lom.UncacheDel()
157158

158159
debug.AssertFunc(func() bool {
159-
if lom.isLockedExcl() {
160+
locked := lom.IsLocked()
161+
if locked == apc.LockWrite {
160162
return true
161163
}
162164
// NOTE: making "rlock" exception to be able to forcefully rm corrupted object in the GET path
163-
return len(force) > 0 && force[0] && lom.isLockedRW()
165+
return len(force) > 0 && force[0] && locked == apc.LockRead
164166
})
165167
err = lom.RemoveMain()
166168
for copyFQN := range lom.md.copies {

‎core/lom.go

Lines changed: 3 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -633,16 +633,10 @@ func (lom *LOM) whingeSize(size int64) error {
633633

634634
func (lom *LOM) getLocker() *nlc { return &g.locker[lom.CacheIdx()] } // (lif.getLocker())
635635

636-
func (lom *LOM) isLockedExcl() (exclusive bool) {
636+
// returns {apc.LockNone, ...} enum
637+
func (lom *LOM) IsLocked() int {
637638
nlc := lom.getLocker()
638-
_, exclusive = nlc.IsLocked(lom.Uname())
639-
return exclusive
640-
}
641-
642-
func (lom *LOM) isLockedRW() bool {
643-
nlc := lom.getLocker()
644-
rc, exclusive := nlc.IsLocked(lom.Uname())
645-
return exclusive || rc > 0
639+
return nlc.IsLocked(lom.Uname())
646640
}
647641

648642
func (lom *LOM) TryLock(exclusive bool) bool {

‎core/namelocker.go

Lines changed: 12 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ import (
88
"sync"
99
"time"
1010

11+
"github.com/NVIDIA/aistore/api/apc"
1112
"github.com/NVIDIA/aistore/cmn/cos"
1213
"github.com/NVIDIA/aistore/cmn/debug"
1314

@@ -78,15 +79,21 @@ func (nlc *nlc) init() {
7879
nlc.m = make(map[string]*lockInfo, initCapacity)
7980
}
8081

81-
func (nlc *nlc) IsLocked(uname string) (rc int, exclusive bool) {
82+
// returns {apc.LockNone, ...} enum
83+
func (nlc *nlc) IsLocked(uname string) int {
8284
nlc.mu.Lock()
8385
defer nlc.mu.Unlock()
8486

8587
lockInfo, found := nlc.m[uname]
86-
if !found {
87-
return
88-
}
89-
return int(lockInfo.rc), lockInfo.exclusive
88+
switch {
89+
case !found:
90+
return apc.LockNone
91+
case lockInfo.exclusive:
92+
return apc.LockWrite
93+
case lockInfo.rc > 0:
94+
return apc.LockRead
95+
}
96+
return apc.LockNone
9097
}
9198

9299
func (nlc *nlc) TryLock(uname string, exclusive bool) (locked bool) {

0 commit comments

Comments
 (0)
0