diff --git a/cmd/erasure-server-pool-lock-bulk_test.go b/cmd/erasure-server-pool-lock-bulk_test.go index 12166a2cd..cfb437d96 100644 --- a/cmd/erasure-server-pool-lock-bulk_test.go +++ b/cmd/erasure-server-pool-lock-bulk_test.go @@ -122,9 +122,8 @@ func TestDeleteObjectsInSetsRunsGroupsConcurrently(t *testing.T) { // the result its own delete reports, and a group that completed before keeps // its success. func TestDeleteObjectsInSetsQueuedGroupsCancel(t *testing.T) { - prev := deleteSetGroupsLimit - deleteSetGroupsLimit = 1 - t.Cleanup(func() { deleteSetGroupsLimit = prev }) + deleteSetGroupsLimit.Store(1) + t.Cleanup(func() { deleteSetGroupsLimit.Store(0) }) for _, blocked := range []int{0, 1} { t.Run(fmt.Sprintf("blocked=%d", blocked), func(t *testing.T) { z, bucket := bulkDeletePools(t, 1) diff --git a/cmd/erasure-server-pool-lock-copy_test.go b/cmd/erasure-server-pool-lock-copy_test.go index da088fd5f..24457cfc7 100644 --- a/cmd/erasure-server-pool-lock-copy_test.go +++ b/cmd/erasure-server-pool-lock-copy_test.go @@ -44,38 +44,42 @@ func ldCopyAPI(t *testing.T, mode string) (*erasureServerPools, string, http.Han // A lock is taken on the cleaned name, so an invalid name that aliases // another one locks the same resource. A copy whose source or destination is // such an alias is refused before it locks anything, as the release refused -// it when it read the source. +// it when it read the source: the refusal comes at once while another writer +// holds the cleaned name. func TestLockDomainCopyRejectsAliasNames(t *testing.T) { for _, mode := range append(slices.Clone(ldModes), "pools") { t.Run(mode, func(t *testing.T) { z, bucket, router := ldCopyAPI(t, mode) - prev := globalOperationTimeout - globalOperationTimeout = newDynamicTimeout(5*time.Second, 5*time.Second) - t.Cleanup(func() { globalOperationTimeout = prev }) ldMustPut(t, z, bucket, "alias/x", "victim") + ldMustPut(t, z, bucket, "alias/src", "source") for _, tc := range []struct{ src, dst string }{ {"alias//x", "alias/x"}, {"alias/./x", "alias/x"}, - {"alias/x", "alias//x"}, + {"alias/src", "alias//x"}, + {"alias/src", "alias/./x"}, } { t.Run(fmt.Sprintf("%s->%s", tc.src, tc.dst), func(t *testing.T) { + // Another writer holds the cleaned name of the alias. + release := ldHoldLock(t, z.NewObjectNSLock(bucket, "alias/x")) + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) start := time.Now() - rec := ldCopyRequest(t.Context(), t, router, bucket, tc.src, tc.dst, nil) - if took := time.Since(start); took > time.Second { + rec := ldCopyRequest(ctx, t, router, bucket, tc.src, tc.dst, nil) + took := time.Since(start) + cancel() + release() + if took > time.Second { t.Errorf("refused after %v", took) } if rec.Code != http.StatusBadRequest { t.Errorf("status %d, want 400: %s", rec.Code, rec.Body.String()) } - ctx, cancel := context.WithTimeout(t.Context(), time.Second) - defer cancel() - if err := ldPut(ctx, z, bucket, "alias/x", "victim"); err != nil { - t.Errorf("a PUT of alias/x after the copy: %v", err) - } if body, _ := ldBody(t, z, bucket, "alias/x"); body != "victim" { t.Errorf("alias/x = %q", body) } + if body, _ := ldBody(t, z, bucket, "alias/src"); body != "source" { + t.Errorf("alias/src = %q", body) + } }) } }) @@ -109,37 +113,59 @@ func TestLockDomainCopyErrorPrecedence(t *testing.T) { } // lockObjects merges the requests of one resource in one lock table, also for -// names that differ before cleaning; the same resource in two distributed -// lock domains stays two locks. +// names that differ before cleaning, into one lock that is a write lock when +// either asks for one. The same resource in two distributed lock domains +// stays two locks: a read lock in the alias's domain and a write lock in the +// clean name's. func TestLockObjectsMergesByResource(t *testing.T) { for _, mode := range ldModes { t.Run(mode, func(t *testing.T) { z, bucket := ldPool(t, mode) pool := z.serverPools[0] - prev := globalOperationTimeout - globalOperationTimeout = newDynamicTimeout(3*time.Second, 3*time.Second) - t.Cleanup(func() { globalOperationTimeout = prev }) - for i := range 16 { - clean := fmt.Sprintf("merge-%d/x", i) - alias := fmt.Sprintf("merge-%d//x", i) + split := !pool.sharesLocalNSLock() + // Names whose alias hashes to the clean name's set, and, where the + // sets have their own lockers, names whose alias does not. + pick := func(sameSet bool) (string, string) { + for i := range 1000 { + clean, alias := fmt.Sprintf("merge-%d/x", i), fmt.Sprintf("merge-%d//x", i) + if (pool.getHashedSetIndex(alias) == pool.getHashedSetIndex(clean)) == sameSet { + return clean, alias + } + } + t.Fatalf("no names with sameSet=%v", sameSet) + return "", "" + } + cases := []bool{true} + if split { + cases = append(cases, false) + } + for _, sameSet := range cases { + clean, alias := pick(sameSet) + ctx, cancel := context.WithTimeout(t.Context(), 3*time.Second) start := time.Now() - _, unlock, err := z.lockObjects(t.Context(), + _, unlock, err := z.lockObjects(ctx, objectLockRequest{bucket: bucket, object: alias}, objectLockRequest{bucket: bucket, object: clean, write: true}) + cancel() if err != nil { t.Fatalf("%s + %s: %v after %v", alias, clean, err, time.Since(start)) } if took := time.Since(start); took > time.Second { t.Errorf("%s + %s took %v", alias, clean, took) } - sameTable := pool.sharesLocalNSLock() || pool.getHashedSetIndex(alias) == pool.getHashedSetIndex(clean) - // The resource is write locked in the clean name's table. + // The clean name's table holds a write lock. if ldTry(t, pool.getHashedSet(clean).NewNSLock(bucket, clean), false) { - t.Errorf("%s: not write locked", clean) + t.Errorf("%s: readers admitted in its own table", clean) } - if !sameTable && !ldTry(t, pool.getHashedSet(alias).NewNSLock(bucket, clean), false) { - // The alias holds only a read lock in its own domain. - t.Errorf("%s: alias domain write locked", alias) + if split && !sameSet { + // The alias's table holds the alias's read lock. + aliasSet := pool.getHashedSet(alias) + if !ldTry(t, aliasSet.NewNSLock(bucket, alias), false) { + t.Errorf("%s: readers excluded in its own table", alias) + } + if ldTry(t, aliasSet.NewNSLock(bucket, alias), true) { + t.Errorf("%s: writers admitted in its own table, the alias lock is missing", alias) + } } unlock() } diff --git a/cmd/erasure-server-pool-lock-wait_test.go b/cmd/erasure-server-pool-lock-wait_test.go index c513ace13..4e45194fe 100644 --- a/cmd/erasure-server-pool-lock-wait_test.go +++ b/cmd/erasure-server-pool-lock-wait_test.go @@ -35,13 +35,14 @@ import ( func nsLocksEvents(t *testing.T) chan string { t.Helper() events := make(chan string, 256) - getNSLocksEvent = func(event string, index int) { + f := func(event string, index int) { select { case events <- fmt.Sprintf("%s:%d", event, index): default: } } - t.Cleanup(func() { getNSLocksEvent = nil }) + getNSLocksEvent.Store(&f) + t.Cleanup(func() { getNSLocksEvent.Store(nil) }) return events } @@ -61,6 +62,20 @@ func waitNSLocksEvent(t *testing.T, events chan string, want string) { } } +// ldFreeFresh waits, bounded, until a fresh instance from lock can take the +// lock: a distributed release can still be on its way after it was asked +// for, and a late unlock of an earlier instance never releases a new one. +func ldFreeFresh(t *testing.T, lock func() RWLocker, write bool) bool { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for !ldTry(t, lock(), write) { + if time.Now().After(deadline) { + return false + } + } + return true +} + type nsLocksResult struct { unlock func() err error @@ -104,7 +119,7 @@ func TestGetNSLocksWaitsHoldingNothing(t *testing.T) { waitNSLocksEvent(t, events, "idle-wait:1") // The event says that a was released; a distributed unlock may // still be on its way. - if !ldFree(t, lock("a"), true) { + if !ldFreeFresh(t, func() RWLocker { return lock("a") }, true) { t.Fatal("a is not released while b is waited for") } releaseA := ldHoldLock(t, lock("a")) @@ -137,7 +152,7 @@ func TestGetNSLocksWaitsHoldingNothing(t *testing.T) { t.Error("not both locks held") } res.unlock() - if !ldFree(t, lock("a"), true) || !ldFree(t, lock("b"), true) { + if !ldFreeFresh(t, func() RWLocker { return lock("a") }, true) || !ldFreeFresh(t, func() RWLocker { return lock("b") }, true) { t.Error("locks left behind") } }) @@ -167,7 +182,7 @@ func TestGetNSLocksIdleWaitCancel(t *testing.T) { case <-time.After(5 * time.Second): t.Fatal("a canceled idle wait did not return") } - if !ldFree(t, lock("a"), true) { + if !ldFreeFresh(t, func() RWLocker { return lock("a") }, true) { t.Error("a left behind") } }) @@ -199,7 +214,7 @@ func TestGetNSLocksIdleWaitDeadline(t *testing.T) { case <-time.After(timeout + 5*time.Second): t.Fatal("the idle wait outlived the deadline") } - if !ldFree(t, lock("a"), true) { + if !ldFreeFresh(t, func() RWLocker { return lock("a") }, true) { t.Error("a left behind") } }) @@ -264,7 +279,7 @@ func TestLockDomainCopyDoesNotStarveDestination(t *testing.T) { done <- ldCopyRequest(t.Context(), t, router, bucket, src, dst, nil).Code }() waitNSLocksEvent(t, events, "idle-wait:1") - if !ldFree(t, z.NewObjectNSLock(bucket, dst), true) { + if !ldFreeFresh(t, func() RWLocker { return z.NewObjectNSLock(bucket, dst) }, true) { t.Fatal("the destination is not released while the source is waited for") } for range 3 { diff --git a/cmd/erasure-server-pool-locks.go b/cmd/erasure-server-pool-locks.go index 67d255234..ecc280aa4 100644 --- a/cmd/erasure-server-pool-locks.go +++ b/cmd/erasure-server-pool-locks.go @@ -24,6 +24,7 @@ import ( "slices" "strings" "sync" + "sync/atomic" "time" "github.com/minio/minio-go/v7/pkg/set" @@ -153,11 +154,11 @@ const ( // getNSLocksEvent, when a test sets it, is told when getNSLocks gives up a // bounded wait ("bounded-timeout"), starts to wait for that lock alone // ("idle-wait") and got it ("idle-acquired"), with the index of the lock. -var getNSLocksEvent func(event string, index int) +var getNSLocksEvent atomic.Pointer[func(event string, index int)] func nsLocksEvent(event string, index int) { - if f := getNSLocksEvent; f != nil { - f(event, index) + if f := getNSLocksEvent.Load(); f != nil { + (*f)(event, index) } } @@ -344,7 +345,7 @@ func (z *erasureServerPools) deleteObjectsInSets(ctx context.Context, bucket str } slices.Sort(order) - sem := make(chan struct{}, min(len(order), deleteSetGroupsLimit)) + sem := make(chan struct{}, min(len(order), deleteSetGroupsLimitValue())) var wg sync.WaitGroup for _, idx := range order { members := groups[idx] @@ -359,9 +360,18 @@ func (z *erasureServerPools) deleteObjectsInSets(ctx context.Context, bucket str wg.Wait() } -// deleteSetGroupsLimit bounds how many sets a bulk delete of one pool works -// on at once. -var deleteSetGroupsLimit = 16 +// deleteSetGroupsDefault bounds how many sets a bulk delete of one pool works +// on at once; a test can lower it through deleteSetGroupsLimit. +const deleteSetGroupsDefault = 16 + +var deleteSetGroupsLimit atomic.Int32 + +func deleteSetGroupsLimitValue() int { + if n := int(deleteSetGroupsLimit.Load()); n > 0 { + return n + } + return deleteSetGroupsDefault +} // deleteObjectsInSet deletes the members of objects, which all hash to er, // under er's lock, and writes their results. Nothing else writes them.