// Fill the result field, after all execution for
// other states which the sequence number is smaller,
// i.e., the sequence number of the last committed message is
// one smaller than the current message.
Fill the result field, after all execution for
other states which the sequence number is smaller,
i.e., the sequence number of the last committed message is
one smaller than the current message.
// Fill the result field after all execution for // other states which the sequence number is smaller // i . e . the sequence number of the last committed message is // one smaller than the current message .
func (node *Node) executeMsg() {
	var committedMsgs []*consensus.RequestMsg
	pairs := make(map[int64]*MsgPair)
	for {
		msgPair := <-node.MsgExecution
		pairs[msgPair.committedMsg.SequenceID] = msgPair
		committedMsgs = make([]*consensus.RequestMsg, 0)
		var isExecuted bool = false
		for _, cmsg := range node.CommittedMsgs {
			if cmsg.SequenceID == msgPair.committedMsg.SequenceID {
				isExecuted = true
		break
			}
		}
		if isExecuted {
			continue
		}
		for {
			var lastSequenceID int64 = 0
			msgTotalCnt := len(node.CommittedMsgs)
			if msgTotalCnt > 0 {
				lastCommittedMsg := node.CommittedMsgs[msgTotalCnt - 1]
				lastSequenceID = lastCommittedMsg.SequenceID
			}
			p := pairs[lastSequenceID + 1]
			if p == nil {
				break
			}
			committedMsgs = append(committedMsgs, p.committedMsg)
			LogStage("Commit", true)
			p.replyMsg.Result = "Executed"
			node.CommittedMsgs = append(node.CommittedMsgs, p.committedMsg)
			node.Broadcast(p.replyMsg, "/reply")
			LogStage("Reply", true)
			if (lastSequenceID + 1) % periodCheckPoint == 0 {
				LogStage("CHECKPOINT", false)
				for sequenceid := node.StableCheckPoint;
				    sequenceid < lastSequenceID + 1;
				    sequenceid += periodCheckPoint {
					if !node.CheckPointMissCheck(sequenceid) {
						break
					}
					checkPointMsg := node.createCheckPointMsg(sequenceid + periodCheckPoint, node.MyInfo.NodeID)
					node.Broadcast(checkPointMsg, "/checkpoint")
					node.CheckPoint(checkPointMsg)
				}
			}
			delete(pairs, lastSequenceID + 1)
		}
		for _, v := range committedMsgs {
			state, _ := node.getState(v.SequenceID) 
			digest := state.GetDigest()
			fmt.Printf("***committedMsgs[%d]: clientID=%s, operation=%s, timestamp=%d, ReqMsg (digest)=%s***\n",
			           v.SequenceID, v.ClientID, v.Operation, v.Timestamp, digest)
		}
	}
}
func ( node * Node ) executeMsg ( ) { var committedMsgs [ ] * consensus . RequestMsg pairs := make ( map [ int64 ] * MsgPair ) for { msgPair := <- node . MsgExecution pairs [ msgPair . committedMsg . SequenceID ] = msgPair committedMsgs = make ( [ ] * consensus . RequestMsg , 0 ) var isExecuted bool = false for _ , cmsg := range node . CommittedMsgs { if cmsg . SequenceID == msgPair . committedMsg . SequenceID { isExecuted = true break } } if isExecuted { continue } for { var lastSequenceID int64 = 0 msgTotalCnt := len ( node . CommittedMsgs ) if msgTotalCnt > 0 { lastCommittedMsg := node . CommittedMsgs [ msgTotalCnt - 1 ] lastSequenceID = lastCommittedMsg . SequenceID } p := pairs [ lastSequenceID + 1 ] if p == nil { break } committedMsgs = append ( committedMsgs , p . committedMsg ) LogStage ( " " , true ) p . replyMsg . Result = " " node . CommittedMsgs = append ( node . CommittedMsgs , p . committedMsg ) node . Broadcast ( p . replyMsg , " " ) LogStage ( " " , true ) if ( lastSequenceID + 1 ) % periodCheckPoint == 0 { LogStage ( " " , false ) for sequenceid := node . StableCheckPoint ; sequenceid < lastSequenceID + 1 ; sequenceid += periodCheckPoint { if ! node . CheckPointMissCheck ( sequenceid ) { break } checkPointMsg := node . createCheckPointMsg ( sequenceid + periodCheckPoint , node . MyInfo . NodeID ) node . Broadcast ( checkPointMsg , " " ) node . CheckPoint ( checkPointMsg ) } } delete ( pairs , lastSequenceID + 1 ) } for _ , v := range committedMsgs { state , _ := node . getState ( v . SequenceID ) digest := state . GetDigest ( ) fmt . Printf ( " \n " , v . SequenceID , v . ClientID , v . Operation , v . Timestamp , digest ) } } }
Fill the result field, after all execution for
other states which the sequence number is smaller,
i.e., the sequence number of the last committed message is
one smaller than the current message.
Fill the result field after all execution for other states which the sequence number is smaller i . e . the sequence number of the last committed message is one smaller than the current message .
// if msg with sequence number n is already executed, skip to send a reply of the msg with n // Execute operation for all the consecutive messages. // Find the last committed message. // Stop execution if the message for the // current sequence is not ready to execute. // Add the committed message in a private log queue // to print the orderly executed messages. // TODO: execute appropriate operation. // After executing the operation, log the // corresponding committed message to node. // Broadcast reply. // Create checkpoint every `periodCheckPoint` committed message. // Send CHECKPOINT message until it is possible. // Print all committed messages. // Must succeed.