test: add contentLock, a per-key mutex for content-hash exclusion
Introduces the keyed exclusion primitive that StoreSource and evictSourceBlob will hold across their full operation, so a store and an eviction racing on identical content bytes cannot interleave. Covered here in isolation: same-key exclusion, independence across distinct keys, and that the entry map does not grow unbounded.
This commit is contained in:
64
internal/imgcache/contentlock.go
Normal file
64
internal/imgcache/contentlock.go
Normal file
@@ -0,0 +1,64 @@
|
|||||||
|
package imgcache
|
||||||
|
|
||||||
|
import "sync"
|
||||||
|
|
||||||
|
// contentLock provides per-key mutual exclusion for content-hash keyed
|
||||||
|
// operations. StoreSource and evictSourceBlob each hold a content
|
||||||
|
// hash's lock for the full duration of their file-plus-accounting-row
|
||||||
|
// work, so a store and an eviction racing on identical content bytes
|
||||||
|
// (the real SHA-256 content-addressed dedup case, not a contrived one)
|
||||||
|
// can never interleave: the unlink of an evicted blob's file can never
|
||||||
|
// race the creation of a fresh database row for a concurrently
|
||||||
|
// re-stored copy of the same content. Entries are removed once no
|
||||||
|
// goroutine holds or is waiting for them, so a long-running process
|
||||||
|
// does not accumulate memory proportional to the number of distinct
|
||||||
|
// content hashes it has ever seen.
|
||||||
|
type contentLock struct {
|
||||||
|
mu sync.Mutex
|
||||||
|
entries map[string]*contentLockEntry
|
||||||
|
}
|
||||||
|
|
||||||
|
// contentLockEntry is one key's exclusion lock plus a count of
|
||||||
|
// goroutines currently holding or waiting to acquire it, used to know
|
||||||
|
// when it is safe to remove the entry from the map.
|
||||||
|
type contentLockEntry struct {
|
||||||
|
mu sync.Mutex
|
||||||
|
count int
|
||||||
|
}
|
||||||
|
|
||||||
|
// newContentLock creates an empty contentLock.
|
||||||
|
func newContentLock() *contentLock {
|
||||||
|
return &contentLock{entries: make(map[string]*contentLockEntry)}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Lock acquires exclusive access for key, blocking until it is
|
||||||
|
// available, and returns a function that releases it. The caller must
|
||||||
|
// invoke the returned function exactly once to release the lock.
|
||||||
|
func (c *contentLock) Lock(key string) func() {
|
||||||
|
c.mu.Lock()
|
||||||
|
|
||||||
|
entry, ok := c.entries[key]
|
||||||
|
if !ok {
|
||||||
|
entry = &contentLockEntry{}
|
||||||
|
c.entries[key] = entry
|
||||||
|
}
|
||||||
|
|
||||||
|
entry.count++
|
||||||
|
|
||||||
|
c.mu.Unlock()
|
||||||
|
|
||||||
|
entry.mu.Lock()
|
||||||
|
|
||||||
|
return func() {
|
||||||
|
entry.mu.Unlock()
|
||||||
|
|
||||||
|
c.mu.Lock()
|
||||||
|
|
||||||
|
entry.count--
|
||||||
|
if entry.count == 0 {
|
||||||
|
delete(c.entries, key)
|
||||||
|
}
|
||||||
|
|
||||||
|
c.mu.Unlock()
|
||||||
|
}
|
||||||
|
}
|
||||||
133
internal/imgcache/contentlock_test.go
Normal file
133
internal/imgcache/contentlock_test.go
Normal file
@@ -0,0 +1,133 @@
|
|||||||
|
package imgcache
|
||||||
|
|
||||||
|
import (
|
||||||
|
"sync"
|
||||||
|
"sync/atomic"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
// TestContentLockExcludesSameKey verifies that two goroutines locking
|
||||||
|
// the same key never run their critical sections concurrently.
|
||||||
|
func TestContentLockExcludesSameKey(t *testing.T) {
|
||||||
|
lock := newContentLock()
|
||||||
|
|
||||||
|
var (
|
||||||
|
active int32
|
||||||
|
maxSeen int32
|
||||||
|
wg sync.WaitGroup
|
||||||
|
)
|
||||||
|
|
||||||
|
const goroutines = 20
|
||||||
|
|
||||||
|
wg.Add(goroutines)
|
||||||
|
|
||||||
|
for i := 0; i < goroutines; i++ {
|
||||||
|
go func() {
|
||||||
|
defer wg.Done()
|
||||||
|
|
||||||
|
unlock := lock.Lock("same-key")
|
||||||
|
defer unlock()
|
||||||
|
|
||||||
|
n := atomic.AddInt32(&active, 1)
|
||||||
|
|
||||||
|
for {
|
||||||
|
seen := atomic.LoadInt32(&maxSeen)
|
||||||
|
if n <= seen || atomic.CompareAndSwapInt32(&maxSeen, seen, n) {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
time.Sleep(time.Millisecond)
|
||||||
|
|
||||||
|
atomic.AddInt32(&active, -1)
|
||||||
|
}()
|
||||||
|
}
|
||||||
|
|
||||||
|
wg.Wait()
|
||||||
|
|
||||||
|
if maxSeen != 1 {
|
||||||
|
t.Errorf("max concurrent holders of the same key = %d, want 1", maxSeen)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestContentLockAllowsDifferentKeys verifies that locking distinct
|
||||||
|
// keys does not serialize unrelated work: all goroutines must be able
|
||||||
|
// to enter their critical sections at once, proven by every one of
|
||||||
|
// them reaching the rendezvous point before any is allowed to
|
||||||
|
// proceed.
|
||||||
|
func TestContentLockAllowsDifferentKeys(t *testing.T) {
|
||||||
|
lock := newContentLock()
|
||||||
|
|
||||||
|
const goroutines = 20
|
||||||
|
|
||||||
|
var (
|
||||||
|
wg sync.WaitGroup
|
||||||
|
inside int32
|
||||||
|
reached = make(chan struct{}, goroutines)
|
||||||
|
)
|
||||||
|
|
||||||
|
wg.Add(goroutines)
|
||||||
|
|
||||||
|
release := make(chan struct{})
|
||||||
|
|
||||||
|
for i := 0; i < goroutines; i++ {
|
||||||
|
key := string(rune('a' + i))
|
||||||
|
|
||||||
|
go func() {
|
||||||
|
defer wg.Done()
|
||||||
|
|
||||||
|
unlock := lock.Lock(key)
|
||||||
|
defer unlock()
|
||||||
|
|
||||||
|
atomic.AddInt32(&inside, 1)
|
||||||
|
reached <- struct{}{}
|
||||||
|
<-release
|
||||||
|
}()
|
||||||
|
}
|
||||||
|
|
||||||
|
// Every goroutine must reach the rendezvous point (i.e. acquire its
|
||||||
|
// own key's lock) without needing any other to release first. If
|
||||||
|
// keys were incorrectly serialized onto one underlying lock, only
|
||||||
|
// one would get here and this would time out.
|
||||||
|
for i := 0; i < goroutines; i++ {
|
||||||
|
select {
|
||||||
|
case <-reached:
|
||||||
|
case <-time.After(2 * time.Second):
|
||||||
|
t.Fatalf("only %d/%d goroutines locking distinct keys made progress; "+
|
||||||
|
"keys may be incorrectly serialized", i, goroutines)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if n := atomic.LoadInt32(&inside); n != goroutines {
|
||||||
|
t.Errorf("goroutines inside their critical section = %d, want %d", n, goroutines)
|
||||||
|
}
|
||||||
|
|
||||||
|
close(release)
|
||||||
|
wg.Wait()
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestContentLockRemovesEntryAfterUnlock verifies that the internal
|
||||||
|
// entries map does not grow without bound: once no goroutine holds or
|
||||||
|
// awaits a key, its entry is removed.
|
||||||
|
func TestContentLockRemovesEntryAfterUnlock(t *testing.T) {
|
||||||
|
lock := newContentLock()
|
||||||
|
|
||||||
|
unlock := lock.Lock("k")
|
||||||
|
|
||||||
|
lock.mu.Lock()
|
||||||
|
if _, ok := lock.entries["k"]; !ok {
|
||||||
|
lock.mu.Unlock()
|
||||||
|
t.Fatal("entry missing while lock is held")
|
||||||
|
}
|
||||||
|
lock.mu.Unlock()
|
||||||
|
|
||||||
|
unlock()
|
||||||
|
|
||||||
|
lock.mu.Lock()
|
||||||
|
defer lock.mu.Unlock()
|
||||||
|
|
||||||
|
if _, ok := lock.entries["k"]; ok {
|
||||||
|
t.Error("entry for key still present after the last holder unlocked")
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user