lsm/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2025-10-21 20:07:23 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

preallocate disk space during compaction also, add cleanup for failed compaction

Browse Source
This commit is contained in:
Chris Lu
2017-08-29 23:59:53 -07:00
parent f7c22f0159
commit 58344980e4
10 changed files with 92 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
weed/topology/topology_vacuum.go
View File

@@ -6,6 +6,7 @@ import (
"net/url"
"time"
"fmt"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/storage"
"github.com/chrislusf/seaweedfs/weed/util"
@@ -37,13 +38,13 @@ func batchVacuumVolumeCheck(vl *VolumeLayout, vid storage.VolumeId, locationlist
}
return isCheckSuccess
}
func batchVacuumVolumeCompact(vl *VolumeLayout, vid storage.VolumeId, locationlist *VolumeLocationList) bool {
func batchVacuumVolumeCompact(vl *VolumeLayout, vid storage.VolumeId, locationlist *VolumeLocationList, preallocate int64) bool {
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)
if e := vacuumVolume_Compact(url, vid); e != nil {
if e := vacuumVolume_Compact(url, vid, preallocate); e != nil {
glog.V(0).Infoln(index, "Error when vacuuming", vid, "on", url, e)
ch <- false
} else {
@@ -80,7 +81,18 @@ func batchVacuumVolumeCommit(vl *VolumeLayout, vid storage.VolumeId, locationlis
}
return isCommitSuccess
}
func (t *Topology) Vacuum(garbageThreshold string) int {
func batchVacuumVolumeCleanup(vl *VolumeLayout, vid storage.VolumeId, locationlist *VolumeLocationList) {
for _, dn := range locationlist.list {
glog.V(0).Infoln("Start cleaning up", vid, "on", dn.Url())
if e := vacuumVolume_Cleanup(dn.Url(), vid); e != nil {
glog.V(0).Infoln("Error when cleaning up", vid, "on", dn.Url(), e)
} else {
glog.V(0).Infoln("Complete cleaning up", vid, "on", dn.Url())
}
}
}
func (t *Topology) Vacuum(garbageThreshold string, preallocate int64) int {
glog.V(0).Infof("Start vacuum on demand with threshold:%s", garbageThreshold)
for _, col := range t.collectionMap.Items() {
c := col.(*Collection)
@@ -99,7 +111,7 @@ func (t *Topology) Vacuum(garbageThreshold string) int {
glog.V(0).Infof("check vacuum on collection:%s volume:%d", c.Name, vid)
if batchVacuumVolumeCheck(volumeLayout, vid, locationlist, garbageThreshold) {
if batchVacuumVolumeCompact(volumeLayout, vid, locationlist) {
if batchVacuumVolumeCompact(volumeLayout, vid, locationlist, preallocate) {
batchVacuumVolumeCommit(volumeLayout, vid, locationlist)
}
}
@@ -133,9 +145,10 @@ func vacuumVolume_Check(urlLocation string, vid storage.VolumeId, garbageThresho
}
return nil, ret.Result
}
func vacuumVolume_Compact(urlLocation string, vid storage.VolumeId) error {
func vacuumVolume_Compact(urlLocation string, vid storage.VolumeId, preallocate int64) error {
values := make(url.Values)
values.Add("volume", vid.String())
values.Add("preallocate", fmt.Sprintf("%d", preallocate))
jsonBlob, err := util.Post("http://"+urlLocation+"/admin/vacuum/compact", values)
if err != nil {
return err
@@ -165,3 +178,19 @@ func vacuumVolume_Commit(urlLocation string, vid storage.VolumeId) error {
}
return nil
}
func vacuumVolume_Cleanup(urlLocation string, vid storage.VolumeId) error {
values := make(url.Values)
values.Add("volume", vid.String())
jsonBlob, err := util.Post("http://"+urlLocation+"/admin/vacuum/cleanup", values)
if err != nil {
return err
}
var ret VacuumVolumeResult
if err := json.Unmarshal(jsonBlob, &ret); err != nil {
return err
}
if ret.Error != "" {
return errors.New(ret.Error)
}
return nil
}