From e6a49dc5337bb73a6334647bb34a44570b3ef6d7 Mon Sep 17 00:00:00 2001 From: Nikita Mochalov Date: Thu, 10 Aug 2023 01:30:36 +0300 Subject: [PATCH] Fix resource leaks (#4737) * Fix division by zero * Fix file handle leak * Fix file handle leak * Fix file handle leak * Fix goroutine leak --- weed/storage/disk_location.go | 48 +++++++++++++---------- weed/storage/erasure_coding/ec_encoder.go | 4 ++ weed/storage/erasure_coding/ec_shard.go | 3 +- weed/storage/erasure_coding/ec_volume.go | 8 ++-- weed/storage/needle_map_sorted_file.go | 1 + 5 files changed, 40 insertions(+), 24 deletions(-) diff --git a/weed/storage/disk_location.go b/weed/storage/disk_location.go index 2ee1548a2..aa3f35eac 100644 --- a/weed/storage/disk_location.go +++ b/weed/storage/disk_location.go @@ -35,6 +35,7 @@ type DiskLocation struct { ecVolumesLock sync.RWMutex isDiskSpaceLow bool + closeCh chan struct{} } func GenerateDirUuid(dir string) (dirUuidString string, err error) { @@ -80,7 +81,17 @@ func NewDiskLocation(dir string, maxVolumeCount int32, minFreeSpace util.MinFree } location.volumes = make(map[needle.VolumeId]*Volume) location.ecVolumes = make(map[needle.VolumeId]*erasure_coding.EcVolume) - go location.CheckDiskSpace() + location.closeCh = make(chan struct{}) + go func() { + for { + select { + case <-location.closeCh: + return + case <-time.After(time.Minute): + location.CheckDiskSpace() + } + } + }() return location } @@ -384,6 +395,7 @@ func (l *DiskLocation) Close() { } l.ecVolumesLock.Unlock() + close(l.closeCh) return } @@ -420,26 +432,22 @@ func (l *DiskLocation) UnUsedSpace(volumeSizeLimit uint64) (unUsedSpace uint64) } func (l *DiskLocation) CheckDiskSpace() { - for { - if dir, e := filepath.Abs(l.Directory); e == nil { - s := stats.NewDiskStatus(dir) - stats.VolumeServerResourceGauge.WithLabelValues(l.Directory, "all").Set(float64(s.All)) - stats.VolumeServerResourceGauge.WithLabelValues(l.Directory, "used").Set(float64(s.Used)) - stats.VolumeServerResourceGauge.WithLabelValues(l.Directory, "free").Set(float64(s.Free)) + if dir, e := filepath.Abs(l.Directory); e == nil { + s := stats.NewDiskStatus(dir) + stats.VolumeServerResourceGauge.WithLabelValues(l.Directory, "all").Set(float64(s.All)) + stats.VolumeServerResourceGauge.WithLabelValues(l.Directory, "used").Set(float64(s.Used)) + stats.VolumeServerResourceGauge.WithLabelValues(l.Directory, "free").Set(float64(s.Free)) - isLow, desc := l.MinFreeSpace.IsLow(s.Free, s.PercentFree) - if isLow != l.isDiskSpaceLow { - l.isDiskSpaceLow = !l.isDiskSpaceLow - } - - logLevel := glog.Level(4) - if l.isDiskSpaceLow { - logLevel = glog.Level(0) - } - - glog.V(logLevel).Infof("dir %s %s", dir, desc) + isLow, desc := l.MinFreeSpace.IsLow(s.Free, s.PercentFree) + if isLow != l.isDiskSpaceLow { + l.isDiskSpaceLow = !l.isDiskSpaceLow } - time.Sleep(time.Minute) - } + logLevel := glog.Level(4) + if l.isDiskSpaceLow { + logLevel = glog.Level(0) + } + + glog.V(logLevel).Infof("dir %s %s", dir, desc) + } } diff --git a/weed/storage/erasure_coding/ec_encoder.go b/weed/storage/erasure_coding/ec_encoder.go index a60f76bc0..7d68de2e6 100644 --- a/weed/storage/erasure_coding/ec_encoder.go +++ b/weed/storage/erasure_coding/ec_encoder.go @@ -120,6 +120,10 @@ func generateMissingEcFiles(baseFileName string, bufferSize int, largeBlockSize func encodeData(file *os.File, enc reedsolomon.Encoder, startOffset, blockSize int64, buffers [][]byte, outputs []*os.File) error { bufferSize := int64(len(buffers[0])) + if bufferSize == 0 { + glog.Fatal("unexpected zero buffer size") + } + batchCount := blockSize / bufferSize if blockSize%bufferSize != 0 { glog.Fatalf("unexpected block size %d buffer size %d", blockSize, bufferSize) diff --git a/weed/storage/erasure_coding/ec_shard.go b/weed/storage/erasure_coding/ec_shard.go index f43280fb1..19ee17636 100644 --- a/weed/storage/erasure_coding/ec_shard.go +++ b/weed/storage/erasure_coding/ec_shard.go @@ -2,7 +2,6 @@ package erasure_coding import ( "fmt" - "github.com/seaweedfs/seaweedfs/weed/storage/types" "os" "path" "strconv" @@ -10,6 +9,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/stats" "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/types" ) type ShardId uint8 @@ -39,6 +39,7 @@ func NewEcVolumeShard(diskType types.DiskType, dirname string, collection string } ecdFi, statErr := v.ecdFile.Stat() if statErr != nil { + _ = v.ecdFile.Close() return nil, fmt.Errorf("can not stat ec volume shard %s%s: %v", baseFileName, ToExt(int(shardId)), statErr) } v.ecdFileSize = ecdFi.Size() diff --git a/weed/storage/erasure_coding/ec_volume.go b/weed/storage/erasure_coding/ec_volume.go index ddee742a8..3e48e2c2d 100644 --- a/weed/storage/erasure_coding/ec_volume.go +++ b/weed/storage/erasure_coding/ec_volume.go @@ -3,19 +3,20 @@ package erasure_coding import ( "errors" "fmt" - "github.com/seaweedfs/seaweedfs/weed/pb" - "github.com/seaweedfs/seaweedfs/weed/storage/volume_info" - "golang.org/x/exp/slices" "math" "os" "sync" "time" + "golang.org/x/exp/slices" + + "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" "github.com/seaweedfs/seaweedfs/weed/storage/idx" "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/storage/volume_info" ) var ( @@ -52,6 +53,7 @@ func NewEcVolume(diskType types.DiskType, dir string, dirIdx string, collection } ecxFi, statErr := ev.ecxFile.Stat() if statErr != nil { + _ = ev.ecxFile.Close() return nil, fmt.Errorf("can not stat ec volume index %s.ecx: %v", indexBaseFileName, statErr) } ev.ecxFileSize = ecxFi.Size() diff --git a/weed/storage/needle_map_sorted_file.go b/weed/storage/needle_map_sorted_file.go index 977237f85..0433ffa0d 100644 --- a/weed/storage/needle_map_sorted_file.go +++ b/weed/storage/needle_map_sorted_file.go @@ -35,6 +35,7 @@ func NewSortedFileNeedleMap(indexBaseFileName string, indexFile *os.File) (m *So glog.V(1).Infof("Loading %s...", indexFile.Name()) mm, indexLoadError := newNeedleMapMetricFromIndexFile(indexFile) if indexLoadError != nil { + _ = m.dbFile.Close() return nil, indexLoadError } m.mapMetric = *mm