mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-12 06:48:45 +01:00
ec: don't index ev.Shards[0] when the last shard was unmounted (#12538)
* ec: don't index ev.Shards[0] when the last shard was unmounted * ec: skip a runtime emptied before the scrub merge reads its shards FindAllEcVolumes releases each location lock before MergeEcRuntimes runs, so an unmount can leave a merged runtime with no shards and the width scan indexed Shards[-1]. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * ec: treat a scrub merge with no shard left as a vanished volume Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * ec: read Shards once in ShardSize so a concurrent unmount cannot empty it Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * ec: read each runtime's Shards once in the scrub merge The vanished check and the slot construction read Shards separately, so an unmount between them produced a non-nil view with no shards. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * ec: anchor the scrub merge on a generation that still holds shards An unmount can empty the newest runtime while an older one keeps its shards, which left a merged view anchored on the emptied generation with no slots and the holder reported as skipped. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Mansur Akhmadi <10739460+mansur-akhmadi@users.noreply.github.com> Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com> Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
4 files changed
+152
-10
No files matched your search
@@ -624,8 +624,8 @@ func (ev *EcVolume) EcIndexBaseFileName() string {
|
||||
}
|
||||
|
||||
func (ev *EcVolume) ShardSize() uint64 {
|
||||
if len(ev.Shards) > 0 {
|
||||
return uint64(ev.Shards[0].Size())
|
||||
if shards := ev.Shards; len(shards) > 0 {
|
||||
return uint64(shards[0].Size())
|
||||
}
|
||||
return 0
|
||||
}
|
||||
@@ -1010,11 +1010,13 @@ func (ev *EcVolume) LocateEcShardNeedle(needleId types.NeedleId, version needle.
|
||||
}
|
||||
|
||||
intervals = ev.LocateEcShardNeedleInterval(version, offset.ToActualOffset(), size)
|
||||
if len(intervals) == 0 && !size.IsDeleted() {
|
||||
return types.Offset{}, 0, nil, fmt.Errorf("ec volume %d has no mounted shard to locate needle %s", ev.VolumeId, needleId)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func (ev *EcVolume) LocateEcShardNeedleInterval(version needle.Version, offset int64, size types.Size) (intervals []Interval) {
|
||||
shard := ev.Shards[0]
|
||||
var shardSize int64
|
||||
if ev.datFileSize > 0 {
|
||||
// Use datFileSize to calculate the shardSize to match the EC encoding logic.
|
||||
@@ -1025,7 +1027,14 @@ func (ev *EcVolume) LocateEcShardNeedleInterval(version needle.Version, offset i
|
||||
// Subtract 1 to handle the ambiguous case where ecdFileSize is an exact
|
||||
// multiple of ErasureCodingLargeBlockSize but the data is actually in small
|
||||
// blocks (e.g., datFileSize was just under DataShards*ErasureCodingLargeBlockSize).
|
||||
shardSize = shard.ecdFileSize - 1
|
||||
// A reader can still hold this EcVolume after its last shard was
|
||||
// unmounted, so the slice may be empty. Return no intervals then;
|
||||
// LocateEcShardNeedle reports it as an error.
|
||||
shards := ev.Shards
|
||||
if len(shards) == 0 {
|
||||
return nil
|
||||
}
|
||||
shardSize = shards[0].ecdFileSize - 1
|
||||
}
|
||||
// calculate the locations in the ec shards
|
||||
intervals = LocateData(ev.ECContext.LargeBlockSize(), ev.ECContext.SmallBlockSize(), shardSize, offset, types.Size(needle.GetActualSize(size, version)))
|
||||
|
||||
@@ -35,14 +35,22 @@ type MergedEcRuntimes struct {
|
||||
}
|
||||
|
||||
// MergeEcRuntimes resolves runtimes (one vid's mounts, in location order) into
|
||||
// a merged view. Returns nil for an empty slice (the vanished-volume case).
|
||||
// a merged view. Returns nil when no runtime holds a shard (the vanished-volume
|
||||
// case). Each runtime's Shards is read once, and a runtime emptied by a
|
||||
// concurrent unmount neither anchors the generation nor is reported skipped.
|
||||
func MergeEcRuntimes(runtimes []*EcVolume) *MergedEcRuntimes {
|
||||
if len(runtimes) == 0 {
|
||||
held := make(map[*EcVolume][]*EcVolumeShard, len(runtimes))
|
||||
for _, v := range runtimes {
|
||||
if shards := v.Shards; len(shards) > 0 {
|
||||
held[v] = shards
|
||||
}
|
||||
}
|
||||
if len(held) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
anchorGen := int64(0)
|
||||
for _, v := range runtimes {
|
||||
for v := range held {
|
||||
if v.EncodeTsNs > anchorGen {
|
||||
anchorGen = v.EncodeTsNs
|
||||
}
|
||||
@@ -57,7 +65,7 @@ func MergeEcRuntimes(runtimes []*EcVolume) *MergedEcRuntimes {
|
||||
|
||||
anchor := genMatches[0]
|
||||
for _, v := range genMatches {
|
||||
if len(v.Shards) > 0 {
|
||||
if len(held[v]) > 0 {
|
||||
anchor = v
|
||||
break
|
||||
}
|
||||
@@ -73,13 +81,17 @@ func MergeEcRuntimes(runtimes []*EcVolume) *MergedEcRuntimes {
|
||||
total := ecDataShards(anchor) + ecParityShards(anchor)
|
||||
width := total
|
||||
for _, v := range merged {
|
||||
if n := int(v.Shards[len(v.Shards)-1].ShardId) + 1; n > width {
|
||||
shards := held[v]
|
||||
if len(shards) == 0 {
|
||||
continue
|
||||
}
|
||||
if n := int(shards[len(shards)-1].ShardId) + 1; n > width {
|
||||
width = n
|
||||
}
|
||||
}
|
||||
slots := make([]*EcVolumeShard, width)
|
||||
for _, v := range merged {
|
||||
for _, shard := range v.Shards {
|
||||
for _, shard := range held[v] {
|
||||
id := int(shard.ShardId)
|
||||
if id >= len(slots) {
|
||||
continue
|
||||
@@ -92,6 +104,9 @@ func MergeEcRuntimes(runtimes []*EcVolume) *MergedEcRuntimes {
|
||||
|
||||
var skipped []string
|
||||
for pos, v := range runtimes {
|
||||
if len(held[v]) == 0 {
|
||||
continue
|
||||
}
|
||||
if v.EncodeTsNs != anchorGen {
|
||||
skipped = append(skipped, fmt.Sprintf(
|
||||
"EC volume %d shards at %s (position %d) belong to encode run %d but the scrub anchors on %d; they were not verified",
|
||||
|
||||
@@ -0,0 +1,55 @@
|
||||
package erasure_coding
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
|
||||
)
|
||||
|
||||
// FindAllEcVolumes returns runtimes whose last shard can be unmounted before
|
||||
// the merge reads them.
|
||||
func TestMergeEcRuntimesWithEmptiedRuntime(t *testing.T) {
|
||||
newRuntime := func(shardIds ...ShardId) *EcVolume {
|
||||
ev := &EcVolume{
|
||||
VolumeId: needle.VolumeId(7),
|
||||
ECContext: NewDefaultECContext("", 7),
|
||||
EncodeTsNs: 100,
|
||||
}
|
||||
for _, id := range shardIds {
|
||||
ev.Shards = append(ev.Shards, &EcVolumeShard{VolumeId: 7, ShardId: id})
|
||||
}
|
||||
return ev
|
||||
}
|
||||
emptied := newRuntime()
|
||||
holder := newRuntime(2, 5)
|
||||
|
||||
merged := MergeEcRuntimes([]*EcVolume{emptied, holder})
|
||||
require.NotNil(t, merged)
|
||||
assert.Same(t, holder, merged.Anchor)
|
||||
assert.Len(t, merged.Merged, 2)
|
||||
assert.Same(t, holder.Shards[0], merged.Slots[2])
|
||||
assert.Same(t, holder.Shards[1], merged.Slots[5])
|
||||
}
|
||||
|
||||
func TestMergeEcRuntimesAllEmptiedIsVanished(t *testing.T) {
|
||||
emptied := func() *EcVolume {
|
||||
return &EcVolume{VolumeId: 7, ECContext: NewDefaultECContext("", 7), EncodeTsNs: 100}
|
||||
}
|
||||
assert.Nil(t, MergeEcRuntimes([]*EcVolume{emptied(), emptied()}))
|
||||
}
|
||||
|
||||
func TestMergeEcRuntimesEmptiedNewerGenerationDoesNotAnchor(t *testing.T) {
|
||||
newer := &EcVolume{VolumeId: 7, ECContext: NewDefaultECContext("", 7), EncodeTsNs: 200}
|
||||
older := &EcVolume{VolumeId: 7, ECContext: NewDefaultECContext("", 7), EncodeTsNs: 100}
|
||||
older.Shards = []*EcVolumeShard{{VolumeId: 7, ShardId: 3}}
|
||||
|
||||
merged := MergeEcRuntimes([]*EcVolume{newer, older})
|
||||
require.NotNil(t, merged)
|
||||
assert.Same(t, older, merged.Anchor)
|
||||
assert.Equal(t, []*EcVolume{older}, merged.Merged)
|
||||
assert.Same(t, older.Shards[0], merged.Slots[3])
|
||||
assert.Empty(t, merged.Skipped)
|
||||
}
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
|
||||
@@ -283,3 +284,65 @@ func TestLocateEcShardNeedleMatchesInterval(t *testing.T) {
|
||||
expectedIntervals := ev.LocateEcShardNeedleInterval(needle.GetCurrentVersion(), offset.ToActualOffset(), size)
|
||||
assert.Equal(t, expectedIntervals, intervals)
|
||||
}
|
||||
|
||||
// A reader holds the EcVolume it got from FindEcVolume after the location lock
|
||||
// is released. If the last shard is unmounted in that window (ec.balance,
|
||||
// ec.decode, volumeServer.evacuate), ev.Shards is empty when the reader locates
|
||||
// the needle; indexing it panicked the volume server.
|
||||
func TestLocateEcShardNeedleIntervalAfterLastShardUnmounted(t *testing.T) {
|
||||
const datFileSize = 10 * 1118830592
|
||||
version := needle.GetCurrentVersion()
|
||||
offset, size := int64(5_000_000_000), types.Size(1167)
|
||||
withShard := []*EcVolumeShard{{ecdFileSize: datFileSize / DataShardsCount}}
|
||||
|
||||
newVolume := func(dat int64, shards []*EcVolumeShard) *EcVolume {
|
||||
return &EcVolume{
|
||||
VolumeId: needle.VolumeId(389),
|
||||
Shards: shards,
|
||||
ECContext: NewDefaultECContext("", 389),
|
||||
datFileSize: dat,
|
||||
}
|
||||
}
|
||||
|
||||
// datFileSize from .vif does not need any mounted shard.
|
||||
want := newVolume(datFileSize, withShard).LocateEcShardNeedleInterval(version, offset, size)
|
||||
assert.NotEmpty(t, want)
|
||||
assert.Equal(t, want, newVolume(datFileSize, nil).LocateEcShardNeedleInterval(version, offset, size))
|
||||
assert.Equal(t, want, newVolume(datFileSize, []*EcVolumeShard{}).LocateEcShardNeedleInterval(version, offset, size))
|
||||
|
||||
// Legacy volumes size the shards from a mounted one; with none left
|
||||
// there is nothing to locate.
|
||||
assert.NotEmpty(t, newVolume(0, withShard).LocateEcShardNeedleInterval(version, offset, size))
|
||||
assert.Empty(t, newVolume(0, nil).LocateEcShardNeedleInterval(version, offset, size))
|
||||
assert.Empty(t, newVolume(0, []*EcVolumeShard{}).LocateEcShardNeedleInterval(version, offset, size))
|
||||
}
|
||||
|
||||
// The read and delete paths go through LocateEcShardNeedle. Without a mounted
|
||||
// shard to size a legacy volume they must get an error, not empty intervals
|
||||
// that the read path would parse as a needle header.
|
||||
func TestLocateEcShardNeedleWithoutMountedShards(t *testing.T) {
|
||||
ecxFile, err := os.OpenFile("./test_files/389.ecx", os.O_RDONLY, 0)
|
||||
require.NoError(t, err)
|
||||
defer ecxFile.Close()
|
||||
stat, err := ecxFile.Stat()
|
||||
require.NoError(t, err)
|
||||
|
||||
needleId, _ := types.ParseNeedleId("0f0edb92")
|
||||
newVolume := func(dat int64) *EcVolume {
|
||||
return &EcVolume{
|
||||
VolumeId: needle.VolumeId(389),
|
||||
ecxFile: ecxFile,
|
||||
ecxFileSize: stat.Size(),
|
||||
ECContext: NewDefaultECContext("", 389),
|
||||
datFileSize: dat,
|
||||
}
|
||||
}
|
||||
|
||||
_, _, intervals, err := newVolume(0).LocateEcShardNeedle(needleId, needle.GetCurrentVersion())
|
||||
assert.ErrorContains(t, err, "no mounted shard")
|
||||
assert.Empty(t, intervals)
|
||||
|
||||
_, _, intervals, err = newVolume(10*1118830592).LocateEcShardNeedle(needleId, needle.GetCurrentVersion())
|
||||
assert.NoError(t, err)
|
||||
assert.NotEmpty(t, intervals)
|
||||
}
|
||||
Reference in new issue
Block a user