Browse Source

volume: follow compactionBytePerSecond

related to https://github.com/chrislusf/seaweedfs/issues/1108
tikv
Chris Lu 5 years ago
parent
commit
d439d83772
  1. 2
      weed/command/backup.go
  2. 2
      weed/command/compact.go
  3. 3
      weed/storage/store_vacuum.go
  4. 12
      weed/storage/volume_vacuum.go
  5. 2
      weed/storage/volume_vacuum_test.go

2
weed/command/backup.go

@ -119,7 +119,7 @@ func runBackup(cmd *Command, args []string) bool {
} }
if v.SuperBlock.CompactionRevision < uint16(stats.CompactRevision) { if v.SuperBlock.CompactionRevision < uint16(stats.CompactRevision) {
if err = v.Compact2(30 * 1024 * 1024 * 1024); err != nil {
if err = v.Compact2(30*1024*1024*1024, 0); err != nil {
fmt.Printf("Compact Volume before synchronizing %v\n", err) fmt.Printf("Compact Volume before synchronizing %v\n", err)
return true return true
} }

2
weed/command/compact.go

@ -50,7 +50,7 @@ func runCompact(cmd *Command, args []string) bool {
glog.Fatalf("Compact Volume [ERROR] %s\n", err) glog.Fatalf("Compact Volume [ERROR] %s\n", err)
} }
} else { } else {
if err = v.Compact2(preallocate); err != nil {
if err = v.Compact2(preallocate, 0); err != nil {
glog.Fatalf("Compact Volume [ERROR] %s\n", err) glog.Fatalf("Compact Volume [ERROR] %s\n", err)
} }
} }

3
weed/storage/store_vacuum.go

@ -16,8 +16,7 @@ func (s *Store) CheckCompactVolume(volumeId needle.VolumeId) (float64, error) {
} }
func (s *Store) CompactVolume(vid needle.VolumeId, preallocate int64, compactionBytePerSecond int64) error { func (s *Store) CompactVolume(vid needle.VolumeId, preallocate int64, compactionBytePerSecond int64) error {
if v := s.findVolume(vid); v != nil { if v := s.findVolume(vid); v != nil {
return v.Compact2(preallocate) // compactionBytePerSecond
// return v.Compact(preallocate, compactionBytePerSecond)
return v.Compact2(preallocate, compactionBytePerSecond)
} }
return fmt.Errorf("volume id %d is not found during compact", vid) return fmt.Errorf("volume id %d is not found during compact", vid)
} }

12
weed/storage/volume_vacuum.go

@ -57,7 +57,7 @@ func (v *Volume) Compact(preallocate int64, compactionBytePerSecond int64) error
} }
// compact a volume based on deletions in .idx files // compact a volume based on deletions in .idx files
func (v *Volume) Compact2(preallocate int64) error {
func (v *Volume) Compact2(preallocate int64, compactionBytePerSecond int64) error {
if v.MemoryMapMaxSizeMb != 0 { //it makes no sense to compact in memory if v.MemoryMapMaxSizeMb != 0 { //it makes no sense to compact in memory
return nil return nil
@ -73,7 +73,7 @@ func (v *Volume) Compact2(preallocate int64) error {
v.lastCompactIndexOffset = v.IndexFileSize() v.lastCompactIndexOffset = v.IndexFileSize()
v.lastCompactRevision = v.SuperBlock.CompactionRevision v.lastCompactRevision = v.SuperBlock.CompactionRevision
glog.V(3).Infof("creating copies for volume %d ...", v.Id) glog.V(3).Infof("creating copies for volume %d ...", v.Id)
return copyDataBasedOnIndexFile(filePath+".dat", filePath+".idx", filePath+".cpd", filePath+".cpx", v.SuperBlock, v.Version(), preallocate)
return copyDataBasedOnIndexFile(filePath+".dat", filePath+".idx", filePath+".cpd", filePath+".cpx", v.SuperBlock, v.Version(), preallocate, compactionBytePerSecond)
} }
func (v *Volume) CommitCompact() error { func (v *Volume) CommitCompact() error {
@ -366,7 +366,7 @@ func (v *Volume) copyDataAndGenerateIndexFile(dstName, idxName string, prealloca
return return
} }
func copyDataBasedOnIndexFile(srcDatName, srcIdxName, dstDatName, datIdxName string, sb super_block.SuperBlock, version needle.Version, preallocate int64) (err error) {
func copyDataBasedOnIndexFile(srcDatName, srcIdxName, dstDatName, datIdxName string, sb super_block.SuperBlock, version needle.Version, preallocate int64, compactionBytePerSecond int64) (err error) {
var ( var (
srcDatBackend, dstDatBackend backend.BackendStorageFile srcDatBackend, dstDatBackend backend.BackendStorageFile
dataFile *os.File dataFile *os.File
@ -395,6 +395,8 @@ func copyDataBasedOnIndexFile(srcDatName, srcIdxName, dstDatName, datIdxName str
dstDatBackend.WriteAt(sb.Bytes(), 0) dstDatBackend.WriteAt(sb.Bytes(), 0)
newOffset := int64(sb.BlockSize()) newOffset := int64(sb.BlockSize())
writeThrottler := util.NewWriteThrottler(compactionBytePerSecond)
oldNm.AscendingVisit(func(value needle_map.NeedleValue) error { oldNm.AscendingVisit(func(value needle_map.NeedleValue) error {
offset, size := value.Offset, value.Size offset, size := value.Offset, value.Size
@ -419,7 +421,9 @@ func copyDataBasedOnIndexFile(srcDatName, srcIdxName, dstDatName, datIdxName str
if _, _, _, err = n.Append(dstDatBackend, sb.Version); err != nil { if _, _, _, err = n.Append(dstDatBackend, sb.Version); err != nil {
return fmt.Errorf("cannot append needle: %s", err) return fmt.Errorf("cannot append needle: %s", err)
} }
newOffset += n.DiskSize(version)
delta := n.DiskSize(version)
newOffset += delta
writeThrottler.MaybeSlowdown(delta)
glog.V(4).Infoln("saving key", n.Id, "volume offset", offset, "=>", newOffset, "data_size", n.Size) glog.V(4).Infoln("saving key", n.Id, "volume offset", offset, "=>", newOffset, "data_size", n.Size)
return nil return nil

2
weed/storage/volume_vacuum_test.go

@ -84,7 +84,7 @@ func TestCompaction(t *testing.T) {
} }
startTime := time.Now() startTime := time.Now()
v.Compact2(0)
v.Compact2(0, 0)
speed := float64(v.ContentSize()) / time.Now().Sub(startTime).Seconds() speed := float64(v.ContentSize()) / time.Now().Sub(startTime).Seconds()
t.Logf("compaction speed: %.2f bytes/s", speed) t.Logf("compaction speed: %.2f bytes/s", speed)

Loading…
Cancel
Save