Passing all of 3B except split

Author: Junlong Gao

entrypoint.sh

@@ -12,7 +12,10 @@ while true; do
     iter=$((iter+1))
     echo "iteration $iter"
 
-    make project1 | tee out
-    make project2 | tee out
-    make project3a | tee out
+    if ! go test ./kv/test_raftstore -run TestConfChange -v | tee output; then
+        break
+    fi
+    #make project1 | tee out
+    #make project2 | tee out
+    #make project3a | tee out
 done

kv/raftstore/peer.go

@@ -376,7 +376,7 @@ func (p *peer) sendRaftMessage(msg eraftpb.Message, trans Transport) error {
 	if toPeer == nil {
 		return fmt.Errorf("failed to lookup recipient peer %v in region %v", msg.To, p.regionId)
 	}
-	log.Debugf("%v, send raft msg %v from %v to %v", p.Tag, msg.MsgType, fromPeer, toPeer)
+	log.Infof("%v, send raft msg %v from %v to %v", p.Tag, msg.MsgType, fromPeer, toPeer)
 
 	sendMsg.FromPeer = &fromPeer
 	sendMsg.ToPeer = toPeer

kv/raftstore/peer_msg_handler.go

@@ -57,12 +57,33 @@ func (d *peerMsgHandler) HandleRaftReady() {
 	if err != nil {
 		panic(err)
 	}
-	// if this node is just joining a group with a snapshot from leader, also restore
-	// the region state including version/conf numbering
+
+	kvWB := engine_util.WriteBatch{}
 	if snapshotRet != nil {
-		d.ctx.storeMeta.regionRanges.ReplaceOrInsert(&regionItem{region: snapshotRet.Region})
-		// assert d.Region() == this one
+		// if this node is just joining a group with a snapshot from leader, also restore
+		// the region state including version/conf numbering
+		d.SetRegion(snapshotRet.Region)
+
+		d.ctx.storeMeta.regions[d.regionId] = d.Region()
+		d.ctx.storeMeta.regionRanges.ReplaceOrInsert(&regionItem{region: d.Region()})
+
+		d.peer.peerCache = map[uint64]*metapb.Peer{}
+		for _, p := range d.Region().Peers {
+			d.insertPeerCache(p)
+		}
+
+	}
+
+	// The peer can be just started after restoring snapshot, or just started but without any
+	// need of snapshot, in case system reuses peer id (which it really should not be doing).
+	// Persist the region info and set it as normal here.
+	if d.RaftGroup.Raft.IsMember(d.PeerId()) {
+		kvWB.SetMeta(meta.RegionStateKey(d.regionId), &rspb.RegionLocalState{
+			State:  rspb.PeerState_Normal,
+			Region: d.Region(),
+		})
 	}
+
 	applied := d.peerStorage.applyState.AppliedIndex
 	// send messages here:
 	for _, m := range rd.Messages {
@@ -75,8 +96,11 @@ func (d *peerMsgHandler) HandleRaftReady() {
 	}
 
 	// apply the committed entries to the state machine
-	kvWB := engine_util.WriteBatch{}
 	for _, m := range rd.CommittedEntries {
+		if m.Index <= d.peerStorage.applyState.AppliedIndex {
+			log.Infof("Node %v see stale index %v <= %v", d.PeerId(), m.Index, d.peerStorage.applyState.AppliedIndex)
+			continue
+		}
 		log.Infof("%v applying index %v, region %v", d.Tag, m.Index, d.Region())
 		if m.EntryType == pb.EntryType_EntryConfChange {
 			var cc pb.ConfChange
@@ -121,7 +145,7 @@ func (d *peerMsgHandler) HandleRaftReady() {
 					}
 				}
 				if idx < 0 {
-					panic(idx)
+					log.Panicf("%v:%v, looking for %v", d.Tag, d.Region(), cc.NodeId)
 				}
 				if cc.NodeId == d.PeerId() {
 					// Don't increase the conf number here. Otherwise when getting restarted,
@@ -152,6 +176,9 @@ func (d *peerMsgHandler) HandleRaftReady() {
 				}
 			}
 
+			// finally, advance applied state
+			d.peerStorage.applyState.AppliedIndex = m.Index
+			kvWB.SetMeta(meta.ApplyStateKey(d.Region().Id), d.peerStorage.applyState)
 		} else if len(m.Data) > 0 {
 			var r raft_cmdpb.RaftCmdRequest
 			err := r.Unmarshal(m.Data)
@@ -161,10 +188,6 @@ func (d *peerMsgHandler) HandleRaftReady() {
 
 			log.Debug("Node %v processing index %v, term %v", d.PeerId(), m.Index, m.Term)
 
-			if m.Index <= d.peerStorage.applyState.AppliedIndex {
-				log.Debug("Node %v see stale index %v <= %v", d.PeerId(), m.Index, d.peerStorage.applyState.AppliedIndex)
-				continue
-			}
 			for _, cmd := range r.Requests {
 				switch cmd.CmdType {
 				case raft_cmdpb.CmdType_Put:
@@ -422,6 +445,7 @@ func (d *peerMsgHandler) proposeRaftCommand(msg *raft_cmdpb.RaftCmdRequest, cb *
 		case raft_cmdpb.AdminCmdType_ChangePeer:
 			// There can be at most one pending peer adding/removing, so we check if this is done.
 			changeReq := msg.AdminRequest.ChangePeer
+			log.Infof("Issuing conf change request: %v", changeReq)
 			switch changeReq.ChangeType {
 			case pb.ConfChangeType_AddNode:
 				if d.RaftGroup.Raft.IsMember(changeReq.Peer.Id) {
@@ -447,7 +471,6 @@ func (d *peerMsgHandler) proposeRaftCommand(msg *raft_cmdpb.RaftCmdRequest, cb *
 				}
 				cb.Done(resp)
 			case pb.ConfChangeType_RemoveNode:
-				log.Infof("Getting remove node request: %v", changeReq)
 				if !d.RaftGroup.Raft.IsMember(changeReq.Peer.Id) {
 					// done
 					log.Infof("Done with node request :%v", changeReq)
@@ -706,7 +729,7 @@ func (d *peerMsgHandler) checkSnapshot(msg *rspb.RaftMessage) (*snap.SnapKey, er
 		}
 	}
 	if !contains {
-		log.Infof("%s %s doesn't contains peer %d, skip", d.Tag, snapRegion, peerID)
+		log.Infof("%s %s doesn't contain peer %d, skip", d.Tag, snapRegion, peerID)
 		return &key, nil
 	}
 	meta := d.ctx.storeMeta

kv/raftstore/peer_storage.go

@@ -167,8 +167,11 @@ func (ps *PeerStorage) Snapshot() (eraftpb.Snapshot, error) {
 			return snapshot, raft.ErrSnapshotTemporarilyUnavailable
 		}
 		ps.snapState.StateType = snap.SnapState_Relax
-		if snapshot.GetMetadata() != nil && ps.validateSnap(&snapshot) {
-			return snapshot, nil
+		if snapshot.GetMetadata() != nil {
+			ps.snapTriedCnt = 0
+			if ps.validateSnap(&snapshot) {
+				return snapshot, nil
+			}
 		} else {
 			log.Warnf("%s failed to try generating snapshot, times: %d", ps.Tag, ps.snapTriedCnt)
 		}
@@ -358,7 +361,6 @@ func (ps *PeerStorage) ApplySnapshot(snapshot *eraftpb.Snapshot, kvWB *engine_ut
 	}
 
 	oldRegion := ps.Region()
-	ps.region = snapData.Region
 
 	err := ps.clearMeta(raftWB, kvWB)
 	if err != nil {
@@ -367,11 +369,11 @@ func (ps *PeerStorage) ApplySnapshot(snapshot *eraftpb.Snapshot, kvWB *engine_ut
 	done := make(chan bool)
 	ps.snapState = snap.SnapState{StateType: snap.SnapState_Applying}
 	ps.regionSched <- &runner.RegionTaskApply{
-		RegionId: ps.Region().Id,
+		RegionId: snapData.Region.Id,
 		Notifier: done,
 		SnapMeta: snapshot.Metadata,
-		StartKey: ps.Region().StartKey,
-		EndKey:   ps.Region().EndKey,
+		StartKey: snapData.Region.StartKey,
+		EndKey:   snapData.Region.EndKey,
 	}
 
 	log.Infof("%v applying snapshot %v %v", ps.Tag, snapshot.Metadata.Index, snapshot.Metadata.Term)
@@ -398,11 +400,11 @@ func (ps *PeerStorage) ApplySnapshot(snapshot *eraftpb.Snapshot, kvWB *engine_ut
 	if term != snapshot.Metadata.Term {
 		panic(term)
 	}
-	log.Infof("%v after applying snapshot, got %v %v %v", ps.Tag, snapshot.Metadata.Index, term, ps.Region())
+	log.Infof("%v after applying snapshot, got %v %v %v", ps.Tag, snapshot.Metadata.Index, term, snapData.Region)
 	// ps.clearExtraData()
 	return &ApplySnapResult{
 		PrevRegion: oldRegion,
-		Region:     ps.Region(),
+		Region:     snapData.Region,
 	}, nil
 }
 

kv/test_raftstore/scheduler.go

@@ -539,7 +539,7 @@ func MustSamePeers(left *metapb.Region, right *metapb.Region) {
 	}
 	for _, p := range left.GetPeers() {
 		if FindPeer(right, p.GetStoreId()) == nil {
-			panic("not found the peer")
+			log.Infof("not found the peer l:%v vs r:%v", left, right)
 		}
 	}
 }

raft/raft.go

@@ -239,7 +239,7 @@ func (r *Raft) sendAppend(to uint64) bool {
 		if err != nil {
 			panic(err)
 		}
-		log.Debugf("Leader %v sending snapshot %v to %v", r.id, r.RaftLog.committed, to)
+		log.Debugf("Leader %v sending snapshot %v to %v", r.id, snap.Metadata, to)
 		r.send(pb.Message{
 			MsgType: pb.MessageType_MsgSnapshot,
 			To:      to,
@@ -292,12 +292,16 @@ func (r *Raft) handleAppendEntries(m pb.Message) {
 
 	r.electionElapsed = 0
 	if !r.RaftLog.CheckMatch(m.Index, m.LogTerm) {
+		// XXX use committed as a reply can be further optimized by calling r.RaftLog.Compare
+		// and get the first conflict index. cf. the dissertation.
 		r.send(pb.Message{
 			To:      m.From,
 			MsgType: pb.MessageType_MsgAppendResponse,
 			Reject:  true,
+
+			Index: r.RaftLog.committed, // hint a reply
 		})
-		log.Infof("Node %v reject append (%v,%v)", r.id, m.Index, m.Term)
+		log.Infof("Node %v reject append (%v,%v), hint %v", r.id, m.Index, m.LogTerm, r.RaftLog.committed)
 	} else {
 		if len(m.Entries) == 0 {
 			r.send(pb.Message{
@@ -362,6 +366,8 @@ func (r *Raft) tryUpdateCommitIdx(newCommitIdx uint64) bool {
 		log.Panicf("Term goes backwards: %v -> %v", term, r.Term)
 	}
 	if term != r.Term {
+		// assert term < r.Term
+		// Leader cannot use an older term entry to advance commit index. cf. Figure 8.
 		return false
 	}
 
@@ -388,10 +394,10 @@ func (r *Raft) handleAppendEntriesResp(m pb.Message) {
 	}
 
 	if m.Reject {
-		// If the response of append is duplicated (or leader sent 2 append entries get 2 same responses),
-		// don't blindly decrease the next index.
 		if r.Prs[m.From].Next > 0 {
-			r.Prs[m.From].Next = max(r.Prs[m.From].Next-1, r.Prs[m.From].Match+1)
+			// If the response of append is duplicated (or leader sent 2 append entries get 2 same responses),
+			// don't blindly decrease the next index.
+			r.Prs[m.From].Next = m.Index + 1
 		}
 		r.sendAppend(m.From)
 	} else {