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(master): avoid the Apply and Snapshot processes at the same time. #22660472

Browse Source 
Signed-off-by: Victor1319 <zengxuewei@oppo.com>
This commit is contained in:
Victor1319 2024-10-22 15:37:06 +08:00 committed by AmazingChi
parent 17b4921396
commit 46b540ac7c
2 changed files with 14 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
master/metadata_fsm.go
View File

@ -19,6 +19,7 @@ import (
"fmt"
"io"
"strconv"
"sync"
"github.com/cubefs/cubefs/depends/tiglabs/raft"
"github.com/cubefs/cubefs/depends/tiglabs/raft/proto"
@ -51,6 +52,7 @@ type MetadataFsm struct {
snapshotHandler raftApplySnapshotHandler
UserAppCmdHandler raftUserCmdApplyHandler
onSnapshot bool
raftLk sync.Mutex
}
func newMetadataFsm(store *raftstore.RocksDBStore, retainsLog uint64, rs *raft.RaftServer) (fsm *MetadataFsm) {
@ -105,6 +107,9 @@ func (mf *MetadataFsm) restoreApplied() {
// Apply implements the interface of raft.StateMachine
func (mf *MetadataFsm) Apply(command []byte, index uint64) (resp interface{}, err error) {
mf.raftLk.Lock()
defer mf.raftLk.Unlock()
log.LogWarnf("action[Apply] apply index(%v)", index)
cmd := new(RaftCmd)
if err = cmd.Unmarshal(command); err != nil {
@ -179,6 +184,9 @@ func (mf *MetadataFsm) Apply(command []byte, index uint64) (resp interface{}, er
// ApplyMemberChange implements the interface of raft.StateMachine
func (mf *MetadataFsm) ApplyMemberChange(confChange *proto.ConfChange, index uint64) (interface{}, error) {
mf.raftLk.Lock()
defer mf.raftLk.Unlock()
var err error
if mf.peerChangeHandler != nil {
err = mf.peerChangeHandler(confChange)
@ -188,6 +196,9 @@ func (mf *MetadataFsm) ApplyMemberChange(confChange *proto.ConfChange, index uin
// Snapshot implements the interface of raft.StateMachine
func (mf *MetadataFsm) Snapshot() (proto.Snapshot, error) {
mf.raftLk.Lock()
defer mf.raftLk.Unlock()
snapshot := mf.store.RocksDBSnapshot()
iterator := mf.store.Iterator(snapshot)
iterator.SeekToFirst()

3
metanode/partition_fsm.go
View File

@ -654,6 +654,9 @@ func (mp *metaPartition) fsmVersionOp(reqData []byte) (err error) {
// ApplyMemberChange apply changes to the raft member.
func (mp *metaPartition) ApplyMemberChange(confChange *raftproto.ConfChange, index uint64) (resp interface{}, err error) {
mp.nonIdempotent.Lock()
defer mp.nonIdempotent.Unlock()
defer func() {
if err == nil {
mp.uploadApplyID(index)