From 39278e068e4553dab7cd93df07dde1cf5b5fd752 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Mon, 26 Oct 2026 21:11:49 +0800 Subject: [PATCH] fix: isolate storage mutation guards by exact resource Replace process-wide hash-slot mutexes with reference-counted locks keyed by drive and path. Retain entries before waiting, acquire unique paths in order, and reclaim entries only after all holders and waiters finish. Add real storage API regressions for unrelated colliding objects and drives, plus coverage for cancellation, multi-path order and waiter lifetimes. Signed-off-by: Feng Ruohang --- cmd/object-mutation-isolation_test.go | 439 ++++++++++++++++++++++++++ cmd/object-mutation.go | 69 ++-- 2 files changed, 485 insertions(+), 23 deletions(-) create mode 100644 cmd/object-mutation-isolation_test.go diff --git a/cmd/object-mutation-isolation_test.go b/cmd/object-mutation-isolation_test.go new file mode 100644 index 000000000..072f8453e --- /dev/null +++ b/cmd/object-mutation-isolation_test.go @@ -0,0 +1,439 @@ +// Copyright (c) 2026 Feng Ruohang +// +// This file is part of Silo Object Storage. +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package cmd + +import ( + "context" + "errors" + "fmt" + "path/filepath" + "slices" + "sync" + "sync/atomic" + "testing" + "time" +) + +// Keep the old slot calculation here to reproduce the false sharing between +// distinct resources, including resources on separate storage drives. +func mutationGuardLegacySlot(drive, path string) uint64 { + h := uint64(14695981039346656037) + for _, part := range []string{drive, path} { + for i := range part { + h = (h ^ uint64(part[i])) * 1099511628211 + } + h *= 1099511628211 + } + return h % 65536 +} + +func mutationGuardCollision(t *testing.T, drive string, target uint64) string { + t.Helper() + for i := 0; i < 2000000; i++ { + key := fmt.Sprintf("independent-%d", i) + if mutationGuardLegacySlot(drive, pathJoin("bucket", key)) == target { + return key + } + } + t.Fatal("no collision found in bounded search") + return "" +} + +func TestMutationGuardIsolatesStorageResources(t *testing.T) { + blocked, rootA, err := newXLStorageTestSetup(t) + if err != nil { + t.Fatal(err) + } + defer blocked.Close() + healthy, rootB, err := newXLStorageTestSetup(t) + if err != nil { + t.Fatal(err) + } + defer healthy.Close() + defer mutationGuardCheckEmpty(t, rootA, rootB) + for _, disk := range []*xlStorageDiskIDCheck{blocked, healthy} { + if err := disk.MakeVol(t.Context(), "bucket"); err != nil { + t.Fatal(err) + } + } + const object = "blocked" + target := mutationGuardLegacySlot(rootA, pathJoin("bucket", object)) + type mutation struct { + name, key string + disk *xlStorageDiskIDCheck + done chan error + } + mutations := []mutation{ + {name: "same-key", key: object, disk: blocked}, + {name: "different-key-same-drive-collision", key: mutationGuardCollision(t, rootA, target), disk: blocked}, + {name: "different-key-healthy-drive-collision", key: mutationGuardCollision(t, rootB, target), disk: healthy}, + {name: "same-path-healthy-drive", key: object, disk: healthy}, + } + fixture := func(name string) FileInfo { + fi := newFileInfo(name, 2, 2) + fi.DataDir, fi.ModTime, fi.Erasure.Index = mustGetUUID(), UTCNow(), 1 + fi.Fresh = true + fi.Metadata = map[string]string{"x-amz-meta-probe": "before"} + return fi + } + for _, m := range mutations { + if err := m.disk.WriteMetadata(t.Context(), "", "bucket", m.key, fixture(m.key)); err != nil { + t.Fatal(err) + } + } + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + var lastSuccess int64 + gate := &stagedRenameContext{ + Context: ctx, meta: filepath.Join(rootA, minioMetaTmpBucket, "staged", xlStorageFormatFile), + entered: make(chan struct{}), release: make(chan struct{}), + tracker: healthDiskCtxValue{lastSuccess: &lastSuccess}, + } + var once sync.Once + release := func() { once.Do(func() { close(gate.release) }) } + var workers sync.WaitGroup + defer func() { + release() + workers.Wait() + }() + holderDone := make(chan error, 1) + workers.Go(func() { + _, err := blocked.RenameData(gate, minioMetaTmpBucket, "staged", fixture(object), "bucket", object, RenameOptions{}) + holderDone <- err + }) + select { + case <-gate.entered: + case <-time.After(5 * time.Second): + t.Fatal("RenameData did not reach the staged metadata barrier") + } + // Cancellation cannot release the guard while the admitted worker can + // still commit. Unrelated mutations must nevertheless be able to finish. + cancel() + for i := range mutations { + m := &mutations[i] + m.done = make(chan error, 1) + workers.Go(func() { + m.done <- m.disk.UpdateMetadata(t.Context(), "bucket", m.key, FileInfo{ + Metadata: map[string]string{"x-amz-meta-probe": "after"}, + }, UpdateMetadataOpts{}) + }) + } + mutationGuardWaitRefs(t, rootA, pathJoin("bucket", object), 2) + for _, m := range mutations[1:] { + select { + case err := <-m.done: + if err != nil { + t.Errorf("%s: %v", m.name, err) + } + t.Logf("%s completed while RenameData remained paused", m.name) + case <-time.After(2 * time.Second): + t.Errorf("%s blocked by an unrelated paused mutation", m.name) + } + } + select { + case err := <-holderDone: + t.Fatalf("canceled RenameData returned before its worker finished: %v", err) + default: + } + select { + case err := <-mutations[0].done: + t.Fatalf("same-key mutation escaped the held guard: %v", err) + default: + } + release() + select { + case err := <-holderDone: + if !errors.Is(err, context.Canceled) { + t.Errorf("canceled RenameData result: %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("RenameData did not finish after releasing the barrier") + } + select { + case err := <-mutations[0].done: + if err != nil { + t.Fatal(err) + } + case <-time.After(5 * time.Second): + t.Fatal("same-key mutation did not finish after releasing the holder") + } + workers.Wait() + for _, m := range mutations { + fi, err := m.disk.ReadVersion(t.Context(), "", "bucket", m.key, "", ReadOptions{}) + if err != nil || fi.Metadata["x-amz-meta-probe"] != "after" { + t.Errorf("%s persisted metadata: %q, %v", m.name, fi.Metadata["x-amz-meta-probe"], err) + } + } +} + +func mutationGuardCheckEmpty(t *testing.T, drives ...string) { + t.Helper() + storageMutationLocks.Lock() + defer storageMutationLocks.Unlock() + for key, lock := range storageMutationLocks.locks { + if slices.Contains(drives, key.drive) { + t.Errorf("unused guard retained for %+v with %d references", key, lock.refs) + } + } +} + +func mutationGuardWaitRefs(t *testing.T, drive, path string, refs int) *storageMutationLock { + t.Helper() + deadline := time.NewTimer(5 * time.Second) + defer deadline.Stop() + tick := time.NewTicker(time.Millisecond) + defer tick.Stop() + key := storageMutationKey{drive, path} + for { + storageMutationLocks.Lock() + lock := storageMutationLocks.locks[key] + ready := lock != nil && lock.refs == refs + storageMutationLocks.Unlock() + if ready { + return lock + } + select { + case <-deadline.C: + t.Fatalf("guard %+v did not reach %d references", key, refs) + case <-tick.C: + } + } +} + +func TestMutationGuardRetainsWaitersAndRemovesIdleEntries(t *testing.T) { + s := &xlStorage{drivePath: t.TempDir()} + defer mutationGuardCheckEmpty(t, s.drivePath) + unlock := sync.OnceFunc(s.lockMutations("object")) + defer unlock() + original := mutationGuardWaitRefs(t, s.drivePath, "object", 1) + acquired, release, finished := make(chan struct{}), make(chan struct{}), make(chan struct{}) + unblock := sync.OnceFunc(func() { close(release) }) + defer unblock() + go func() { + defer close(finished) + defer s.lockMutations("object", "object")() + close(acquired) + <-release + }() + if lock := mutationGuardWaitRefs(t, s.drivePath, "object", 2); lock != original { + t.Error("waiting operation registered a second mutex for the same resource") + } + select { + case <-acquired: + t.Error("waiter acquired before the holder released") + default: + } + unlock() + select { + case <-acquired: + case <-time.After(5 * time.Second): + t.Fatal("waiter did not acquire after release") + } + if lock := mutationGuardWaitRefs(t, s.drivePath, "object", 1); lock != original { + t.Error("holder release discarded the queued operation's mutex") + } + unblock() + select { + case <-finished: + case <-time.After(5 * time.Second): + t.Fatal("waiter did not release") + } + mutationGuardCheckEmpty(t, s.drivePath) + // The registry must not grow with the historical object count. + for i := 0; i < 1000; i++ { + s.lockMutations(fmt.Sprintf("object-%d", i))() + } + s.lockMutations()() +} + +func TestMutationGuardOrdersAndDeduplicatesPaths(t *testing.T) { + s := &xlStorage{drivePath: t.TempDir()} + defer mutationGuardCheckEmpty(t, s.drivePath) + paths := []string{"b", "a", "b", "a"} + before := slices.Clone(paths) + unlock := s.lockMutations(paths...) + mutationGuardWaitRefs(t, s.drivePath, "a", 1) + mutationGuardWaitRefs(t, s.drivePath, "b", 1) + unlock() + if !slices.Equal(paths, before) { + t.Fatalf("changed caller's path slice: %v", paths) + } + const workers, iterations = 8, 500 + start, done := make(chan struct{}), make(chan struct{}) + var wg sync.WaitGroup + var protected atomic.Int64 + var active atomic.Int32 + for i := 0; i < workers; i++ { + worker := i + wg.Go(func() { + <-start + order := []string{"a", "b", "a"} + if worker%2 != 0 { + order = []string{"b", "a", "b"} + } + for j := 0; j < iterations; j++ { + unlock := s.lockMutations(order...) + if active.Add(1) != 1 { + t.Error("overlapping guards entered concurrently") + } + protected.Add(1) + active.Add(-1) + unlock() + } + }) + } + go func() { wg.Wait(); close(done) }() + close(start) + select { + case <-done: + if got := protected.Load(); got != workers*iterations { + t.Fatalf("overlapping guards lost updates: got %d, want %d", got, workers*iterations) + } + case <-time.After(5 * time.Second): + t.Fatal("opposing path orders or duplicate paths deadlocked") + } +} + +// Measure registry/allocation cost separately from storage I/O. Independent +// workers use two resources, as RenameData does, and shared workers use one. +func BenchmarkMutationGuard(b *testing.B) { + s := &xlStorage{drivePath: "/benchmark/drive"} + b.Run("single", func(b *testing.B) { + b.ReportAllocs() + for b.Loop() { + s.lockMutations("bucket/object")() + } + }) + b.Run("parallel-independent", func(b *testing.B) { + b.ReportAllocs() + var worker atomic.Uint64 + b.RunParallel(func(pb *testing.PB) { + id := worker.Add(1) + paths := []string{fmt.Sprintf("staging/%d", id), fmt.Sprintf("bucket/object-%d", id)} + for pb.Next() { + s.lockMutations(paths...)() + } + }) + }) + b.Run("parallel-shared", func(b *testing.B) { + b.ReportAllocs() + b.RunParallel(func(pb *testing.PB) { + for pb.Next() { + s.lockMutations("bucket/object")() + } + }) + }) +} + +// A call must retain paths it has not reached yet, even across xlStorage +// instances for the same drive. No resource wait may hold the registry mutex. +func TestMutationGuardRetainsUnacquiredPaths(t *testing.T) { + s := &xlStorage{drivePath: t.TempDir()} + sameDrive := &xlStorage{drivePath: s.drivePath} + defer mutationGuardCheckEmpty(t, s.drivePath) + unlockA := sync.OnceFunc(s.lockMutations("a")) + defer unlockA() + acquired, release, done := make(chan struct{}), make(chan struct{}), make(chan struct{}) + unblock := sync.OnceFunc(func() { close(release) }) + defer unblock() + go func() { + defer close(done) + defer sameDrive.lockMutations("b", "a")() + close(acquired) + <-release + }() + mutationGuardWaitRefs(t, s.drivePath, "a", 2) + original := mutationGuardWaitRefs(t, s.drivePath, "b", 1) + // The queued call has not locked b yet: sorting must put a first. + unlockB := sync.OnceFunc(s.lockMutations("b")) + defer unlockB() + if got := mutationGuardWaitRefs(t, s.drivePath, "b", 2); got != original { + t.Fatal("unacquired path got a different mutex") + } + unlockA() + select { + case <-acquired: + t.Fatal("multi-path call escaped the independently held second path") + default: + } + unlockB() + select { + case <-acquired: + case <-time.After(5 * time.Second): + t.Fatal("multi-path call did not acquire after both paths were released") + } + if got := mutationGuardWaitRefs(t, s.drivePath, "b", 1); got != original { + t.Fatal("release replaced an entry retained by a pending multi-path call") + } + unblock() + <-done + // The variadic helper has no hidden four-path limit. + s.lockMutations("e", "d", "c", "b", "a", "a")() +} + +func TestMutationGuardCanceledWaiterRetainsSerialization(t *testing.T) { + disk, root, err := newXLStorageTestSetup(t) + if err != nil { + t.Fatal(err) + } + defer disk.Close() + defer mutationGuardCheckEmpty(t, root) + if err := disk.MakeVol(t.Context(), "bucket"); err != nil { + t.Fatal(err) + } + fi := newFileInfo("object", 2, 2) + fi.DataDir, fi.ModTime, fi.Erasure.Index = mustGetUUID(), UTCNow(), 1 + fi.Fresh = true + fi.Metadata = map[string]string{"x-amz-meta-probe": "before"} + if err := disk.WriteMetadata(t.Context(), "", "bucket", "object", fi); err != nil { + t.Fatal(err) + } + // Use a separate instance with the same drive identity as the API wrapper. + s := &xlStorage{drivePath: root} + unlock := sync.OnceFunc(s.lockMutations("bucket/object")) + defer unlock() + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + done := make(chan error, 1) + go func() { + done <- disk.UpdateMetadata(ctx, "bucket", "object", FileInfo{ + Metadata: map[string]string{"x-amz-meta-probe": "canceled"}, + }, UpdateMetadataOpts{}) + }() + mutationGuardWaitRefs(t, root, "bucket/object", 2) + cancel() + select { + case err := <-done: + t.Fatalf("canceled API returned with worker still queued: %v", err) + case <-time.After(50 * time.Millisecond): + } + mutationGuardWaitRefs(t, root, "bucket/object", 2) + unlock() + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Fatalf("canceled waiter: %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("canceled waiter did not finish after holder release") + } + got, err := disk.ReadVersion(t.Context(), "", "bucket", "object", "", ReadOptions{}) + if err != nil || got.Metadata["x-amz-meta-probe"] != "before" { + t.Fatalf("canceled metadata update persisted: %v, %v", got.Metadata, err) + } +} diff --git a/cmd/object-mutation.go b/cmd/object-mutation.go index 788ebcdc6..baa05a5d4 100644 --- a/cmd/object-mutation.go +++ b/cmd/object-mutation.go @@ -17,36 +17,59 @@ import ( // mutation of the same version revokes it, including a no-op metadata update. const rollbackIDKey = ReservedMetadataPrefixLower + "rollback-id" -// Bound memory independently of object count. Include the physical drive so -// shards of one object are not serialized across drives. Multiple paths are -// locked in slot order, including collisions, to avoid lock inversion. -var storageMutationLocks [65536]sync.Mutex +type storageMutationKey struct { + drive, path string +} + +type storageMutationLock struct { + mu sync.Mutex + refs int // Holders and waiters; protected by the registry mutex. +} + +// Keep exact resource locks only while they have holders or waiters. The +// registry mutex protects references, never storage I/O or resource waits. +var storageMutationLocks = struct { + sync.Mutex + locks map[storageMutationKey]*storageMutationLock +}{locks: make(map[storageMutationKey]*storageMutationLock)} func (s *xlStorage) lockMutations(paths ...string) func() { - var slots [4]int - n := 0 - for _, path := range paths { - h := uint64(14695981039346656037) - for _, part := range []string{s.drivePath, path} { - for i := range part { - h = (h ^ uint64(part[i])) * 1099511628211 - } - h *= 1099511628211 - } - slot := int(h % uint64(len(storageMutationLocks))) - if !slices.Contains(slots[:n], slot) { - slots[n] = slot - n++ + // Each call uses one drive. Sorting its unique paths gives all overlapping + // calls the same acquisition order without changing the caller's slice. + paths = slices.Clone(paths) + slices.Sort(paths) + paths = slices.Compact(paths) + drive := s.drivePath + locks := make([]*storageMutationLock, len(paths)) + // Retain every entry before waiting so unlock cannot remove a mutex that + // a queued operation still needs. + storageMutationLocks.Lock() + for i, path := range paths { + key := storageMutationKey{drive, path} + lock := storageMutationLocks.locks[key] + if lock == nil { + lock = &storageMutationLock{} + storageMutationLocks.locks[key] = lock } + lock.refs++ + locks[i] = lock } - slices.Sort(slots[:n]) - for _, slot := range slots[:n] { - storageMutationLocks[slot].Lock() + storageMutationLocks.Unlock() + for _, lock := range locks { + lock.mu.Lock() } return func() { - for i := n - 1; i >= 0; i-- { - storageMutationLocks[slots[i]].Unlock() + for i := len(locks) - 1; i >= 0; i-- { + locks[i].mu.Unlock() } + storageMutationLocks.Lock() + for i, path := range paths { + locks[i].refs-- + if locks[i].refs == 0 { + delete(storageMutationLocks.locks, storageMutationKey{drive, path}) + } + } + storageMutationLocks.Unlock() } }