diff options
| -rw-r--r-- | weed/util/lock_table.go | 44 |
1 files changed, 27 insertions, 17 deletions
diff --git a/weed/util/lock_table.go b/weed/util/lock_table.go index d10b51334..8ec93f7e7 100644 --- a/weed/util/lock_table.go +++ b/weed/util/lock_table.go @@ -16,11 +16,12 @@ type LockTable[T comparable] struct { } type LockEntry struct { - mu sync.Mutex - waiters []*ActiveLock // ordered waiters that are blocked by exclusive locks - activeLockOwnerCount int32 - lockType LockType - cond *sync.Cond + mu sync.Mutex + waiters []*ActiveLock // ordered waiters that are blocked by exclusive locks + activeSharedLockOwnerCount int32 + activeExclusiveLockOwnerCount int32 + lockType LockType + cond *sync.Cond } type LockType int @@ -34,6 +35,7 @@ type ActiveLock struct { ID int64 isDeleted bool intention string // for debugging + lockType LockType } func NewLockTable[T comparable]() *LockTable[T] { @@ -42,9 +44,9 @@ func NewLockTable[T comparable]() *LockTable[T] { } } -func (lt *LockTable[T]) NewActiveLock(intention string) *ActiveLock { +func (lt *LockTable[T]) NewActiveLock(intention string, lockType LockType) *ActiveLock { id := atomic.AddInt64(<.lockIdSeq, 1) - l := &ActiveLock{ID: id, intention: intention} + l := &ActiveLock{ID: id, intention: intention, lockType: lockType} return l } @@ -59,13 +61,13 @@ func (lt *LockTable[T]) AcquireLock(intention string, key T, lockType LockType) } lt.mu.Unlock() - lock = lt.NewActiveLock(intention) + lock = lt.NewActiveLock(intention, lockType) // If the lock is held exclusively, wait entry.mu.Lock() - if len(entry.waiters) > 0 || lockType == ExclusiveLock { + if len(entry.waiters) > 0 || lockType == ExclusiveLock || entry.activeExclusiveLockOwnerCount > 0 { if glog.V(4) { - fmt.Printf("ActiveLock %d %s wait for %+v type=%v with waiters %d active %d.\n", lock.ID, lock.intention, key, lockType, len(entry.waiters), entry.activeLockOwnerCount) + fmt.Printf("ActiveLock %d %s wait for %+v type=%v with waiters %d active r%d w%d.\n", lock.ID, lock.intention, key, lockType, len(entry.waiters), entry.activeSharedLockOwnerCount, entry.activeExclusiveLockOwnerCount) if len(entry.waiters) > 0 { for _, waiter := range entry.waiters { fmt.Printf(" %d", waiter.ID) @@ -75,11 +77,11 @@ func (lt *LockTable[T]) AcquireLock(intention string, key T, lockType LockType) } entry.waiters = append(entry.waiters, lock) if lockType == ExclusiveLock { - for !lock.isDeleted && ((len(entry.waiters) > 0 && lock.ID != entry.waiters[0].ID) || entry.activeLockOwnerCount > 0) { + for !lock.isDeleted && ((len(entry.waiters) > 0 && lock.ID != entry.waiters[0].ID) || entry.activeExclusiveLockOwnerCount > 0 || entry.activeSharedLockOwnerCount > 0) { entry.cond.Wait() } } else { - for !lock.isDeleted && (len(entry.waiters) > 0 && lock.ID != entry.waiters[0].ID) { + for !lock.isDeleted && (len(entry.waiters) > 0 && lock.ID != entry.waiters[0].ID) || entry.activeExclusiveLockOwnerCount > 0 { entry.cond.Wait() } } @@ -89,12 +91,11 @@ func (lt *LockTable[T]) AcquireLock(intention string, key T, lockType LockType) entry.cond.Broadcast() } } - entry.activeLockOwnerCount++ // Otherwise, grant the lock entry.lockType = lockType if glog.V(4) { - fmt.Printf("ActiveLock %d %s locked %+v type=%v with waiters %d active %d.\n", lock.ID, lock.intention, key, lockType, len(entry.waiters), entry.activeLockOwnerCount) + fmt.Printf("ActiveLock %d %s locked %+v type=%v with waiters %d active r%d w%d.\n", lock.ID, lock.intention, key, lockType, len(entry.waiters), entry.activeSharedLockOwnerCount, entry.activeExclusiveLockOwnerCount) if len(entry.waiters) > 0 { for _, waiter := range entry.waiters { fmt.Printf(" %d", waiter.ID) @@ -102,6 +103,11 @@ func (lt *LockTable[T]) AcquireLock(intention string, key T, lockType LockType) fmt.Printf("\n") } } + if lock.lockType == ExclusiveLock { + entry.activeExclusiveLockOwnerCount++ + } else { + entry.activeSharedLockOwnerCount++ + } entry.mu.Unlock() return lock @@ -129,12 +135,12 @@ func (lt *LockTable[T]) ReleaseLock(key T, lock *ActiveLock) { } // If there are no waiters, release the lock - if len(entry.waiters) == 0 { + if len(entry.waiters) == 0 && entry.activeExclusiveLockOwnerCount <= 0 && entry.activeSharedLockOwnerCount <= 0 { delete(lt.locks, key) } if glog.V(4) { - fmt.Printf("ActiveLock %d %s unlocked %+v type=%v with waiters %d active %d.\n", lock.ID, lock.intention, key, entry.lockType, len(entry.waiters), entry.activeLockOwnerCount) + fmt.Printf("ActiveLock %d %s unlocked %+v type=%v with waiters %d active r%d w%d.\n", lock.ID, lock.intention, key, entry.lockType, len(entry.waiters), entry.activeSharedLockOwnerCount, entry.activeExclusiveLockOwnerCount) if len(entry.waiters) > 0 { for _, waiter := range entry.waiters { fmt.Printf(" %d", waiter.ID) @@ -142,7 +148,11 @@ func (lt *LockTable[T]) ReleaseLock(key T, lock *ActiveLock) { fmt.Printf("\n") } } - entry.activeLockOwnerCount-- + if lock.lockType == ExclusiveLock { + entry.activeExclusiveLockOwnerCount-- + } else { + entry.activeSharedLockOwnerCount-- + } // Notify the next waiter entry.cond.Broadcast() |