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(blobstore): waiting apply snapshot finish when save new raft log

Browse Source 
Signed-off-by: heymingwei <gongwilliam@163.com>
This commit is contained in:
heymingwei 2025-05-14 08:37:11 +00:00 committed by mingwei
parent c5389d27ad
commit ec6ca0af10
1 changed files with 19 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
blobstore/common/raftserver/server.go
View File

@ -51,8 +51,9 @@ type RaftServer interface {
// to raft storage concurrently; the application must read
// raftDone before assuming the raft messages are stable.
type apply struct {
entries []pb.Entry
snapshot pb.Snapshot
entries []pb.Entry
snapshot pb.Snapshot
snapFinishCh chan struct{}
}
type raftServer struct {
@ -329,6 +330,11 @@ func (s *raftServer) raftApply() {
s.applyEntries(entries)
s.applySnapshotFinish(snap)
s.applyWait.Trigger(s.store.Applied())
// only take effect when has snapshot
if ap.snapFinishCh != nil {
close(ap.snapFinishCh)
}
case snapMsg := <-s.snapMsgc:
go s.processSnapshotMessage(snapMsg)
case snap := <-s.snapshotC:
@ -515,6 +521,10 @@ func (s *raftServer) raftStart() {
snapshot: rd.Snapshot,
}
if !raft.IsEmptySnap(ap.snapshot) {
ap.snapFinishCh = make(chan struct{})
}
select {
case s.applyc <- ap:
case <-s.stopc:
@ -522,6 +532,13 @@ func (s *raftServer) raftStart() {
}
s.tr.Send(s.processMessages(rd.Messages))
// waiting apply snapshot is finished
if ap.snapFinishCh != nil {
log.Debugf("waiting apply snapshot finish...")
<-ap.snapFinishCh
log.Debugf("apply snapshot finish...")
}
err := s.store.Save(rd.HardState, rd.Entries)
if err != nil {
log.Panicf("save raft entries error: %v", err)