enhance(metanode):snapshot. version message from master need raft sub…

…mit and quened

Signed-off-by: leonrayang <[email protected]>

metanode/manager.go

@@ -88,6 +88,7 @@ type metadataManager struct {
 	cpuUtil              atomicutil.Float64
 	samplerDone          chan struct{}
 	volUpdating          *sync.Map //map[string]*verOp2Phase
+	verUpdateChan        chan []byte
 }
 
 func (m *metadataManager) getPacketLabels(p *Packet) (labels map[string]string) {

metanode/manager_op.go

@@ -2312,8 +2312,8 @@ func (m *metadataManager) commitCreateVersion(VolumeID string, VerSeq uint64, Op
 			if _, ok := partition.IsLeader(); !ok {
 				return true
 			}
-			log.LogInfof("action[commitCreateVersion] volume %v mp  %v do MultiVersionOp verseq %v", VolumeID, id, VerSeq)
-			if err = partition.MultiVersionOp(Op, VerSeq, nil); err != nil {
+			log.LogInfof("action[commitCreateVersion] volume %v mp  %v do HandleVersionOp verseq %v", VolumeID, id, VerSeq)
+			if err = partition.HandleVersionOp(Op, VerSeq, nil); err != nil {
 				return false
 			}
 			return true

metanode/multi_ver_test.go

@@ -1223,7 +1223,7 @@ func TestCheckVerList(t *testing.T) {
 	mockCtrl := gomock.NewController(t)
 	defer mockCtrl.Finish()
 	mp = mockPartitionRaftForTest(mockCtrl)
-
+	mp.verUpdateChan = make(chan []byte, 100)
 	mp.multiVersionList = &proto.VolVersionInfoList{
 		VerList: []*proto.VolVersionInfo{
 			{Ver: 20, Status: proto.VersionNormal},
@@ -1239,6 +1239,9 @@ func TestCheckVerList(t *testing.T) {
 	}
 
 	mp.checkVerList(masterList)
+	verData := <-mp.verUpdateChan
+	mp.submit(opFSMVersionOp, verData)
+
 	assert.True(t, mp.verSeq == 50)
 	assert.True(t, mp.multiVersionList.VerList[len(mp.multiVersionList.VerList)-1].Ver == 50)
 
@@ -1247,7 +1250,12 @@ func TestCheckVerList(t *testing.T) {
 			{Ver: 20, Status: proto.VersionNormal},
 			{Ver: 40, Status: proto.VersionNormal}},
 	}
+
 	mp.checkVerList(masterList)
+	verData = <-mp.verUpdateChan
+	mp.submit(opFSMVersionOp, verData)
+
 	assert.True(t, mp.verSeq == 40)
 	assert.True(t, len(mp.multiVersionList.VerList) == 2)
+	mp.stop()
 }

metanode/partition.go

@@ -211,7 +211,7 @@ type OpMultipart interface {
 
 // MultiVersion operation from master or client
 type OpMultiVersion interface {
-	MultiVersionOp(op uint8, verSeq uint64, verList []*proto.VolVersionInfo) (err error)
+	HandleVersionOp(op uint8, verSeq uint64, verList []*proto.VolVersionInfo) (err error)
 	fsmVersionOp(reqData []byte) (err error)
 	GetAllVersionInfo(req *proto.MultiVersionOpRequest, p *Packet) (err error)
 	GetSpecVersionInfo(req *proto.MultiVersionOpRequest, p *Packet) (err error)
@@ -504,6 +504,7 @@ type metaPartition struct {
 	verSeq                 uint64
 	multiVersionList       *proto.VolVersionInfoList
 	versionLock            sync.Mutex
+	verUpdateChan          chan []byte
 }
 
 func (mp *metaPartition) IsForbidden() bool {
@@ -649,6 +650,38 @@ func (mp *metaPartition) Stop() {
 	}
 }
 
+func (mp *metaPartition) versionInit(isCreate bool) (err error) {
+	var verList *proto.VolVersionInfoList
+	verList, err = masterClient.AdminAPI().GetVerList(mp.config.VolName)
+
+	if err != nil {
+		log.LogErrorf("action[onStart] GetVerList err[%v]", err)
+		return
+	}
+	if isCreate || len(mp.multiVersionList.VerList) == 0 {
+		for _, info := range verList.VerList {
+			if info.Status != proto.VersionNormal {
+				continue
+			}
+			mp.multiVersionList.VerList = append(mp.multiVersionList.VerList, info)
+		}
+
+		log.LogDebugf("action[onStart] verList %v", mp.multiVersionList.VerList)
+		if err = mp.storeInitMultiversion(); err != nil {
+			return
+		}
+	}
+
+	vlen := len(mp.multiVersionList.VerList)
+	if vlen > 0 {
+		mp.verSeq = mp.multiVersionList.VerList[vlen-1].Ver
+	}
+
+	go mp.runVersionOp()
+
+	return
+}
+
 func (mp *metaPartition) onStart(isCreate bool) (err error) {
 	defer func() {
 		if err == nil {
@@ -677,38 +710,17 @@ func (mp *metaPartition) onStart(isCreate bool) (err error) {
 
 	var (
 		volumeInfo *proto.SimpleVolView
-		verList    *proto.VolVersionInfoList
 	)
 	if volumeInfo, err = masterClient.AdminAPI().GetVolumeSimpleInfo(mp.config.VolName); err != nil {
 		log.LogErrorf("action[onStart] GetVolumeSimpleInfo err[%v]", err)
 		return
 	}
 
 	mp.vol.volDeleteLockTime = volumeInfo.DeleteLockTime
-	verList, err = masterClient.AdminAPI().GetVerList(mp.config.VolName)
 
-	if err != nil {
-		log.LogErrorf("action[onStart] GetVerList err[%v]", err)
+	if err = mp.versionInit(isCreate); err != nil {
 		return
 	}
-	if isCreate || len(mp.multiVersionList.VerList) == 0 {
-		for _, info := range verList.VerList {
-			if info.Status != proto.VersionNormal {
-				continue
-			}
-			mp.multiVersionList.VerList = append(mp.multiVersionList.VerList, info)
-		}
-
-		log.LogDebugf("action[onStart] verList %v", mp.multiVersionList.VerList)
-		if err = mp.storeInitMultiversion(); err != nil {
-			return
-		}
-	}
-
-	vlen := len(mp.multiVersionList.VerList)
-	if vlen > 0 {
-		mp.verSeq = mp.multiVersionList.VerList[vlen-1].Ver
-	}
 
 	mp.volType = volumeInfo.VolType
 	var ebsClient *blobstore.BlobStoreClient

metanode/partition_fsm.go

@@ -448,6 +448,19 @@ func (mp *metaPartition) Apply(command []byte, index uint64) (resp interface{},
 	return
 }
 
+func (mp *metaPartition) runVersionOp() {
+	mp.verUpdateChan = make(chan []byte, 100)
+	for {
+		select {
+		case verData := <-mp.verUpdateChan:
+			mp.submit(opFSMVersionOp, verData)
+		case <-mp.stopC:
+			log.LogWarnf("runVersionOp exit!")
+			return
+		}
+	}
+}
+
 func (mp *metaPartition) fsmVersionOp(reqData []byte) (err error) {
 	mp.multiVersionList.Lock()
 	defer mp.multiVersionList.Unlock()
@@ -462,7 +475,7 @@ func (mp *metaPartition) fsmVersionOp(reqData []byte) (err error) {
 	if opData.Op == proto.CreateVersionCommit {
 		cnt := len(mp.multiVersionList.VerList)
 		if cnt > 0 && mp.multiVersionList.VerList[cnt-1].Ver >= opData.VerSeq {
-			log.LogErrorf("action[MultiVersionOp] reqeust seq %v lessOrEqual last exist snapshot seq %v",
+			log.LogErrorf("action[HandleVersionOp] reqeust seq %v lessOrEqual last exist snapshot seq %v",
 				mp.multiVersionList.VerList[cnt-1].Ver, opData.VerSeq)
 			return
 		}

metanode/partition_op_extent.go

@@ -17,12 +17,12 @@ package metanode
 import (
 	"encoding/json"
 	"fmt"
+	"github.com/cubefs/cubefs/util/exporter"
 	"os"
 	"sort"
 
 	"github.com/cubefs/cubefs/proto"
 	"github.com/cubefs/cubefs/util/errors"
-	"github.com/cubefs/cubefs/util/exporter"
 	"github.com/cubefs/cubefs/util/log"
 )
 
@@ -247,23 +247,28 @@ func (mp *metaPartition) checkVerList(masterListInfo *proto.VolVersionInfoList)
 			lastSeq = VerList[i].Ver
 			return false
 		})
-		if err = mp.MultiVersionOp(proto.SyncAllVersionList, lastSeq, VerList); err != nil {
+		if err = mp.HandleVersionOp(proto.SyncAllVersionList, lastSeq, VerList); err != nil {
 			return
 		}
 	}
 	return
 }
 
-func (mp *metaPartition) MultiVersionOp(op uint8, verSeq uint64, verList []*proto.VolVersionInfo) (err error) {
+func (mp *metaPartition) HandleVersionOp(op uint8, verSeq uint64, verList []*proto.VolVersionInfo) (err error) {
 
 	verData := &VerOpData{
 		Op:      op,
 		VerSeq:  verSeq,
 		VerList: verList,
 	}
 	data, _ := json.Marshal(verData)
-	_, err = mp.submit(opFSMVersionOp, data)
 
+	select {
+	case mp.verUpdateChan <- data:
+		log.LogDebugf("mp %v verSeq %v op %v be pushed to queue", mp.config.PartitionId, verSeq, op)
+	default:
+		err = fmt.Errorf("mp %v version update channel full, verdata %v not be executed", mp.config.PartitionId, string(data))
+	}
 	return
 }
 

repl/repl_protocol.go

@@ -275,6 +275,7 @@ func (rp *ReplProtocol) readPkgAndPrepare() (err error) {
 	}
 	//log.LogDebugf("action[readPkgAndPrepare] packet(%v) op %v from remote(%v) conn(%v) ",
 	//	request.GetUniqueLogId(), request.Opcode, rp.sourceConn.RemoteAddr().String(), rp.sourceConn)
+
 	if err = request.resolveFollowersAddr(); err != nil {
 		err = rp.putResponse(request)
 		return