|
|
|
@ -51,14 +51,6 @@ type RWLocker interface { |
|
|
|
|
RUnlock() |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// RWLockerSync - internal locker interface.
|
|
|
|
|
type RWLockerSync interface { |
|
|
|
|
GetLock(id, source string, timeout time.Duration) bool |
|
|
|
|
Unlock() |
|
|
|
|
GetRLock(id, source string, timeout time.Duration) bool |
|
|
|
|
RUnlock() |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// Initialize distributed locking only in case of distributed setup.
|
|
|
|
|
// Returns lock clients and the node index for the current server.
|
|
|
|
|
func newDsyncNodes(endpoints EndpointList) (clnts []dsync.NetLocker, myNode int) { |
|
|
|
@ -101,8 +93,11 @@ func newDsyncNodes(endpoints EndpointList) (clnts []dsync.NetLocker, myNode int) |
|
|
|
|
func newNSLock(isDistXL bool) *nsLockMap { |
|
|
|
|
nsMutex := nsLockMap{ |
|
|
|
|
isDistXL: isDistXL, |
|
|
|
|
lockMap: make(map[nsParam]*nsLock), |
|
|
|
|
} |
|
|
|
|
if isDistXL { |
|
|
|
|
return &nsMutex |
|
|
|
|
} |
|
|
|
|
nsMutex.lockMap = make(map[nsParam]*nsLock) |
|
|
|
|
return &nsMutex |
|
|
|
|
} |
|
|
|
|
|
|
|
|
@ -119,7 +114,7 @@ type nsParam struct { |
|
|
|
|
|
|
|
|
|
// nsLock - provides primitives for locking critical namespace regions.
|
|
|
|
|
type nsLock struct { |
|
|
|
|
RWLockerSync |
|
|
|
|
*lsync.LRWMutex |
|
|
|
|
ref uint |
|
|
|
|
} |
|
|
|
|
|
|
|
|
@ -141,13 +136,8 @@ func (n *nsLockMap) lock(volume, path string, lockSource, opsID string, readLock |
|
|
|
|
nsLk, found := n.lockMap[param] |
|
|
|
|
if !found { |
|
|
|
|
n.lockMap[param] = &nsLock{ |
|
|
|
|
RWLockerSync: func() RWLockerSync { |
|
|
|
|
if n.isDistXL { |
|
|
|
|
return dsync.NewDRWMutex(pathJoin(volume, path), globalDsync) |
|
|
|
|
} |
|
|
|
|
return &lsync.LRWMutex{} |
|
|
|
|
}(), |
|
|
|
|
ref: 1, |
|
|
|
|
LRWMutex: &lsync.LRWMutex{}, |
|
|
|
|
ref: 1, |
|
|
|
|
} |
|
|
|
|
nsLk = n.lockMap[param] |
|
|
|
|
} else { |
|
|
|
@ -259,8 +249,49 @@ func (n *nsLockMap) ForceUnlock(volume, path string) { |
|
|
|
|
delete(n.lockMap, nsParam{volume, path}) |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// lockInstance - frontend/top-level interface for namespace locks.
|
|
|
|
|
type lockInstance struct { |
|
|
|
|
// dsync's distributed lock instance.
|
|
|
|
|
type distLockInstance struct { |
|
|
|
|
rwMutex *dsync.DRWMutex |
|
|
|
|
volume, path, opsID string |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// Lock - block until write lock is taken or timeout has occurred.
|
|
|
|
|
func (di *distLockInstance) GetLock(timeout *dynamicTimeout) (timedOutErr error) { |
|
|
|
|
lockSource := getSource() |
|
|
|
|
start := UTCNow() |
|
|
|
|
|
|
|
|
|
if !di.rwMutex.GetLock(di.opsID, lockSource, timeout.Timeout()) { |
|
|
|
|
timeout.LogFailure() |
|
|
|
|
return OperationTimedOut{Path: di.path} |
|
|
|
|
} |
|
|
|
|
timeout.LogSuccess(UTCNow().Sub(start)) |
|
|
|
|
return nil |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// Unlock - block until write lock is released.
|
|
|
|
|
func (di *distLockInstance) Unlock() { |
|
|
|
|
di.rwMutex.Unlock() |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// RLock - block until read lock is taken or timeout has occurred.
|
|
|
|
|
func (di *distLockInstance) GetRLock(timeout *dynamicTimeout) (timedOutErr error) { |
|
|
|
|
lockSource := getSource() |
|
|
|
|
start := UTCNow() |
|
|
|
|
if !di.rwMutex.GetRLock(di.opsID, lockSource, timeout.Timeout()) { |
|
|
|
|
timeout.LogFailure() |
|
|
|
|
return OperationTimedOut{Path: di.path} |
|
|
|
|
} |
|
|
|
|
timeout.LogSuccess(UTCNow().Sub(start)) |
|
|
|
|
return nil |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// RUnlock - block until read lock is released.
|
|
|
|
|
func (di *distLockInstance) RUnlock() { |
|
|
|
|
di.rwMutex.RUnlock() |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// localLockInstance - frontend/top-level interface for namespace locks.
|
|
|
|
|
type localLockInstance struct { |
|
|
|
|
ns *nsLockMap |
|
|
|
|
volume, path, opsID string |
|
|
|
|
} |
|
|
|
@ -270,11 +301,14 @@ type lockInstance struct { |
|
|
|
|
// volume, path and operation ID.
|
|
|
|
|
func (n *nsLockMap) NewNSLock(volume, path string) RWLocker { |
|
|
|
|
opsID := mustGetUUID() |
|
|
|
|
return &lockInstance{n, volume, path, opsID} |
|
|
|
|
if n.isDistXL { |
|
|
|
|
return &distLockInstance{dsync.NewDRWMutex(pathJoin(volume, path), globalDsync), volume, path, opsID} |
|
|
|
|
} |
|
|
|
|
return &localLockInstance{n, volume, path, opsID} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// Lock - block until write lock is taken or timeout has occurred.
|
|
|
|
|
func (li *lockInstance) GetLock(timeout *dynamicTimeout) (timedOutErr error) { |
|
|
|
|
func (li *localLockInstance) GetLock(timeout *dynamicTimeout) (timedOutErr error) { |
|
|
|
|
lockSource := getSource() |
|
|
|
|
start := UTCNow() |
|
|
|
|
readLock := false |
|
|
|
@ -287,13 +321,13 @@ func (li *lockInstance) GetLock(timeout *dynamicTimeout) (timedOutErr error) { |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// Unlock - block until write lock is released.
|
|
|
|
|
func (li *lockInstance) Unlock() { |
|
|
|
|
func (li *localLockInstance) Unlock() { |
|
|
|
|
readLock := false |
|
|
|
|
li.ns.unlock(li.volume, li.path, li.opsID, readLock) |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// RLock - block until read lock is taken or timeout has occurred.
|
|
|
|
|
func (li *lockInstance) GetRLock(timeout *dynamicTimeout) (timedOutErr error) { |
|
|
|
|
func (li *localLockInstance) GetRLock(timeout *dynamicTimeout) (timedOutErr error) { |
|
|
|
|
lockSource := getSource() |
|
|
|
|
start := UTCNow() |
|
|
|
|
readLock := true |
|
|
|
@ -306,7 +340,7 @@ func (li *lockInstance) GetRLock(timeout *dynamicTimeout) (timedOutErr error) { |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// RUnlock - block until read lock is released.
|
|
|
|
|
func (li *lockInstance) RUnlock() { |
|
|
|
|
func (li *localLockInstance) RUnlock() { |
|
|
|
|
readLock := true |
|
|
|
|
li.ns.unlock(li.volume, li.path, li.opsID, readLock) |
|
|
|
|
} |
|
|
|
|