diff --git a/cmd/erasure-server-pool-lock-bulk_test.go b/cmd/erasure-server-pool-lock-bulk_test.go new file mode 100644 index 000000000..12166a2cd --- /dev/null +++ b/cmd/erasure-server-pool-lock-bulk_test.go @@ -0,0 +1,195 @@ +// Copyright (c) 2026 PGSTY +// +// This file is part of Silo Object Storage stack +// +// 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" + "sync" + "sync/atomic" + "testing" + "time" +) + +// A bulk delete of a single pool whose sets have their own lockers deletes +// the groups of different sets concurrently, as the sets of one bulk delete +// did before the lock-domain change. + +// bulkGateDisk calls enter before the bulk deletes of its set and counts +// them. +type bulkGateDisk struct { + StorageAPI + enter func(ctx context.Context) error + calls *atomic.Int32 +} + +func (d bulkGateDisk) DeleteVersions(ctx context.Context, volume string, versions []FileInfoVersions, opts DeleteOptions) []error { + d.calls.Add(1) + if err := d.enter(ctx); err != nil { + errs := make([]error, len(versions)) + for i := range errs { + errs[i] = err + } + return errs + } + return d.StorageAPI.DeleteVersions(ctx, volume, versions, opts) +} + +// bulkSplit makes the single pool of z n sets with their own local lock +// tables that share its drives. enter(set) runs before each bulk delete on a +// drive of that set; calls counts them per set. +func bulkSplit(t *testing.T, z *erasureServerPools, n int, enter func(ctx context.Context, set int) error) []*atomic.Int32 { + t.Helper() + pool := z.serverPools[0] + sets, count := pool.sets, pool.setCount + t.Cleanup(func() { pool.sets, pool.setCount = sets, count }) + disks := sets[0].getDisks + calls := make([]*atomic.Int32, n) + split := make([]*erasureObjects, n) + for i := range n { + set := *sets[0] + set.setIndex = i + set.nsMutex = newNSLock(false) + calls[i] = &atomic.Int32{} + set.getDisks = func() []StorageAPI { + ds := disks() + for j, d := range ds { + if d != nil { + ds[j] = bulkGateDisk{StorageAPI: d, enter: func(ctx context.Context) error { return enter(ctx, i) }, calls: calls[i]} + } + } + return ds + } + split[i] = &set + } + pool.sets, pool.setCount = split, n + return calls +} + +func TestDeleteObjectsInSetsRunsGroupsConcurrently(t *testing.T) { + z, bucket := bulkDeletePools(t, 1) + // The first bulk delete of each set waits until the other set's first + // one has started too: run one after the other, they cannot meet. + var arrived sync.WaitGroup + arrived.Add(2) + once := []sync.Once{{}, {}} + var missed atomic.Bool + bulkSplit(t, z, 2, func(ctx context.Context, set int) error { + once[set].Do(func() { + arrived.Done() + met := make(chan struct{}) + go func() { arrived.Wait(); close(met) }() + select { + case <-met: + case <-time.After(5 * time.Second): + missed.Store(true) + } + }) + return nil + }) + k0, k1 := ldKey(t, z, "concurrent", 0), ldKey(t, z, "concurrent", 1) + ldMustPut(t, z, bucket, k0, "zero") + ldMustPut(t, z, bucket, k1, "one") + _, errs := z.DeleteObjects(t.Context(), bucket, []ObjectToDelete{{ObjectV: ObjectV{ObjectName: k0}}, {ObjectV: ObjectV{ObjectName: k1}}}, ObjectOptions{}) + if err := errors.Join(errs...); err != nil { + t.Fatal(err) + } + if missed.Load() { + t.Fatal("the groups of the two sets ran one after the other") + } + ldAbsent(t, z, bucket, k0) + ldAbsent(t, z, bucket, k1) +} + +// With fewer workers than groups, a group that is admitted after the request +// was canceled does no storage work. A group in flight at the cancel keeps +// 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 }) + for _, blocked := range []int{0, 1} { + t.Run(fmt.Sprintf("blocked=%d", blocked), func(t *testing.T) { + z, bucket := bulkDeletePools(t, 1) + entered, gate := make(chan struct{}, 1), make(chan struct{}) + var once sync.Once + calls := bulkSplit(t, z, 3, func(ctx context.Context, set int) error { + if set == blocked { + once.Do(func() { entered <- struct{}{} }) + <-gate + } + return nil + }) + keys := make([]string, 3) + objects := make([]ObjectToDelete, 3) + for set := range keys { + keys[set] = ldKey(t, z, fmt.Sprintf("queued-%d", blocked), set) + ldMustPut(t, z, bucket, keys[set], "body") + objects[set] = ObjectToDelete{ObjectV: ObjectV{ObjectName: keys[set]}} + } + for _, c := range calls { + c.Store(0) + } + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + dobjects := make([]DeletedObject, 3) + for i := range dobjects { + dobjects[i].ObjectName = keys[i] + } + derrs := make([]error, 3) + done := make(chan struct{}) + go func() { + defer close(done) + z.deleteObjectsInSets(ctx, bucket, objects, ObjectOptions{}, dobjects, derrs, newDynamicTimeout(10*time.Second, 10*time.Second)) + }() + select { + case <-entered: + case <-time.After(10 * time.Second): + t.Fatal("the blocked group did not start") + } + cancel() + close(gate) + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("deleteObjectsInSets did not return") + } + for set := range keys { + switch { + case set < blocked: + if derrs[set] != nil || !dobjects[set].found { + t.Errorf("set %d completed before the cancel: %v found=%v", set, derrs[set], dobjects[set].found) + } + case set == blocked: + if !errors.Is(derrs[set], context.Canceled) { + t.Errorf("set %d in flight: %v, want context.Canceled", set, derrs[set]) + } + default: + if !errors.Is(derrs[set], context.Canceled) { + t.Errorf("queued set %d: %v, want context.Canceled", set, derrs[set]) + } + if n := calls[set].Load(); n != 0 { + t.Errorf("queued set %d made %d drive calls", set, n) + } + } + } + }) + } +} diff --git a/cmd/erasure-server-pool-locks.go b/cmd/erasure-server-pool-locks.go index d0f2b2799..67d255234 100644 --- a/cmd/erasure-server-pool-locks.go +++ b/cmd/erasure-server-pool-locks.go @@ -23,6 +23,7 @@ import ( "math/rand/v2" "slices" "strings" + "sync" "time" "github.com/minio/minio-go/v7/pkg/set" @@ -318,16 +319,18 @@ func (s *erasureSets) sharesLocalNSLock() bool { // deleteObjectsInSets is the bulk delete of a single pool whose sets do not // share one lock table. The names are grouped by the set they hash to, the -// lock domain of their writers. The groups run one after another in ascending -// set index; each takes its set's lock over all of its names in one -// acquisition, deletes its sub-batch, and releases the lock before the next -// group starts. No two sets' locks are held at once. +// lock domain of their writers. The groups run concurrently, up to +// deleteSetGroupsLimit at a time, as the sets of one bulk delete did before; +// each takes its set's lock over all of its names in one acquisition, deletes +// its sub-batch, and releases the lock. A group holds only its own set's lock +// and waits for no other, so the groups cannot hold each other. // // Each group's lock is taken in the request's context. A group whose lock -// cannot be taken fails as a whole and the next group still runs; once the -// request is canceled the remaining groups fail without being attempted. The -// results keep their positions in objects. Replication, events and auditing -// stay with the callers, on the merged results. +// cannot be taken fails as a whole and the other groups still run; a group +// that starts after the request was canceled fails without storage work. +// Each result is written only by the group of its object and keeps its +// position in objects. Replication, events and auditing stay with the +// callers, on the merged results. func (z *erasureServerPools) deleteObjectsInSets(ctx context.Context, bucket string, objects []ObjectToDelete, opts ObjectOptions, dobjects []DeletedObject, derrs []error, timeout *dynamicTimeout) { pool := z.serverPools[0] groups := make(map[int][]int) @@ -341,45 +344,63 @@ func (z *erasureServerPools) deleteObjectsInSets(ctx context.Context, bucket str } slices.Sort(order) + sem := make(chan struct{}, min(len(order), deleteSetGroupsLimit)) + var wg sync.WaitGroup for _, idx := range order { members := groups[idx] - fail := func(err error) { - for _, i := range members { - derrs[i] = err - } + wg.Add(1) + sem <- struct{}{} + go func() { + defer wg.Done() + defer func() { <-sem }() + z.deleteObjectsInSet(ctx, bucket, pool.sets[idx], objects, members, opts, dobjects, derrs, timeout) + }() + } + wg.Wait() +} + +// deleteSetGroupsLimit bounds how many sets a bulk delete of one pool works +// on at once. +var deleteSetGroupsLimit = 16 + +// deleteObjectsInSet deletes the members of objects, which all hash to er, +// under er's lock, and writes their results. Nothing else writes them. +func (z *erasureServerPools) deleteObjectsInSet(ctx context.Context, bucket string, er *erasureObjects, objects []ObjectToDelete, members []int, opts ObjectOptions, dobjects []DeletedObject, derrs []error, timeout *dynamicTimeout) { + fail := func(err error) { + for _, i := range members { + derrs[i] = err } - if err := ctx.Err(); err != nil { - fail(err) + } + if err := ctx.Err(); err != nil { + fail(err) + return + } + names := set.NewStringSet() + batch := make([]ObjectToDelete, len(members)) + for j, i := range members { + names.Add(objects[i].ObjectName) + batch[j] = objects[i] + } + resources := names.ToSlice() + lkctx, unlock, err := getNSLocks(ctx, []nsLockRequest{{ + newLock: func() RWLocker { return er.NewNSLock(bucket, resources...) }, + write: true, + atomic: er.nsMutex.isDistErasure || len(resources) == 1, + }}, timeout) + if err != nil { + fail(err) + return + } + deleted, errs := er.DeleteObjects(lkctx.Context(), bucket, batch, opts) + unlock() + // Merge as the pools loop of DeleteObjects does for one pool. + for j, i := range members { + if errs[j] == nil && deleted[j].found { + dobjects[i], derrs[i] = deleted[j], nil continue } - er := pool.sets[idx] - names := set.NewStringSet() - batch := make([]ObjectToDelete, len(members)) - for j, i := range members { - names.Add(objects[i].ObjectName) - batch[j] = objects[i] - } - resources := names.ToSlice() - lkctx, unlock, err := getNSLocks(ctx, []nsLockRequest{{ - newLock: func() RWLocker { return er.NewNSLock(bucket, resources...) }, - write: true, - atomic: er.nsMutex.isDistErasure || len(resources) == 1, - }}, timeout) - if err != nil { - fail(err) - continue - } - deleted, errs := er.DeleteObjects(lkctx.Context(), bucket, batch, opts) - unlock() - // Merge as the pools loop of DeleteObjects does for one pool. - for j, i := range members { - if errs[j] == nil && deleted[j].found { - dobjects[i], derrs[i] = deleted[j], nil - continue - } - if derrs[i] == nil { - dobjects[i], derrs[i] = deleted[j], errs[j] - } + if derrs[i] == nil { + dobjects[i], derrs[i] = deleted[j], errs[j] } } }