diff --git a/internal/imgcache/contentlock.go b/internal/imgcache/contentlock.go new file mode 100644 index 0000000..3dac7dc --- /dev/null +++ b/internal/imgcache/contentlock.go @@ -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() + } +} diff --git a/internal/imgcache/contentlock_test.go b/internal/imgcache/contentlock_test.go new file mode 100644 index 0000000..1eb56e5 --- /dev/null +++ b/internal/imgcache/contentlock_test.go @@ -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") + } +}