aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--weed/storage/disk_location.go48
-rw-r--r--weed/storage/erasure_coding/ec_encoder.go4
-rw-r--r--weed/storage/erasure_coding/ec_shard.go3
-rw-r--r--weed/storage/erasure_coding/ec_volume.go8
-rw-r--r--weed/storage/needle_map_sorted_file.go1
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))
-
- 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)
- }
+ 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
+ }
- glog.V(logLevel).Infof("dir %s %s", dir, desc)
+ logLevel := glog.Level(4)
+ if l.isDiskSpaceLow {
+ logLevel = glog.Level(0)
}
- time.Sleep(time.Minute)
- }
+ 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