seaweedfs/weed/topology/topology_vacuum.go

165 lines
5.7 KiB
Go
Raw Normal View History

2012-11-07 09:51:43 +00:00
package topology
import (
2018-10-15 06:12:43 +00:00
"context"
2019-02-18 20:11:52 +00:00
"google.golang.org/grpc"
2012-11-07 09:51:43 +00:00
"time"
"github.com/chrislusf/seaweedfs/weed/glog"
2018-10-15 06:12:43 +00:00
"github.com/chrislusf/seaweedfs/weed/operation"
"github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
"github.com/chrislusf/seaweedfs/weed/storage"
2012-11-07 09:51:43 +00:00
)
2019-02-18 20:11:52 +00:00
func batchVacuumVolumeCheck(grpcDialOption grpc.DialOption, vl *VolumeLayout, vid storage.VolumeId, locationlist *VolumeLocationList, garbageThreshold float64) bool {
2012-11-24 01:03:27 +00:00
ch := make(chan bool, locationlist.Length())
for index, dn := range locationlist.list {
go func(index int, url string, vid storage.VolumeId) {
2019-02-18 20:11:52 +00:00
err := operation.WithVolumeServerClient(url, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
defer cancel()
resp, err := volumeServerClient.VacuumVolumeCheck(ctx, &volume_server_pb.VacuumVolumeCheckRequest{
2018-10-15 06:12:43 +00:00
VolumdId: uint32(vid),
})
if err != nil {
ch <- false
return err
}
isNeeded := resp.GarbageRatio > garbageThreshold
ch <- isNeeded
return nil
})
if err != nil {
glog.V(0).Infof("Checking vacuuming %d on %s: %v", vid, url, err)
2012-11-24 01:03:27 +00:00
}
}(index, dn.Url(), vid)
2012-11-24 01:03:27 +00:00
}
isCheckSuccess := true
for _ = range locationlist.list {
select {
case canVacuum := <-ch:
isCheckSuccess = isCheckSuccess && canVacuum
case <-time.After(30 * time.Minute):
isCheckSuccess = false
break
}
}
return isCheckSuccess
}
2019-02-18 20:11:52 +00:00
func batchVacuumVolumeCompact(grpcDialOption grpc.DialOption, vl *VolumeLayout, vid storage.VolumeId, locationlist *VolumeLocationList, preallocate int64) bool {
2012-11-24 01:03:27 +00:00
vl.removeFromWritable(vid)
ch := make(chan bool, locationlist.Length())
for index, dn := range locationlist.list {
go func(index int, url string, vid storage.VolumeId) {
glog.V(0).Infoln(index, "Start vacuuming", vid, "on", url)
2019-02-18 20:11:52 +00:00
err := operation.WithVolumeServerClient(url, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
2018-10-15 06:12:43 +00:00
_, err := volumeServerClient.VacuumVolumeCompact(context.Background(), &volume_server_pb.VacuumVolumeCompactRequest{
VolumdId: uint32(vid),
})
return err
})
if err != nil {
glog.Errorf("Error when vacuuming %d on %s: %v", vid, url, err)
2012-11-24 01:03:27 +00:00
ch <- false
} else {
2018-10-15 06:12:43 +00:00
glog.V(0).Infof("Complete vacuuming %d on %s", vid, url)
2012-11-24 01:03:27 +00:00
ch <- true
}
}(index, dn.Url(), vid)
2012-11-24 01:03:27 +00:00
}
isVacuumSuccess := true
for _ = range locationlist.list {
select {
2017-08-30 06:11:08 +00:00
case canCommit := <-ch:
isVacuumSuccess = isVacuumSuccess && canCommit
2012-11-24 01:03:27 +00:00
case <-time.After(30 * time.Minute):
isVacuumSuccess = false
break
}
}
return isVacuumSuccess
}
2019-02-18 20:11:52 +00:00
func batchVacuumVolumeCommit(grpcDialOption grpc.DialOption, vl *VolumeLayout, vid storage.VolumeId, locationlist *VolumeLocationList) bool {
2012-11-24 01:03:27 +00:00
isCommitSuccess := true
for _, dn := range locationlist.list {
2019-02-06 13:59:15 +00:00
glog.V(0).Infoln("Start Committing vacuum", vid, "on", dn.Url())
2019-02-18 20:11:52 +00:00
err := operation.WithVolumeServerClient(dn.Url(), grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
_, err := volumeServerClient.VacuumVolumeCommit(context.Background(), &volume_server_pb.VacuumVolumeCommitRequest{
2018-10-15 06:12:43 +00:00
VolumdId: uint32(vid),
})
return err
})
if err != nil {
glog.Errorf("Error when committing vacuum %d on %s: %v", vid, dn.Url(), err)
2012-11-24 01:03:27 +00:00
isCommitSuccess = false
} else {
2019-02-06 13:59:15 +00:00
glog.V(0).Infof("Complete Committing vacuum %d on %s", vid, dn.Url())
2012-11-24 01:03:27 +00:00
}
if isCommitSuccess {
vl.SetVolumeAvailable(dn, vid)
}
2012-11-24 01:03:27 +00:00
}
return isCommitSuccess
}
2019-02-18 20:11:52 +00:00
func batchVacuumVolumeCleanup(grpcDialOption grpc.DialOption, vl *VolumeLayout, vid storage.VolumeId, locationlist *VolumeLocationList) {
for _, dn := range locationlist.list {
glog.V(0).Infoln("Start cleaning up", vid, "on", dn.Url())
2019-02-18 20:11:52 +00:00
err := operation.WithVolumeServerClient(dn.Url(), grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
_, err := volumeServerClient.VacuumVolumeCleanup(context.Background(), &volume_server_pb.VacuumVolumeCleanupRequest{
2018-10-15 06:12:43 +00:00
VolumdId: uint32(vid),
})
return err
})
if err != nil {
glog.Errorf("Error when cleaning up vacuum %d on %s: %v", vid, dn.Url(), err)
} else {
2018-10-15 06:12:43 +00:00
glog.V(0).Infof("Complete cleaning up vacuum %d on %s", vid, dn.Url())
}
}
}
2019-02-18 20:11:52 +00:00
func (t *Topology) Vacuum(grpcDialOption grpc.DialOption, garbageThreshold float64, preallocate int64) int {
2018-12-31 08:06:52 +00:00
glog.V(1).Infof("Start vacuum on demand with threshold: %f", garbageThreshold)
2016-05-30 19:30:26 +00:00
for _, col := range t.collectionMap.Items() {
c := col.(*Collection)
2016-05-30 19:30:26 +00:00
for _, vl := range c.storageType2VolumeLayout.Items() {
2013-11-12 10:21:22 +00:00
if vl != nil {
volumeLayout := vl.(*VolumeLayout)
2019-02-18 20:11:52 +00:00
vacuumOneVolumeLayout(grpcDialOption, volumeLayout, c, garbageThreshold, preallocate)
2018-10-19 03:34:43 +00:00
}
}
}
return 0
}
2019-02-18 20:11:52 +00:00
func vacuumOneVolumeLayout(grpcDialOption grpc.DialOption, volumeLayout *VolumeLayout, c *Collection, garbageThreshold float64, preallocate int64) {
2018-10-19 03:34:43 +00:00
volumeLayout.accessLock.RLock()
tmpMap := make(map[storage.VolumeId]*VolumeLocationList)
2019-01-17 01:17:19 +00:00
for vid, locationList := range volumeLayout.vid2location {
tmpMap[vid] = locationList
2018-10-19 03:34:43 +00:00
}
volumeLayout.accessLock.RUnlock()
2019-01-17 01:17:19 +00:00
for vid, locationList := range tmpMap {
2018-10-19 03:34:43 +00:00
volumeLayout.accessLock.RLock()
isReadOnly, hasValue := volumeLayout.readonlyVolumes[vid]
volumeLayout.accessLock.RUnlock()
if hasValue && isReadOnly {
continue
}
2018-12-31 08:06:52 +00:00
glog.V(2).Infof("check vacuum on collection:%s volume:%d", c.Name, vid)
2019-02-18 20:11:52 +00:00
if batchVacuumVolumeCheck(grpcDialOption, volumeLayout, vid, locationList, garbageThreshold) {
if batchVacuumVolumeCompact(grpcDialOption, volumeLayout, vid, locationList, preallocate) {
batchVacuumVolumeCommit(grpcDialOption, volumeLayout, vid, locationList)
2018-10-24 07:00:01 +00:00
} else {
2019-02-18 20:11:52 +00:00
batchVacuumVolumeCleanup(grpcDialOption, volumeLayout, vid, locationList)
2012-11-07 09:51:43 +00:00
}
}
}
}