diff options
Diffstat (limited to 'weed/storage/erasure_coding')
| -rw-r--r-- | weed/storage/erasure_coding/ec_context.go | 46 | ||||
| -rw-r--r-- | weed/storage/erasure_coding/ec_encoder.go | 113 | ||||
| -rw-r--r-- | weed/storage/erasure_coding/ec_test.go | 19 | ||||
| -rw-r--r-- | weed/storage/erasure_coding/ec_volume.go | 28 | ||||
| -rw-r--r-- | weed/storage/erasure_coding/ec_volume_info.go | 21 |
5 files changed, 175 insertions, 52 deletions
diff --git a/weed/storage/erasure_coding/ec_context.go b/weed/storage/erasure_coding/ec_context.go new file mode 100644 index 000000000..770fe41af --- /dev/null +++ b/weed/storage/erasure_coding/ec_context.go @@ -0,0 +1,46 @@ +package erasure_coding + +import ( + "fmt" + + "github.com/klauspost/reedsolomon" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" +) + +// ECContext encapsulates erasure coding parameters for encoding/decoding operations +type ECContext struct { + DataShards int + ParityShards int + Collection string + VolumeId needle.VolumeId +} + +// Total returns the total number of shards (data + parity) +func (ctx *ECContext) Total() int { + return ctx.DataShards + ctx.ParityShards +} + +// NewDefaultECContext creates a context with default 10+4 shard configuration +func NewDefaultECContext(collection string, volumeId needle.VolumeId) *ECContext { + return &ECContext{ + DataShards: DataShardsCount, + ParityShards: ParityShardsCount, + Collection: collection, + VolumeId: volumeId, + } +} + +// CreateEncoder creates a Reed-Solomon encoder for this context +func (ctx *ECContext) CreateEncoder() (reedsolomon.Encoder, error) { + return reedsolomon.New(ctx.DataShards, ctx.ParityShards) +} + +// ToExt returns the file extension for a given shard index +func (ctx *ECContext) ToExt(shardIndex int) string { + return fmt.Sprintf(".ec%02d", shardIndex) +} + +// String returns a human-readable representation of the EC configuration +func (ctx *ECContext) String() string { + return fmt.Sprintf("%d+%d (total: %d)", ctx.DataShards, ctx.ParityShards, ctx.Total()) +} diff --git a/weed/storage/erasure_coding/ec_encoder.go b/weed/storage/erasure_coding/ec_encoder.go index eeeb156e6..81ebffdcb 100644 --- a/weed/storage/erasure_coding/ec_encoder.go +++ b/weed/storage/erasure_coding/ec_encoder.go @@ -11,6 +11,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/storage/idx" "github.com/seaweedfs/seaweedfs/weed/storage/needle_map" "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/storage/volume_info" "github.com/seaweedfs/seaweedfs/weed/util" ) @@ -18,6 +19,7 @@ const ( DataShardsCount = 10 ParityShardsCount = 4 TotalShardsCount = DataShardsCount + ParityShardsCount + MaxShardCount = 32 // Maximum number of shards since ShardBits is uint32 (bits 0-31) MinTotalDisks = TotalShardsCount/ParityShardsCount + 1 ErasureCodingLargeBlockSize = 1024 * 1024 * 1024 // 1GB ErasureCodingSmallBlockSize = 1024 * 1024 // 1MB @@ -54,20 +56,53 @@ func WriteSortedFileFromIdx(baseFileName string, ext string) (e error) { return nil } -// WriteEcFiles generates .ec00 ~ .ec13 files +// WriteEcFiles generates .ec00 ~ .ec13 files using default EC context func WriteEcFiles(baseFileName string) error { - return generateEcFiles(baseFileName, 256*1024, ErasureCodingLargeBlockSize, ErasureCodingSmallBlockSize) + ctx := NewDefaultECContext("", 0) + return WriteEcFilesWithContext(baseFileName, ctx) +} + +// WriteEcFilesWithContext generates EC files using the provided context +func WriteEcFilesWithContext(baseFileName string, ctx *ECContext) error { + return generateEcFiles(baseFileName, 256*1024, ErasureCodingLargeBlockSize, ErasureCodingSmallBlockSize, ctx) } func RebuildEcFiles(baseFileName string) ([]uint32, error) { - return generateMissingEcFiles(baseFileName, 256*1024, ErasureCodingLargeBlockSize, ErasureCodingSmallBlockSize) + // Attempt to load EC config from .vif file to preserve original configuration + var ctx *ECContext + if volumeInfo, _, found, _ := volume_info.MaybeLoadVolumeInfo(baseFileName + ".vif"); found && volumeInfo.EcShardConfig != nil { + ds := int(volumeInfo.EcShardConfig.DataShards) + ps := int(volumeInfo.EcShardConfig.ParityShards) + + // Validate EC config before using it + if ds > 0 && ps > 0 && ds+ps <= MaxShardCount { + ctx = &ECContext{ + DataShards: ds, + ParityShards: ps, + } + glog.V(0).Infof("Rebuilding EC files for %s with config from .vif: %s", baseFileName, ctx.String()) + } else { + glog.Warningf("Invalid EC config in .vif for %s (data=%d, parity=%d), using default", baseFileName, ds, ps) + ctx = NewDefaultECContext("", 0) + } + } else { + glog.V(0).Infof("Rebuilding EC files for %s with default config", baseFileName) + ctx = NewDefaultECContext("", 0) + } + + return RebuildEcFilesWithContext(baseFileName, ctx) +} + +// RebuildEcFilesWithContext rebuilds missing EC files using the provided context +func RebuildEcFilesWithContext(baseFileName string, ctx *ECContext) ([]uint32, error) { + return generateMissingEcFiles(baseFileName, 256*1024, ErasureCodingLargeBlockSize, ErasureCodingSmallBlockSize, ctx) } func ToExt(ecIndex int) string { return fmt.Sprintf(".ec%02d", ecIndex) } -func generateEcFiles(baseFileName string, bufferSize int, largeBlockSize int64, smallBlockSize int64) error { +func generateEcFiles(baseFileName string, bufferSize int, largeBlockSize int64, smallBlockSize int64, ctx *ECContext) error { file, err := os.OpenFile(baseFileName+".dat", os.O_RDONLY, 0) if err != nil { return fmt.Errorf("failed to open dat file: %w", err) @@ -79,21 +114,21 @@ func generateEcFiles(baseFileName string, bufferSize int, largeBlockSize int64, return fmt.Errorf("failed to stat dat file: %w", err) } - glog.V(0).Infof("encodeDatFile %s.dat size:%d", baseFileName, fi.Size()) - err = encodeDatFile(fi.Size(), baseFileName, bufferSize, largeBlockSize, file, smallBlockSize) + glog.V(0).Infof("encodeDatFile %s.dat size:%d with EC context %s", baseFileName, fi.Size(), ctx.String()) + err = encodeDatFile(fi.Size(), baseFileName, bufferSize, largeBlockSize, file, smallBlockSize, ctx) if err != nil { return fmt.Errorf("encodeDatFile: %w", err) } return nil } -func generateMissingEcFiles(baseFileName string, bufferSize int, largeBlockSize int64, smallBlockSize int64) (generatedShardIds []uint32, err error) { +func generateMissingEcFiles(baseFileName string, bufferSize int, largeBlockSize int64, smallBlockSize int64, ctx *ECContext) (generatedShardIds []uint32, err error) { - shardHasData := make([]bool, TotalShardsCount) - inputFiles := make([]*os.File, TotalShardsCount) - outputFiles := make([]*os.File, TotalShardsCount) - for shardId := 0; shardId < TotalShardsCount; shardId++ { - shardFileName := baseFileName + ToExt(shardId) + shardHasData := make([]bool, ctx.Total()) + inputFiles := make([]*os.File, ctx.Total()) + outputFiles := make([]*os.File, ctx.Total()) + for shardId := 0; shardId < ctx.Total(); shardId++ { + shardFileName := baseFileName + ctx.ToExt(shardId) if util.FileExists(shardFileName) { shardHasData[shardId] = true inputFiles[shardId], err = os.OpenFile(shardFileName, os.O_RDONLY, 0) @@ -111,14 +146,14 @@ func generateMissingEcFiles(baseFileName string, bufferSize int, largeBlockSize } } - err = rebuildEcFiles(shardHasData, inputFiles, outputFiles) + err = rebuildEcFiles(shardHasData, inputFiles, outputFiles, ctx) if err != nil { return nil, fmt.Errorf("rebuildEcFiles: %w", err) } return } -func encodeData(file *os.File, enc reedsolomon.Encoder, startOffset, blockSize int64, buffers [][]byte, outputs []*os.File) error { +func encodeData(file *os.File, enc reedsolomon.Encoder, startOffset, blockSize int64, buffers [][]byte, outputs []*os.File, ctx *ECContext) error { bufferSize := int64(len(buffers[0])) if bufferSize == 0 { @@ -131,7 +166,7 @@ func encodeData(file *os.File, enc reedsolomon.Encoder, startOffset, blockSize i } for b := int64(0); b < batchCount; b++ { - err := encodeDataOneBatch(file, enc, startOffset+b*bufferSize, blockSize, buffers, outputs) + err := encodeDataOneBatch(file, enc, startOffset+b*bufferSize, blockSize, buffers, outputs, ctx) if err != nil { return err } @@ -140,9 +175,9 @@ func encodeData(file *os.File, enc reedsolomon.Encoder, startOffset, blockSize i return nil } -func openEcFiles(baseFileName string, forRead bool) (files []*os.File, err error) { - for i := 0; i < TotalShardsCount; i++ { - fname := baseFileName + ToExt(i) +func openEcFiles(baseFileName string, forRead bool, ctx *ECContext) (files []*os.File, err error) { + for i := 0; i < ctx.Total(); i++ { + fname := baseFileName + ctx.ToExt(i) openOption := os.O_TRUNC | os.O_CREATE | os.O_WRONLY if forRead { openOption = os.O_RDONLY @@ -164,10 +199,10 @@ func closeEcFiles(files []*os.File) { } } -func encodeDataOneBatch(file *os.File, enc reedsolomon.Encoder, startOffset, blockSize int64, buffers [][]byte, outputs []*os.File) error { +func encodeDataOneBatch(file *os.File, enc reedsolomon.Encoder, startOffset, blockSize int64, buffers [][]byte, outputs []*os.File, ctx *ECContext) error { // read data into buffers - for i := 0; i < DataShardsCount; i++ { + for i := 0; i < ctx.DataShards; i++ { n, err := file.ReadAt(buffers[i], startOffset+blockSize*int64(i)) if err != nil { if err != io.EOF { @@ -186,7 +221,7 @@ func encodeDataOneBatch(file *os.File, enc reedsolomon.Encoder, startOffset, blo return err } - for i := 0; i < TotalShardsCount; i++ { + for i := 0; i < ctx.Total(); i++ { _, err := outputs[i].Write(buffers[i]) if err != nil { return err @@ -196,53 +231,57 @@ func encodeDataOneBatch(file *os.File, enc reedsolomon.Encoder, startOffset, blo return nil } -func encodeDatFile(remainingSize int64, baseFileName string, bufferSize int, largeBlockSize int64, file *os.File, smallBlockSize int64) error { +func encodeDatFile(remainingSize int64, baseFileName string, bufferSize int, largeBlockSize int64, file *os.File, smallBlockSize int64, ctx *ECContext) error { var processedSize int64 - enc, err := reedsolomon.New(DataShardsCount, ParityShardsCount) + enc, err := ctx.CreateEncoder() if err != nil { return fmt.Errorf("failed to create encoder: %w", err) } - buffers := make([][]byte, TotalShardsCount) + buffers := make([][]byte, ctx.Total()) for i := range buffers { buffers[i] = make([]byte, bufferSize) } - outputs, err := openEcFiles(baseFileName, false) + outputs, err := openEcFiles(baseFileName, false, ctx) defer closeEcFiles(outputs) if err != nil { return fmt.Errorf("failed to open ec files %s: %v", baseFileName, err) } - for remainingSize > largeBlockSize*DataShardsCount { - err = encodeData(file, enc, processedSize, largeBlockSize, buffers, outputs) + // Pre-calculate row sizes to avoid redundant calculations in loops + largeRowSize := largeBlockSize * int64(ctx.DataShards) + smallRowSize := smallBlockSize * int64(ctx.DataShards) + + for remainingSize >= largeRowSize { + err = encodeData(file, enc, processedSize, largeBlockSize, buffers, outputs, ctx) if err != nil { return fmt.Errorf("failed to encode large chunk data: %w", err) } - remainingSize -= largeBlockSize * DataShardsCount - processedSize += largeBlockSize * DataShardsCount + remainingSize -= largeRowSize + processedSize += largeRowSize } for remainingSize > 0 { - err = encodeData(file, enc, processedSize, smallBlockSize, buffers, outputs) + err = encodeData(file, enc, processedSize, smallBlockSize, buffers, outputs, ctx) if err != nil { return fmt.Errorf("failed to encode small chunk data: %w", err) } - remainingSize -= smallBlockSize * DataShardsCount - processedSize += smallBlockSize * DataShardsCount + remainingSize -= smallRowSize + processedSize += smallRowSize } return nil } -func rebuildEcFiles(shardHasData []bool, inputFiles []*os.File, outputFiles []*os.File) error { +func rebuildEcFiles(shardHasData []bool, inputFiles []*os.File, outputFiles []*os.File, ctx *ECContext) error { - enc, err := reedsolomon.New(DataShardsCount, ParityShardsCount) + enc, err := ctx.CreateEncoder() if err != nil { return fmt.Errorf("failed to create encoder: %w", err) } - buffers := make([][]byte, TotalShardsCount) + buffers := make([][]byte, ctx.Total()) for i := range buffers { if shardHasData[i] { buffers[i] = make([]byte, ErasureCodingSmallBlockSize) @@ -254,7 +293,7 @@ func rebuildEcFiles(shardHasData []bool, inputFiles []*os.File, outputFiles []*o for { // read the input data from files - for i := 0; i < TotalShardsCount; i++ { + for i := 0; i < ctx.Total(); i++ { if shardHasData[i] { n, _ := inputFiles[i].ReadAt(buffers[i], startOffset) if n == 0 { @@ -278,7 +317,7 @@ func rebuildEcFiles(shardHasData []bool, inputFiles []*os.File, outputFiles []*o } // write the data to output files - for i := 0; i < TotalShardsCount; i++ { + for i := 0; i < ctx.Total(); i++ { if !shardHasData[i] { n, _ := outputFiles[i].WriteAt(buffers[i][:inputBufferDataSize], startOffset) if inputBufferDataSize != n { diff --git a/weed/storage/erasure_coding/ec_test.go b/weed/storage/erasure_coding/ec_test.go index b1cc9c441..cbb20832c 100644 --- a/weed/storage/erasure_coding/ec_test.go +++ b/weed/storage/erasure_coding/ec_test.go @@ -23,7 +23,10 @@ func TestEncodingDecoding(t *testing.T) { bufferSize := 50 baseFileName := "1" - err := generateEcFiles(baseFileName, bufferSize, largeBlockSize, smallBlockSize) + // Create default EC context for testing + ctx := NewDefaultECContext("", 0) + + err := generateEcFiles(baseFileName, bufferSize, largeBlockSize, smallBlockSize, ctx) if err != nil { t.Logf("generateEcFiles: %v", err) } @@ -33,16 +36,16 @@ func TestEncodingDecoding(t *testing.T) { t.Logf("WriteSortedFileFromIdx: %v", err) } - err = validateFiles(baseFileName) + err = validateFiles(baseFileName, ctx) if err != nil { t.Logf("WriteSortedFileFromIdx: %v", err) } - removeGeneratedFiles(baseFileName) + removeGeneratedFiles(baseFileName, ctx) } -func validateFiles(baseFileName string) error { +func validateFiles(baseFileName string, ctx *ECContext) error { nm, err := readNeedleMap(baseFileName) if err != nil { return fmt.Errorf("readNeedleMap: %v", err) @@ -60,7 +63,7 @@ func validateFiles(baseFileName string) error { return fmt.Errorf("failed to stat dat file: %v", err) } - ecFiles, err := openEcFiles(baseFileName, true) + ecFiles, err := openEcFiles(baseFileName, true, ctx) if err != nil { return fmt.Errorf("error opening ec files: %w", err) } @@ -184,9 +187,9 @@ func readFromFile(file *os.File, data []byte, ecFileOffset int64) (err error) { return } -func removeGeneratedFiles(baseFileName string) { - for i := 0; i < DataShardsCount+ParityShardsCount; i++ { - fname := fmt.Sprintf("%s.ec%02d", baseFileName, i) +func removeGeneratedFiles(baseFileName string, ctx *ECContext) { + for i := 0; i < ctx.Total(); i++ { + fname := baseFileName + ctx.ToExt(i) os.Remove(fname) } os.Remove(baseFileName + ".ecx") diff --git a/weed/storage/erasure_coding/ec_volume.go b/weed/storage/erasure_coding/ec_volume.go index 839428e7b..3e323163e 100644 --- a/weed/storage/erasure_coding/ec_volume.go +++ b/weed/storage/erasure_coding/ec_volume.go @@ -41,7 +41,8 @@ type EcVolume struct { ecjFileAccessLock sync.Mutex diskType types.DiskType datFileSize int64 - ExpireAtSec uint64 //ec volume destroy time, calculated from the ec volume was created + ExpireAtSec uint64 //ec volume destroy time, calculated from the ec volume was created + ECContext *ECContext // EC encoding parameters } func NewEcVolume(diskType types.DiskType, dir string, dirIdx string, collection string, vid needle.VolumeId) (ev *EcVolume, err error) { @@ -73,9 +74,32 @@ func NewEcVolume(diskType types.DiskType, dir string, dirIdx string, collection ev.Version = needle.Version(volumeInfo.Version) ev.datFileSize = volumeInfo.DatFileSize ev.ExpireAtSec = volumeInfo.ExpireAtSec + + // Initialize EC context from .vif if present; fallback to defaults + if volumeInfo.EcShardConfig != nil { + ds := int(volumeInfo.EcShardConfig.DataShards) + ps := int(volumeInfo.EcShardConfig.ParityShards) + + // Validate shard counts to prevent zero or invalid values + if ds <= 0 || ps <= 0 || ds+ps > MaxShardCount { + glog.Warningf("Invalid EC config in VolumeInfo for volume %d (data=%d, parity=%d), using defaults", vid, ds, ps) + ev.ECContext = NewDefaultECContext(collection, vid) + } else { + ev.ECContext = &ECContext{ + Collection: collection, + VolumeId: vid, + DataShards: ds, + ParityShards: ps, + } + glog.V(1).Infof("Loaded EC config from VolumeInfo for volume %d: %s", vid, ev.ECContext.String()) + } + } else { + ev.ECContext = NewDefaultECContext(collection, vid) + } } else { glog.Warningf("vif file not found,volumeId:%d, filename:%s", vid, dataBaseFileName) volume_info.SaveVolumeInfo(dataBaseFileName+".vif", &volume_server_pb.VolumeInfo{Version: uint32(ev.Version)}) + ev.ECContext = NewDefaultECContext(collection, vid) } ev.ShardLocations = make(map[ShardId][]pb.ServerAddress) @@ -260,7 +284,7 @@ func (ev *EcVolume) LocateEcShardNeedleInterval(version needle.Version, offset i if ev.datFileSize > 0 { // To get the correct LargeBlockRowsCount // use datFileSize to calculate the shardSize to match the EC encoding logic. - shardSize = ev.datFileSize / DataShardsCount + shardSize = ev.datFileSize / int64(ev.ECContext.DataShards) } // calculate the locations in the ec shards intervals = LocateData(ErasureCodingLargeBlockSize, ErasureCodingSmallBlockSize, shardSize, offset, types.Size(needle.GetActualSize(size, version))) diff --git a/weed/storage/erasure_coding/ec_volume_info.go b/weed/storage/erasure_coding/ec_volume_info.go index 53b352168..4d34ccbde 100644 --- a/weed/storage/erasure_coding/ec_volume_info.go +++ b/weed/storage/erasure_coding/ec_volume_info.go @@ -87,7 +87,7 @@ func (ecInfo *EcVolumeInfo) Minus(other *EcVolumeInfo) *EcVolumeInfo { // Copy shard sizes for remaining shards retIndex := 0 - for shardId := ShardId(0); shardId < TotalShardsCount && retIndex < len(ret.ShardSizes); shardId++ { + for shardId := ShardId(0); shardId < ShardId(MaxShardCount) && retIndex < len(ret.ShardSizes); shardId++ { if ret.ShardBits.HasShardId(shardId) { if size, exists := ecInfo.GetShardSize(shardId); exists { ret.ShardSizes[retIndex] = size @@ -119,19 +119,28 @@ func (ecInfo *EcVolumeInfo) ToVolumeEcShardInformationMessage() (ret *master_pb. type ShardBits uint32 // use bits to indicate the shard id, use 32 bits just for possible future extension func (b ShardBits) AddShardId(id ShardId) ShardBits { + if id >= MaxShardCount { + return b // Reject out-of-range shard IDs + } return b | (1 << id) } func (b ShardBits) RemoveShardId(id ShardId) ShardBits { + if id >= MaxShardCount { + return b // Reject out-of-range shard IDs + } return b &^ (1 << id) } func (b ShardBits) HasShardId(id ShardId) bool { + if id >= MaxShardCount { + return false // Out-of-range shard IDs are never present + } return b&(1<<id) > 0 } func (b ShardBits) ShardIds() (ret []ShardId) { - for i := ShardId(0); i < TotalShardsCount; i++ { + for i := ShardId(0); i < ShardId(MaxShardCount); i++ { if b.HasShardId(i) { ret = append(ret, i) } @@ -140,7 +149,7 @@ func (b ShardBits) ShardIds() (ret []ShardId) { } func (b ShardBits) ToUint32Slice() (ret []uint32) { - for i := uint32(0); i < TotalShardsCount; i++ { + for i := uint32(0); i < uint32(MaxShardCount); i++ { if b.HasShardId(ShardId(i)) { ret = append(ret, i) } @@ -164,6 +173,8 @@ func (b ShardBits) Plus(other ShardBits) ShardBits { } func (b ShardBits) MinusParityShards() ShardBits { + // Removes parity shards from the bit mask + // Assumes default 10+4 EC layout where parity shards are IDs 10-13 for i := DataShardsCount; i < TotalShardsCount; i++ { b = b.RemoveShardId(ShardId(i)) } @@ -205,7 +216,7 @@ func (b ShardBits) IndexToShardId(index int) (shardId ShardId, found bool) { } currentIndex := 0 - for i := ShardId(0); i < TotalShardsCount; i++ { + for i := ShardId(0); i < ShardId(MaxShardCount); i++ { if b.HasShardId(i) { if currentIndex == index { return i, true @@ -234,7 +245,7 @@ func (ecInfo *EcVolumeInfo) resizeShardSizes(prevShardBits ShardBits) { // Copy existing sizes to new positions based on current ShardBits if len(ecInfo.ShardSizes) > 0 { newIndex := 0 - for shardId := ShardId(0); shardId < TotalShardsCount && newIndex < expectedLength; shardId++ { + for shardId := ShardId(0); shardId < ShardId(MaxShardCount) && newIndex < expectedLength; shardId++ { if ecInfo.ShardBits.HasShardId(shardId) { // Try to find the size for this shard in the old array using previous ShardBits if oldIndex, found := prevShardBits.ShardIdToIndex(shardId); found && oldIndex < len(ecInfo.ShardSizes) { |
