diff --git a/weed/command/master.go b/weed/command/master.go index 2a5a409678..b12f0a6210 100644 --- a/weed/command/master.go +++ b/weed/command/master.go @@ -95,7 +95,7 @@ func init() { m.volumeSizeLimitMB = cmdMaster.Flag.Uint("volumeSizeLimitMB", util.DefaultVolumeSizeLimitMB, "Master stops directing writes to oversized volumes.") m.fileSizeLimitMB = cmdMaster.Flag.Int("fileSizeLimitMB", 256, "limit the file size accepted by /submit, should match the volume servers' -fileSizeLimitMB (-volume.fileSizeLimitMB under weed server or weed mini, which set this for you)") m.volumePreallocate = cmdMaster.Flag.Bool("volumePreallocate", false, "Preallocate disk space for volumes.") - m.maxParallelVacuumPerServer = cmdMaster.Flag.Int("maxParallelVacuumPerServer", 1, "maximum number of volumes to vacuum in parallel per volume server") + m.maxParallelVacuumPerServer = cmdMaster.Flag.Int("maxParallelVacuumPerServer", 1, "maximum number of volumes to vacuum in parallel per volume server; 0 for no limit (deprecated, use master.vacuum.max_parallel_per_server in master.toml)") m.maxWritableVolumes = cmdMaster.Flag.Int("maxWritableVolumes", 0, "maximum writable volumes in a volume layout (collection + replica placement + ttl + disk type) assignments are spread over, fullest picked first; 0 means no limit") m.vacuumIntervalSeconds = cmdMaster.Flag.Int("vacuumIntervalSeconds", 840, "seconds between automatic vacuum sweeps") m.vacuumDeleteEmptyAfterSeconds = cmdMaster.Flag.Int("vacuumDeleteEmptyAfterSeconds", 0, "automatic sweep deletes volume copies that stay empty this many seconds; 0 disables") diff --git a/weed/command/mini.go b/weed/command/mini.go index 381e438e9c..0d2660438e 100644 --- a/weed/command/mini.go +++ b/weed/command/mini.go @@ -445,7 +445,7 @@ func initMiniMasterFlags() { miniMasterOptions.peers = cmdMini.Flag.String("master.peers", "", "all master nodes in comma separated ip:masterPort list (default: none for single master)") miniMasterOptions.volumeSizeLimitMB = cmdMini.Flag.Uint("master.volumeSizeLimitMB", defaultMiniVolumeSizeMB, "Master stops directing writes to oversized volumes (default: 128MB for mini)") miniMasterOptions.volumePreallocate = cmdMini.Flag.Bool("master.volumePreallocate", false, "Preallocate disk space for volumes.") - miniMasterOptions.maxParallelVacuumPerServer = cmdMini.Flag.Int("master.maxParallelVacuumPerServer", 1, "maximum number of volumes to vacuum in parallel on one volume server") + miniMasterOptions.maxParallelVacuumPerServer = cmdMini.Flag.Int("master.maxParallelVacuumPerServer", 1, "maximum number of volumes to vacuum in parallel on one volume server; 0 for no limit (deprecated, use master.vacuum.max_parallel_per_server in master.toml)") miniMasterOptions.maxWritableVolumes = cmdMini.Flag.Int("master.maxWritableVolumes", 0, "maximum writable volumes in a volume layout (collection + replica placement + ttl + disk type) assignments are spread over, fullest picked first; 0 means no limit") miniMasterOptions.vacuumIntervalSeconds = cmdMini.Flag.Int("master.vacuumIntervalSeconds", 840, "seconds between automatic vacuum sweeps") miniMasterOptions.vacuumDeleteEmptyAfterSeconds = cmdMini.Flag.Int("master.vacuumDeleteEmptyAfterSeconds", 0, "automatic sweep deletes volume copies that stay empty this many seconds; 0 disables") diff --git a/weed/command/scaffold/master.toml b/weed/command/scaffold/master.toml index b7a344e231..3e06fa2259 100644 --- a/weed/command/scaffold/master.toml +++ b/weed/command/scaffold/master.toml @@ -11,6 +11,15 @@ scripts = """{{DEFAULT_MAINTENANCE_SCRIPTS}}""" sleep_minutes = 17 # sleep minutes between each script execution +# vacuum reclaims deleted space by compacting volumes +[master.vacuum] +# concurrent volume compactions per volume server; overrides the deprecated +# -maxParallelVacuumPerServer flag (default 1) when set; 0 = no limit +# max_parallel_per_server = 1 +# concurrent volume copies compacting within one rack; 0 = no limit +max_parallel_per_rack = 0 + + [master.sequencer] type = "raft" # Choose [raft|snowflake] type for storing the file id sequence # when sequencer.type = snowflake, the snowflake id must be different from other masters diff --git a/weed/command/server.go b/weed/command/server.go index 0277ccf421..5e41c6e3bb 100644 --- a/weed/command/server.go +++ b/weed/command/server.go @@ -97,7 +97,7 @@ func init() { masterOptions.peers = cmdServer.Flag.String("master.peers", "", "all master nodes in comma separated ip:masterPort list") masterOptions.volumeSizeLimitMB = cmdServer.Flag.Uint("master.volumeSizeLimitMB", util.DefaultVolumeSizeLimitMB, "Master stops directing writes to oversized volumes.") masterOptions.volumePreallocate = cmdServer.Flag.Bool("master.volumePreallocate", false, "Preallocate disk space for volumes.") - masterOptions.maxParallelVacuumPerServer = cmdServer.Flag.Int("master.maxParallelVacuumPerServer", 1, "maximum number of volumes to vacuum in parallel on one volume server") + masterOptions.maxParallelVacuumPerServer = cmdServer.Flag.Int("master.maxParallelVacuumPerServer", 1, "maximum number of volumes to vacuum in parallel on one volume server; 0 for no limit (deprecated, use master.vacuum.max_parallel_per_server in master.toml)") masterOptions.maxWritableVolumes = cmdServer.Flag.Int("master.maxWritableVolumes", 0, "maximum writable volumes in a volume layout (collection + replica placement + ttl + disk type) assignments are spread over, fullest picked first; 0 means no limit") masterOptions.vacuumIntervalSeconds = cmdServer.Flag.Int("master.vacuumIntervalSeconds", 840, "seconds between automatic vacuum sweeps") masterOptions.vacuumDeleteEmptyAfterSeconds = cmdServer.Flag.Int("master.vacuumDeleteEmptyAfterSeconds", 0, "automatic sweep deletes volume copies that stay empty this many seconds; 0 disables") diff --git a/weed/server/master_grpc_server_volume.go b/weed/server/master_grpc_server_volume.go index 53bd2c27f0..de1e54333c 100644 --- a/weed/server/master_grpc_server_volume.go +++ b/weed/server/master_grpc_server_volume.go @@ -399,7 +399,7 @@ func (ms *MasterServer) VacuumVolume(ctx context.Context, req *master_pb.VacuumV resp := &master_pb.VacuumVolumeResponse{} - ms.Topo.Vacuum(ms.grpcDialOption, float64(req.GarbageThreshold), ms.option.MaxParallelVacuumPerServer, req.VolumeId, req.Collection, ms.preallocateSize, false, 0) + ms.Topo.Vacuum(ms.grpcDialOption, float64(req.GarbageThreshold), ms.option.MaxParallelVacuumPerServer, ms.option.MaxParallelVacuumPerRack, req.VolumeId, req.Collection, ms.preallocateSize, false, 0) return resp, nil } diff --git a/weed/server/master_server.go b/weed/server/master_server.go index 6a5c5e7ac9..dec88ab154 100644 --- a/weed/server/master_server.go +++ b/weed/server/master_server.go @@ -54,6 +54,7 @@ type MasterOption struct { FileSizeLimitMB int VolumePreallocate bool MaxParallelVacuumPerServer int + MaxParallelVacuumPerRack int MaxWritableVolumes int VacuumIntervalSeconds int VacuumDeleteEmptyAfterSeconds int @@ -128,6 +129,11 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers map[string]pb.Se option.VolumeGrowthDisabled = v.GetBool("master.volume_growth.disable") v.SetDefault("master.volume_growth.freeze_on_topology_mismatch", true) v.SetDefault("master.volume_growth.state_loss_recovery_wait", "0s") + v.SetDefault("master.vacuum.max_parallel_per_rack", 0) + option.MaxParallelVacuumPerRack = v.GetInt("master.vacuum.max_parallel_per_rack") + if v.IsSet("master.vacuum.max_parallel_per_server") { + option.MaxParallelVacuumPerServer = v.GetInt("master.vacuum.max_parallel_per_server") + } topology.VolumeGrowStrategy.Copy1Count = v.GetUint32("master.volume_growth.copy_1") topology.VolumeGrowStrategy.Copy2Count = v.GetUint32("master.volume_growth.copy_2") @@ -230,6 +236,7 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers map[string]pb.Se ms.grpcDialOption, ms.option.GarbageThreshold, ms.option.MaxParallelVacuumPerServer, + ms.option.MaxParallelVacuumPerRack, topology.VolumeGrowStrategy.Threshold, ms.preallocateSize, time.Duration(ms.option.VacuumIntervalSeconds)*time.Second, diff --git a/weed/server/master_server_handlers_admin.go b/weed/server/master_server_handlers_admin.go index 293043d679..e6479ea49a 100644 --- a/weed/server/master_server_handlers_admin.go +++ b/weed/server/master_server_handlers_admin.go @@ -61,7 +61,7 @@ func (ms *MasterServer) volumeVacuumHandler(w http.ResponseWriter, r *http.Reque } } // glog.Infoln("garbageThreshold =", gcThreshold) - ms.Topo.Vacuum(ms.grpcDialOption, gcThreshold, ms.option.MaxParallelVacuumPerServer, 0, "", ms.preallocateSize, false, 0) + ms.Topo.Vacuum(ms.grpcDialOption, gcThreshold, ms.option.MaxParallelVacuumPerServer, ms.option.MaxParallelVacuumPerRack, 0, "", ms.preallocateSize, false, 0) ms.dirStatusHandler(w, r) } diff --git a/weed/topology/data_node.go b/weed/topology/data_node.go index 731e607091..444adf7646 100644 --- a/weed/topology/data_node.go +++ b/weed/topology/data_node.go @@ -358,10 +358,10 @@ func (dn *DataNode) matchesLocation(option *VolumeGrowOption) bool { func (dn *DataNode) GetTopology() *Topology { p := dn.Parent() - for p.Parent() != nil { + for p != nil && p.Parent() != nil { p = p.Parent() } - t := p.(*Topology) + t, _ := p.(*Topology) return t } diff --git a/weed/topology/node.go b/weed/topology/node.go index 98e212ba07..be642027e7 100644 --- a/weed/topology/node.go +++ b/weed/topology/node.go @@ -145,9 +145,12 @@ type Node interface { } type NodeImpl struct { - diskUsages *DiskUsages - id NodeId - parent Node + diskUsages *DiskUsages + id NodeId + // parent is an atomic pointer because unlinking runs under the rack's + // lock while readers (vacuum sweeps, usage propagation, String) reach + // it with no shared lock. + parent atomic.Pointer[Node] sync.RWMutex // lock children children map[NodeId]Node // maxVolumeId uses atomic ops so UpAdjustMaxVolumeId (called from the @@ -292,8 +295,8 @@ func (n *NodeImpl) IsLocked() (isTryLock bool) { } func (n *NodeImpl) String() string { - if n.parent != nil { - return n.parent.String() + ":" + string(n.id) + if parent := n.Parent(); parent != nil { + return parent.String() + ":" + string(n.id) } return string(n.id) } @@ -390,7 +393,11 @@ func (n *NodeImpl) ReleaseReservedCapacity(reservationId string) { } } func (n *NodeImpl) SetParent(node Node) { - n.parent = node + if node == nil { + n.parent.Store(nil) + return + } + n.parent.Store(&node) } func (n *NodeImpl) Children() (ret []Node) { @@ -403,7 +410,10 @@ func (n *NodeImpl) Children() (ret []Node) { } func (n *NodeImpl) Parent() Node { - return n.parent + if parent := n.parent.Load(); parent != nil { + return *parent + } + return nil } func (n *NodeImpl) GetValue() interface{} { @@ -482,8 +492,8 @@ func (n *NodeImpl) reserveOneVolumeInternal(r int64, option *VolumeGrowOption, u func (n *NodeImpl) UpAdjustDiskUsageDelta(diskType types.DiskType, diskUsage *DiskUsageCounts) { //can be negative existingDisk := n.getOrCreateDisk(diskType) existingDisk.addDiskUsageCounts(diskUsage) - if n.parent != nil { - n.parent.UpAdjustDiskUsageDelta(diskType, diskUsage) + if parent := n.Parent(); parent != nil { + parent.UpAdjustDiskUsageDelta(diskType, diskUsage) } } func (n *NodeImpl) UpAdjustMaxVolumeId(vid needle.VolumeId) { @@ -497,8 +507,8 @@ func (n *NodeImpl) UpAdjustMaxVolumeId(vid needle.VolumeId) { break } } - if n.parent != nil { - n.parent.UpAdjustMaxVolumeId(vid) + if parent := n.Parent(); parent != nil { + parent.UpAdjustMaxVolumeId(vid) } } func (n *NodeImpl) GetMaxVolumeId() needle.VolumeId { diff --git a/weed/topology/topology_event_handling.go b/weed/topology/topology_event_handling.go index a0df67e221..9dcd6f93e3 100644 --- a/weed/topology/topology_event_handling.go +++ b/weed/topology/topology_event_handling.go @@ -13,7 +13,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/storage" ) -func (t *Topology) StartRefreshWritableVolumes(grpcDialOption grpc.DialOption, garbageThreshold float64, concurrentVacuumLimitPerVolumeServer int, growThreshold float64, preallocate int64, vacuumInterval time.Duration, deleteEmptyAfter time.Duration) { +func (t *Topology) StartRefreshWritableVolumes(grpcDialOption grpc.DialOption, garbageThreshold float64, concurrentVacuumLimitPerVolumeServer int, concurrentVacuumLimitPerRack int, growThreshold float64, preallocate int64, vacuumInterval time.Duration, deleteEmptyAfter time.Duration) { if vacuumInterval <= 0 { vacuumInterval = 14 * time.Minute } @@ -39,7 +39,7 @@ func (t *Topology) StartRefreshWritableVolumes(grpcDialOption grpc.DialOption, g t.EnableVacuumByPlugin() } if !t.IsVacuumDisabled() { - t.Vacuum(grpcDialOption, garbageThreshold, concurrentVacuumLimitPerVolumeServer, 0, "", preallocate, true, deleteEmptyAfter) + t.Vacuum(grpcDialOption, garbageThreshold, concurrentVacuumLimitPerVolumeServer, concurrentVacuumLimitPerRack, 0, "", preallocate, true, deleteEmptyAfter) } } else { stats.MasterReplicaPlacementMismatch.Reset() diff --git a/weed/topology/topology_vacuum.go b/weed/topology/topology_vacuum.go index 83e1ab6f47..0cc779d0ea 100644 --- a/weed/topology/topology_vacuum.go +++ b/weed/topology/topology_vacuum.go @@ -251,7 +251,7 @@ func (t *Topology) batchVacuumVolumeCleanup(grpcDialOption grpc.DialOption, vl * } } -func (t *Topology) Vacuum(grpcDialOption grpc.DialOption, garbageThreshold float64, maxParallelVacuumPerServer int, volumeId uint32, collection string, preallocate int64, automatic bool, deleteEmptyAfter time.Duration) { +func (t *Topology) Vacuum(grpcDialOption grpc.DialOption, garbageThreshold float64, maxParallelVacuumPerServer int, maxParallelVacuumPerRack int, volumeId uint32, collection string, preallocate int64, automatic bool, deleteEmptyAfter time.Duration) { // if there is vacuum going on, return immediately swapped := atomic.CompareAndSwapInt64(&t.vacuumLockCounter, 0, 1) @@ -285,7 +285,7 @@ func (t *Topology) Vacuum(grpcDialOption grpc.DialOption, garbageThreshold float t.vacuumOneVolumeId(grpcDialOption, volumeLayout, c, garbageThreshold, locationList, vid, preallocate, false) } } else { - t.vacuumOneVolumeLayout(grpcDialOption, volumeLayout, c, garbageThreshold, maxParallelVacuumPerServer, preallocate, automatic, deleteEmptyAfter) + t.vacuumOneVolumeLayout(grpcDialOption, volumeLayout, c, garbageThreshold, maxParallelVacuumPerServer, maxParallelVacuumPerRack, preallocate, automatic, deleteEmptyAfter) } } if automatic && t.IsVacuumDisabled() { @@ -299,7 +299,7 @@ func (t *Topology) Vacuum(grpcDialOption grpc.DialOption, garbageThreshold float } } -func (t *Topology) vacuumOneVolumeLayout(grpcDialOption grpc.DialOption, volumeLayout *VolumeLayout, c *Collection, garbageThreshold float64, maxParallelVacuumPerServer int, preallocate int64, automatic bool, deleteEmptyAfter time.Duration) { +func (t *Topology) vacuumOneVolumeLayout(grpcDialOption grpc.DialOption, volumeLayout *VolumeLayout, c *Collection, garbageThreshold float64, maxParallelVacuumPerServer int, maxParallelVacuumPerRack int, preallocate int64, automatic bool, deleteEmptyAfter time.Duration) { volumeLayout.accessLock.RLock() todoVolumeMap := make(map[needle.VolumeId]*VolumeLocationList) @@ -314,14 +314,27 @@ func (t *Topology) vacuumOneVolumeLayout(grpcDialOption grpc.DialOption, volumeL t.deleteEmptyVolumes(grpcDialOption, volumeLayout, todoVolumeMap, deleteEmptyAfter) } - // limiter for each volume server + // limiter for each volume server, and for each rack when a rack cap is + // set — several volume servers can share one rack's disks and network. + // Rack keys are resolved once up front: a node that unregisters mid-sweep + // loses its parent, and a later GetRack call would panic. limiter := make(map[NodeId]int) + rackLimiter := make(map[Node]int) + dnRack := make(map[*DataNode]Node) var limiterLock sync.Mutex for _, locationList := range todoVolumeMap { for _, dn := range locationList.list { if _, ok := limiter[dn.Id()]; !ok { limiter[dn.Id()] = maxParallelVacuumPerServer } + if maxParallelVacuumPerRack > 0 { + if _, ok := dnRack[dn]; !ok { + dnRack[dn] = vacuumRackKey(dn) + } + if _, ok := rackLimiter[dnRack[dn]]; !ok { + rackLimiter[dnRack[dn]] = maxParallelVacuumPerRack + } + } } } @@ -333,38 +346,59 @@ func (t *Topology) vacuumOneVolumeLayout(grpcDialOption grpc.DialOption, volumeL for len(todoVolumeMap) > 0 { pendingVolumeMap := make(map[needle.VolumeId]*VolumeLocationList) for vid, locationList := range todoVolumeMap { - hasEnoughQuota := true + // copies on the same server or rack reserve together; a volume + // that alone exceeds a cap waits for the whole bucket to be free + // instead of pending forever + serverNeeds := make(map[NodeId]int) + rackNeeds := make(map[Node]int) for _, dn := range locationList.list { - limiterLock.Lock() - quota := limiter[dn.Id()] - limiterLock.Unlock() - if quota <= 0 { + serverNeeds[dn.Id()]++ + if maxParallelVacuumPerRack > 0 { + rackNeeds[dnRack[dn]]++ + } + } + limiterLock.Lock() + hasEnoughQuota := true + for id, need := range serverNeeds { + serverNeeds[id] = min(need, maxParallelVacuumPerServer) + if limiter[id] < serverNeeds[id] { + hasEnoughQuota = false + } + } + for id, need := range rackNeeds { + rackNeeds[id] = min(need, maxParallelVacuumPerRack) + if rackLimiter[id] < rackNeeds[id] { hasEnoughQuota = false - break } } if !hasEnoughQuota { + limiterLock.Unlock() pendingVolumeMap[vid] = locationList continue } // debit the quota - for _, dn := range locationList.list { - limiterLock.Lock() - limiter[dn.Id()]-- - limiterLock.Unlock() + for id, need := range serverNeeds { + limiter[id] -= need } + for id, need := range rackNeeds { + rackLimiter[id] -= need + } + limiterLock.Unlock() wg.Add(1) executor.Execute(func() { defer wg.Done() t.vacuumOneVolumeId(grpcDialOption, volumeLayout, c, garbageThreshold, locationList, vid, preallocate, true) // credit the quota - for _, dn := range locationList.list { - limiterLock.Lock() - limiter[dn.Id()]++ - limiterLock.Unlock() + limiterLock.Lock() + for id, need := range serverNeeds { + limiter[id] += need } + for id, need := range rackNeeds { + rackLimiter[id] += need + } + limiterLock.Unlock() select { case quotaFreed <- struct{}{}: default: @@ -390,6 +424,17 @@ func (t *Topology) vacuumOneVolumeLayout(grpcDialOption grpc.DialOption, volumeL } +// vacuumRackKey identifies a data node's rack by the rack node itself, so +// racks that share a name in different data centers still get separate +// quotas. A node that already lost its parent falls back to itself, +// degrading to the per-server cap. +func vacuumRackKey(dn *DataNode) Node { + if rack := dn.Parent(); rack != nil { + return rack + } + return dn +} + // skipReadOnly is set by the background scan and all-volumes sweep, where a // read-only flag usually means an unhealthy disk. Even then a copy that is // read-only because its disk is low on space stays eligible — compaction is diff --git a/weed/topology/topology_vacuum_test.go b/weed/topology/topology_vacuum_test.go index 82643f33b2..106011d594 100644 --- a/weed/topology/topology_vacuum_test.go +++ b/weed/topology/topology_vacuum_test.go @@ -257,13 +257,27 @@ func TestDeleteEmptyVolumesKeepsVidWhenCopyDeleteFails(t *testing.T) { type fakeVacuumServer struct { volume_server_pb.UnimplementedVolumeServerServer - mu sync.Mutex - checks map[uint32]*volume_server_pb.VacuumVolumeCheckResponse - committed []uint32 - compactErr error - hang string - entered chan string - cancelled chan struct{} + mu sync.Mutex + checks map[uint32]*volume_server_pb.VacuumVolumeCheckResponse + committed []uint32 + compactErr error + hang string + entered chan string + cancelled chan struct{} + compactGate chan struct{} + compactTracker *compactTracker +} + +type compactTracker struct { + mu sync.Mutex + inFlight int + max int +} + +func (t *compactTracker) maxInFlight() int { + t.mu.Lock() + defer t.mu.Unlock() + return t.max } func (f *fakeVacuumServer) stall(ctx context.Context, phase string) error { @@ -299,6 +313,28 @@ func (f *fakeVacuumServer) VacuumVolumeCompact(_ *volume_server_pb.VacuumVolumeC if f.hang == "compact" { return f.stall(stream.Context(), "compact") } + if f.compactGate != nil { + select { + case f.entered <- "compact": + default: + } + f.compactTracker.mu.Lock() + f.compactTracker.inFlight++ + if f.compactTracker.inFlight > f.compactTracker.max { + f.compactTracker.max = f.compactTracker.inFlight + } + f.compactTracker.mu.Unlock() + defer func() { + f.compactTracker.mu.Lock() + f.compactTracker.inFlight-- + f.compactTracker.mu.Unlock() + }() + select { + case <-f.compactGate: + case <-stream.Context().Done(): + return stream.Context().Err() + } + } return f.compactErr } @@ -456,7 +492,7 @@ func TestVacuumStalledVolumeServerReleasesGuard(t *testing.T) { done := make(chan struct{}) go func() { - topo.Vacuum(dialOption, 0.3, 1, 1, "c", 0, false, 0) + topo.Vacuum(dialOption, 0.3, 1, 0, 1, "c", 0, false, 0) close(done) }() @@ -530,3 +566,197 @@ func TestVacuumSkipsTieredReplica(t *testing.T) { t.Fatalf("committed %v, want only the local volume 2", fake.committed) // volume 2 is the positive control } } + +// A rack cap bounds copies compacting in parallel inside one rack even when +// every data node still has per-server quota left: volumes whose only copy +// sits on different data nodes of the same rack serialize. +func TestVacuumRackLimitSerializesVolumes(t *testing.T) { + tracker := &compactTracker{} + newFake := func(vid uint32) *fakeVacuumServer { + return &fakeVacuumServer{ + checks: map[uint32]*volume_server_pb.VacuumVolumeCheckResponse{vid: {GarbageRatio: 0.9}}, + entered: make(chan string, 1), + compactGate: make(chan struct{}), + compactTracker: tracker, + } + } + fake1, fake2 := newFake(1), newFake(2) + port1, dialOption := startFakeVolumeServer(t, fake1) + port2, _ := startFakeVolumeServer(t, fake2) + + topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false) + rack := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1") + dns := []*DataNode{ + rack.GetOrCreateDataNode("127.0.0.1", 8080, port1, "127.0.0.1", "dn-1", map[string]uint32{"": 10}), + rack.GetOrCreateDataNode("127.0.0.1", 8081, port2, "127.0.0.1", "dn-2", map[string]uint32{"": 10}), + } + for i, dn := range dns { + v := storage.VolumeInfo{ + Id: needle.VolumeId(i + 1), + Size: 1 << 20, + Collection: "c", + ModifiedAtSecond: time.Now().Unix(), + Version: needle.GetCurrentVersion(), + ReplicaPlacement: &super_block.ReplicaPlacement{}, + Ttl: needle.EMPTY_TTL, + } + dn.UpdateVolumes([]storage.VolumeInfo{v}) + topo.RegisterVolumeLayout(v, dn) + } + + done := make(chan struct{}) + go func() { + topo.Vacuum(dialOption, 0.3, 10, 1, 0, "c", 0, false, 0) + close(done) + }() + + deadline := time.After(15 * time.Second) + var winner *fakeVacuumServer + for winner == nil { + select { + case <-fake1.entered: + winner = fake1 + case <-fake2.entered: + winner = fake2 + case <-deadline: + t.Fatal("no volume reached the compact phase") + } + } + + // while the first compact is gated, the rack cap must keep the second + // volume from starting one + time.Sleep(300 * time.Millisecond) + if got := tracker.maxInFlight(); got != 1 { + t.Fatalf("in-flight compacts = %d, want the rack cap to hold it at 1", got) + } + + close(fake1.compactGate) + close(fake2.compactGate) + + select { + case <-done: + case <-time.After(15 * time.Second): + t.Fatal("vacuum did not finish after the gates opened") + } + for _, fake := range []*fakeVacuumServer{fake1, fake2} { + fake.mu.Lock() + n := len(fake.committed) + fake.mu.Unlock() + if n != 1 { + t.Fatalf("committed volumes = %d, want 1", n) + } + } +} + +// Racks that share a name in different data centers are independent: a per +// rack cap must not serialize volumes sitting in dc1/rack1 and dc2/rack1. +func TestVacuumRackLimitScopesRacksByDataCenter(t *testing.T) { + tracker := &compactTracker{} + newFake := func(vid uint32) *fakeVacuumServer { + return &fakeVacuumServer{ + checks: map[uint32]*volume_server_pb.VacuumVolumeCheckResponse{vid: {GarbageRatio: 0.9}}, + entered: make(chan string, 1), + compactGate: make(chan struct{}), + compactTracker: tracker, + } + } + fake1, fake2 := newFake(1), newFake(2) + port1, dialOption := startFakeVolumeServer(t, fake1) + port2, _ := startFakeVolumeServer(t, fake2) + + topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false) + dns := []*DataNode{ + topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1"). + GetOrCreateDataNode("127.0.0.1", 8080, port1, "127.0.0.1", "dn-1", map[string]uint32{"": 10}), + topo.GetOrCreateDataCenter("dc2").GetOrCreateRack("rack1"). + GetOrCreateDataNode("127.0.0.1", 8081, port2, "127.0.0.1", "dn-2", map[string]uint32{"": 10}), + } + for i, dn := range dns { + v := storage.VolumeInfo{ + Id: needle.VolumeId(i + 1), + Size: 1 << 20, + Collection: "c", + ModifiedAtSecond: time.Now().Unix(), + Version: needle.GetCurrentVersion(), + ReplicaPlacement: &super_block.ReplicaPlacement{}, + Ttl: needle.EMPTY_TTL, + } + dn.UpdateVolumes([]storage.VolumeInfo{v}) + topo.RegisterVolumeLayout(v, dn) + } + + done := make(chan struct{}) + go func() { + topo.Vacuum(dialOption, 0.3, 10, 1, 0, "c", 0, false, 0) + close(done) + }() + + // both volumes sit in different racks, so both compacts start while the + // gates stay closed + deadline := time.After(15 * time.Second) + for _, fake := range []*fakeVacuumServer{fake1, fake2} { + select { + case <-fake.entered: + case <-deadline: + t.Fatal("a volume in an independent rack never reached the compact phase") + } + } + + close(fake1.compactGate) + close(fake2.compactGate) + + select { + case <-done: + case <-time.After(15 * time.Second): + t.Fatal("vacuum did not finish after the gates opened") + } +} + +// A data node that unregisters while its volume is still compacting must not +// crash the sweep when the quota is credited back: the rack key was captured +// before the RPCs ran. +func TestVacuumRackLimitSurvivesUnregister(t *testing.T) { + fake := &fakeVacuumServer{ + checks: map[uint32]*volume_server_pb.VacuumVolumeCheckResponse{1: {GarbageRatio: 0.9}}, + entered: make(chan string, 1), + compactGate: make(chan struct{}), + compactTracker: &compactTracker{}, + } + grpcPort, dialOption := startFakeVolumeServer(t, fake) + + topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false) + dn := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1"). + GetOrCreateDataNode("127.0.0.1", 8080, grpcPort, "127.0.0.1", "dn-1", map[string]uint32{"": 10}) + v := storage.VolumeInfo{ + Id: needle.VolumeId(1), + Size: 1 << 20, + Collection: "c", + ModifiedAtSecond: time.Now().Unix(), + Version: needle.GetCurrentVersion(), + ReplicaPlacement: &super_block.ReplicaPlacement{}, + Ttl: needle.EMPTY_TTL, + } + dn.UpdateVolumes([]storage.VolumeInfo{v}) + topo.RegisterVolumeLayout(v, dn) + + done := make(chan struct{}) + go func() { + topo.Vacuum(dialOption, 0.3, 1, 1, 0, "c", 0, false, 0) + close(done) + }() + + select { + case <-fake.entered: + case <-time.After(15 * time.Second): + t.Fatal("volume never reached the compact phase") + } + + topo.UnRegisterDataNode(dn) + close(fake.compactGate) + + select { + case <-done: + case <-time.After(15 * time.Second): + t.Fatal("vacuum did not finish after the data node unregistered") + } +}