gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 02:00:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix(shardnode): replay raft log by ReadIndex() before start serve, remove load shard asynchronous

Browse Source 
with: #1000004366

Signed-off-by: xiejian <xiejian3@oppo.com>
This commit is contained in:
xiejian 2025-03-18 16:03:00 +08:00 committed by slasher
parent 1ca6fc0554
commit 27327ab3ce
5 changed files with 66 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

57
blobstore/shardnode/storage/disk.go
View File

@ -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)

7
blobstore/shardnode/storage/disk_test.go
View File

@ -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())

44
blobstore/shardnode/storage/shard.go
View File

@ -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

11
blobstore/shardnode/storage/shard_sm.go
View File

@ -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

6
blobstore/shardnode/storage/shard_test.go
View File

@ -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()
}