From 27327ab3ce6f00aea438f65bee8caa948614d0f9 Mon Sep 17 00:00:00 2001 From: xiejian Date: Tue, 18 Mar 2025 16:03:00 +0800 Subject: [PATCH] fix(shardnode): replay raft log by ReadIndex() before start serve, remove load shard asynchronous with: #1000004366 Signed-off-by: xiejian --- blobstore/shardnode/storage/disk.go | 57 ++++++++--------------- blobstore/shardnode/storage/disk_test.go | 7 +-- blobstore/shardnode/storage/shard.go | 44 +++++++++++------ blobstore/shardnode/storage/shard_sm.go | 11 +++-- blobstore/shardnode/storage/shard_test.go | 6 ++- 5 files changed, 66 insertions(+), 59 deletions(-) diff --git a/blobstore/shardnode/storage/disk.go b/blobstore/shardnode/storage/disk.go index 8f5b7d741..3d96a06f7 100644 --- a/blobstore/shardnode/storage/disk.go +++ b/blobstore/shardnode/storage/disk.go @@ -216,8 +216,6 @@ func (d *Disk) Load(ctx context.Context) error { lr := kvStore.List(ctx, dataCF, listKeyPrefix, nil, nil) defer lr.Close() - errCh := make(chan error, 1) - wg := sync.WaitGroup{} for { kg, vg, err := lr.ReadNext() if err != nil { @@ -227,10 +225,10 @@ func (d *Disk) Load(ctx context.Context) error { break } - _suid := decodeShardInfoPrefix(kg.Key()) + suid := decodeShardInfoPrefix(kg.Key()) shardInfo := &shardInfo{} if err = shardInfo.Unmarshal(vg.Value()); err != nil { - span.Warnf("suid[%d] unmarshal shard info failed, err: %v", _suid, err) + span.Warnf("suid[%d] unmarshal shard info failed, err: %v", suid, err) kg.Close() vg.Close() return err @@ -238,42 +236,27 @@ func (d *Disk) Load(ctx context.Context) error { kg.Close() vg.Close() - wg.Add(1) - go func(suid proto.Suid) { - defer wg.Done() - - shard, err := newShard(ctx, shardConfig{ - suid: suid, - diskID: d.diskInfo.DiskID, - ShardBaseConfig: &d.cfg.ShardBaseConfig, - shardInfo: *shardInfo, - store: d.store, - raftManager: d.raftManager, - addrResolver: raftConfig.TransportConfig.Resolver, - disk: d, - }) - if err != nil { - span.Warnf("suid[%d] new shard failed, err: %v", suid, err) - errCh <- errors.Info(err, "new shard failed") - } - - d.shardsMu.Lock() - d.shardsMu.shards[suid] = shard - d.shardsMu.shardCheck[suid.ShardID()] = struct{}{} - d.shardsMu.Unlock() - - shard.Start() - }(_suid) - } - go func() { - wg.Wait() - close(errCh) - }() - for err := range errCh { + shard, err := newShard(ctx, shardConfig{ + suid: suid, + diskID: d.diskInfo.DiskID, + ShardBaseConfig: &d.cfg.ShardBaseConfig, + shardInfo: *shardInfo, + store: d.store, + raftManager: d.raftManager, + addrResolver: raftConfig.TransportConfig.Resolver, + disk: d, + }) if err != nil { - span.Warnf("load disk[%d] failed, err: %s", d.diskInfo.DiskID, err.Error()) + span.Warnf("suid[%d] new shard failed, err: %v", suid, err) return err } + + d.shardsMu.Lock() + d.shardsMu.shards[suid] = shard + d.shardsMu.shardCheck[suid.ShardID()] = struct{}{} + d.shardsMu.Unlock() + + shard.Start() } span.Infof("load disk[%d] success", d.diskInfo.DiskID) diff --git a/blobstore/shardnode/storage/disk_test.go b/blobstore/shardnode/storage/disk_test.go index 9faf97b2d..69d8f8ae6 100644 --- a/blobstore/shardnode/storage/disk_test.go +++ b/blobstore/shardnode/storage/disk_test.go @@ -130,10 +130,11 @@ func TestServerDisk_Shard(t *testing.T) { require.Nil(t, err) // add with same shardID - err = disk.GetDisk().AddShard(ctx, suid, 0, *rgs[i], []clustermgr.ShardUnit{ - {DiskID: diskID, Suid: proto.EncodeSuid(suid.ShardID(), 1, 0)}, + _suid := proto.EncodeSuid(suid.ShardID(), 1, 0) + err = disk.GetDisk().AddShard(ctx, _suid, 0, *rgs[i], []clustermgr.ShardUnit{ + {DiskID: diskID, Suid: _suid}, }) - require.Nil(t, err) + require.NotNil(t, err) } require.Equal(t, shardCnt, d.GetShardCnt()) diff --git a/blobstore/shardnode/storage/shard.go b/blobstore/shardnode/storage/shard.go index 0c9b59810..6579cd498 100644 --- a/blobstore/shardnode/storage/shard.go +++ b/blobstore/shardnode/storage/shard.go @@ -18,6 +18,7 @@ import ( "context" "fmt" "sync" + "sync/atomic" "time" "github.com/cubefs/cubefs/blobstore/api/clustermgr" @@ -39,6 +40,9 @@ import ( const ( shardStatusNormal = shardStatus(1) shardStatusStopReadWrite = shardStatus(2) + + needReadIndex = 0 + noNeedReadIndex = 1 ) var errShardStopWriting = errors.New("shard stop writing") @@ -156,6 +160,9 @@ func newShard(ctx context.Context, cfg shardConfig) (s *shard, err error) { if err != nil { return } + s.shardState.readIndexFunc = func(ctx context.Context) error { + return s.raftGroup.ReadIndex(ctx) + } if len(members) == 1 { err = s.raftGroup.Campaign(ctx) @@ -194,7 +201,7 @@ func (s *shard) UpdateItem(ctx context.Context, h OpHeader, i shardnode.Item) er if err := s.checkShardOptHeader(h); err != nil { return err } - if err := s.shardState.prepRWCheck(); err != nil { + if err := s.shardState.prepRWCheck(ctx); err != nil { return convertStoppingWriteErr(err) } defer s.shardState.prepRWCheckDone() @@ -236,7 +243,7 @@ func (s *shard) GetItems(ctx context.Context, h OpHeader, keys [][]byte) (ret [] if err := s.checkShardOptHeader(h); err != nil { return nil, err } - if err := s.shardState.prepRWCheck(); err != nil { + if err := s.shardState.prepRWCheck(ctx); err != nil { return nil, convertStoppingWriteErr(err) } defer s.shardState.prepRWCheckDone() @@ -271,7 +278,7 @@ func (s *shard) ListItem(ctx context.Context, h OpHeader, prefix, marker []byte, if err := s.checkShardOptHeader(h); err != nil { return nil, nil, err } - if err := s.shardState.prepRWCheck(); err != nil { + if err := s.shardState.prepRWCheck(ctx); err != nil { return nil, nil, convertStoppingWriteErr(err) } defer s.shardState.prepRWCheckDone() @@ -301,7 +308,7 @@ func (s *shard) Insert(ctx context.Context, h OpHeader, kv *KV) error { if err := s.checkShardOptHeader(h); err != nil { return err } - if err := s.shardState.prepRWCheck(); err != nil { + if err := s.shardState.prepRWCheck(ctx); err != nil { return convertStoppingWriteErr(err) } defer s.shardState.prepRWCheckDone() @@ -328,7 +335,7 @@ func (s *shard) Update(ctx context.Context, h OpHeader, kv *KV) error { if err := s.checkShardOptHeader(h); err != nil { return err } - if err := s.shardState.prepRWCheck(); err != nil { + if err := s.shardState.prepRWCheck(ctx); err != nil { return convertStoppingWriteErr(err) } defer s.shardState.prepRWCheckDone() @@ -349,7 +356,7 @@ func (s *shard) Get(ctx context.Context, h OpHeader, key []byte) (ValGetter, err if err := s.checkShardOptHeader(h); err != nil { return nil, err } - if err := s.shardState.prepRWCheck(); err != nil { + if err := s.shardState.prepRWCheck(ctx); err != nil { return nil, convertStoppingWriteErr(err) } defer s.shardState.prepRWCheckDone() @@ -374,7 +381,7 @@ func (s *shard) Delete(ctx context.Context, h OpHeader, key []byte) error { if err := s.checkShardOptHeader(h); err != nil { return err } - if err := s.shardState.prepRWCheck(); err != nil { + if err := s.shardState.prepRWCheck(ctx); err != nil { return convertStoppingWriteErr(err) } defer s.shardState.prepRWCheckDone() @@ -396,7 +403,7 @@ func (s *shard) List(ctx context.Context, h OpHeader, prefix, marker []byte, cou if h.RouteVersion < s.GetRouteVersion() { return nil, apierr.ErrShardRouteVersionNeedUpdate } - if err := s.shardState.prepRWCheck(); err != nil { + if err := s.shardState.prepRWCheck(ctx); err != nil { return nil, convertStoppingWriteErr(err) } defer s.shardState.prepRWCheckDone() @@ -457,7 +464,7 @@ func (s *shard) CreateBlob(ctx context.Context, h OpHeader, kv *KV) (proto.Blob, if err := s.checkShardOptHeader(h); err != nil { return proto.Blob{}, err } - if err := s.shardState.prepRWCheck(); err != nil { + if err := s.shardState.prepRWCheck(ctx); err != nil { return proto.Blob{}, convertStoppingWriteErr(err) } defer s.shardState.prepRWCheckDone() @@ -490,7 +497,7 @@ func (s *shard) UpdateShardRouteVersion(version proto.RouteVersion) { } func (s *shard) Stats(ctx context.Context) (shardnode.ShardStats, error) { - if err := s.shardState.prepRWCheck(); err != nil { + if err := s.shardState.prepRWCheck(ctx); err != nil { return shardnode.ShardStats{}, convertStoppingWriteErr(err) } defer s.shardState.prepRWCheckDone() @@ -539,7 +546,7 @@ func (s *shard) Stats(ctx context.Context) (shardnode.ShardStats, error) { // we should do any memory flush job or dump worker here func (s *shard) Checkpoint(ctx context.Context) error { span := trace.SpanFromContextSafe(ctx) - if err := s.shardState.prepRWCheck(); err != nil { + if err := s.shardState.prepRWCheck(ctx); err != nil { return convertStoppingWriteErr(err) } defer s.shardState.prepRWCheckDone() @@ -573,7 +580,7 @@ func (s *shard) Checkpoint(ctx context.Context) error { } func (s *shard) UpdateShard(ctx context.Context, op proto.ShardUpdateType, node clustermgr.ShardUnit, nodeHost string) error { - if err := s.shardState.prepRWCheck(); err != nil { + if err := s.shardState.prepRWCheck(ctx); err != nil { return convertStoppingWriteErr(err) } defer s.shardState.prepRWCheckDone() @@ -625,7 +632,7 @@ func (s *shard) UpdateShard(ctx context.Context, op proto.ShardUpdateType, node } func (s *shard) TransferLeader(ctx context.Context, diskID proto.DiskID) error { - if err := s.shardState.prepRWCheck(); err != nil { + if err := s.shardState.prepRWCheck(ctx); err != nil { return convertStoppingWriteErr(err) } defer s.shardState.prepRWCheckDone() @@ -949,6 +956,9 @@ type shardState struct { lastSplitTime time.Time splitDone chan struct{} + restartLeaderReadIndex uint32 + readIndexFunc func(ctx context.Context) error + lock sync.RWMutex } @@ -991,7 +1001,13 @@ func (s *shardState) splitStopWriting() { s.splitDone = make(chan struct{}) } -func (s *shardState) prepRWCheck() error { +func (s *shardState) prepRWCheck(ctx context.Context) error { + if atomic.LoadUint32(&s.restartLeaderReadIndex) == needReadIndex { + if err := s.readIndexFunc(ctx); err != nil { + return err + } + atomic.StoreUint32(&s.restartLeaderReadIndex, noNeedReadIndex) + } s.lock.Lock() // allow writing check in the list lock arena diff --git a/blobstore/shardnode/storage/shard_sm.go b/blobstore/shardnode/storage/shard_sm.go index 1b4fb6f52..7ad733184 100644 --- a/blobstore/shardnode/storage/shard_sm.go +++ b/blobstore/shardnode/storage/shard_sm.go @@ -107,7 +107,10 @@ func (s *shardSM) LeaderChange(peerID uint64) error { s.shardInfoMu.Lock() s.shardInfoMu.leader = proto.DiskID(peerID) s.shardInfoMu.Unlock() - // todo: read index before start to serve request + + if peerID > 0 && peerID != uint64(s.disk.DiskID()) { + atomic.StoreUint32(&s.shardState.restartLeaderReadIndex, noNeedReadIndex) + } return nil } @@ -116,7 +119,7 @@ func (s *shardSM) ApplyMemberChange(cc *raft.Member, index uint64) error { span, c := trace.StartSpanFromContext(context.Background(), "") span.Debugf("suid: [%d] apply member change, member:%+v", s.suid, cc) - if err := s.shardState.prepRWCheck(); err != nil { + if err := s.shardState.prepRWCheck(ctx); err != nil { span.Warnf("shard is stop writing by delete") return nil } @@ -173,7 +176,7 @@ func (s *shardSM) ApplyMemberChange(cc *raft.Member, index uint64) error { func (s *shardSM) Snapshot() (raft.Snapshot, error) { span, ctx := trace.StartSpanFromContext(context.Background(), "snapshot") - if err := s.shardState.prepRWCheck(); err != nil { + if err := s.shardState.prepRWCheck(ctx); err != nil { if errors.Is(err, errShardStopWriting) { span.Warnf("shard is stop writing by delete") return nil, nil @@ -214,7 +217,7 @@ func (s *shardSM) ApplySnapshot(ctx context.Context, header raft.RaftSnapshotHea defer snap.Close() span.Debugf("shard[%d] suid[%d] start apply snapshot, index: %d", s.suid.ShardID(), s.suid, snap.Index()) - if err := s.shardState.prepRWCheck(); err != nil { + if err := s.shardState.prepRWCheck(ctx); err != nil { if errors.Is(err, errShardStopWriting) { span.Warnf("shard is stop writing by delete") return nil diff --git a/blobstore/shardnode/storage/shard_test.go b/blobstore/shardnode/storage/shard_test.go index 557d71265..e41414527 100644 --- a/blobstore/shardnode/storage/shard_test.go +++ b/blobstore/shardnode/storage/shard_test.go @@ -69,6 +69,7 @@ func newMockShard(tb testing.TB) (*mockShard, func()) { }, nil).AnyTimes() mockRaftGroup.EXPECT().MemberChange(A, A).Return(nil).AnyTimes() + mockRaftGroup.EXPECT().ReadIndex(A).Return(nil).AnyTimes() s, err := store.NewStore(ctx, &store.Config{ Path: dir, @@ -124,6 +125,9 @@ func newMockShard(tb testing.TB) (*mockShard, func()) { }, diskID: 1, } + shard.shardState.readIndexFunc = func(ctx context.Context) error { + return mockRaftGroup.ReadIndex(ctx) + } return &mockShard{ shard: shard, @@ -146,7 +150,7 @@ func TestServerShard_ShardSplit(t *testing.T) { go func() { for i := 0; i < 100; i++ { - err := mockShard.shard.shardState.prepRWCheck() + err := mockShard.shard.shardState.prepRWCheck(ctx) require.Nil(t, err) mockShard.shard.shardState.prepRWCheckDone() }