noway

mp2/src/raft/raft.go

@@ -70,6 +70,8 @@ type Raft struct {
 	// For Leaders Only - Metadata on Followers
 	nextIndex[]  int 		// Index of next log entry to send to server <i>
 	matchIndex[] int 		// Index of highest log entry known to be replicated on server <i>
+
+	tempConflict int
 }
 
 type ballot struct {
@@ -109,6 +111,9 @@ type AppendEntriesReply struct {
 	// Your data here (2A).
 	Term int // Client's current term if leader needs to update itself
 	Success bool // True if follower contained entries that matched PrevLogIndex and PrevLogTerm
+
+	ConflictIdx int
+	ConflictTerm int
 }
 
 
@@ -121,41 +126,36 @@ func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
 
 	// Refresh timeout
 	rf.mu.Lock()
-	
+	rf.startTime = time.Now()
 	// fmt.Printf("Vote from %d to %d \n", args.CandidateID, rf.me)
 	// fmt.Printf("voted for term %d vote term %d\n", rf.votedFor.term, args.Term)
 	// rf.printLog()
-	if rf.currentTerm > args.Term{
-		
-		reply.Term = rf.currentTerm
-		reply.VoteGranted = false
-		reply.Responded = true 
-	} else if rf.votedFor.term > args.Term{
+	if rf.currentTerm > args.Term || (rf.currentTerm == args.Term && len(rf.log) - 1 > args.LastLogIndex) || rf.votedFor.term > args.Term {
 		
 		reply.Term = rf.currentTerm
 		reply.VoteGranted = false
 		reply.Responded = true 
-	} else if ((rf.votedFor.term == args.Term && rf.votedFor.candidate == -1) ||  (rf.votedFor.term == args.Term && rf.votedFor.candidate == args.CandidateID)) && (len(rf.log) == 0 || (args.LastLogIndex >= len(rf.log) - 1 && args.LastLogTerm == rf.log[len(rf.log) - 1].Term)){
-		rf.startTime = time.Now() 
-		
+		rf.mu.Unlock()
+		return
+	} else if ( (rf.votedFor.term == args.Term && rf.votedFor.candidate == -1) || (rf.votedFor.term == args.Term && rf.votedFor.candidate == args.CandidateID)) && (len(rf.log) == 0 || (args.LastLogIndex >= len(rf.log) - 1 && args.LastLogTerm == rf.log[len(rf.log) - 1].Term)){
 		reply.Term = rf.currentTerm
 		reply.VoteGranted = true
 		reply.Responded = true
 		rf.votedFor.candidate = args.CandidateID
+		rf.state = "FOLLOWER"
 		// fmt.Print("id: " +  strconv.Itoa(rf.me) + " term: " + strconv.Itoa(rf.currentTerm))
 		// fmt.Print(" ")
 		// fmt.Print(rf.state)
 		// fmt.Print(" Vote Term: " + strconv.Itoa(rf.votedFor.term) + " Voted For: " +  strconv.Itoa(rf.votedFor.candidate))
 		// fmt.Print("\n")
 
-	} else if rf.votedFor.term < args.Term && (len(rf.log) == 0 || (args.LastLogIndex >= len(rf.log) - 1 && args.LastLogTerm == rf.log[len(rf.log) - 1].Term)){
-		rf.startTime = time.Now()
-		
+	} else if rf.votedFor.term < args.Term && (len(rf.log) == 0 || args.LastLogTerm > rf.log[len(rf.log) - 1].Term || (args.LastLogIndex >= len(rf.log) - 1 && args.LastLogTerm >= rf.log[len(rf.log) - 1].Term)){		
 		reply.Term = rf.currentTerm
 		reply.VoteGranted = true
 		reply.Responded = true
 		rf.votedFor.candidate = args.CandidateID
 		rf.votedFor.term = args.Term
+		rf.state = "FOLLOWER"
 		//print("here")
 		// fmt.Print("id: " +  strconv.Itoa(rf.me) + " term: " + strconv.Itoa(rf.currentTerm))
 		// fmt.Print(" ")
@@ -164,7 +164,6 @@ func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
 		// fmt.Print("\n")
 		
 	} else {
-		
 		reply.Term = rf.currentTerm
 		reply.VoteGranted = false
 		reply.Responded = true 
@@ -189,6 +188,7 @@ func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply
 	if rf.state == "LEADER" && args.Term > rf.currentTerm{
 		rf.state = "FOLLOWER"
 	}
+
 	if rf.currentTerm > args.Term && rf.commitIndex >= args.LeaderCommit {
 		//this might be wrong && might be wrong
 
@@ -248,6 +248,19 @@ func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply
 	} else {
 		//add else statment later to handle optimization
 		reply.Success = false
+		if args.PrevLogIndex >= len(rf.log) {
+			reply.ConflictIdx = len(rf.log) - 1
+			reply.ConflictTerm = -1
+		} else {
+			wrongTerm := rf.log[args.PrevLogIndex].Term
+			reply.ConflictTerm = wrongTerm
+			for i := 0; i < len(rf.log); i++ {
+				if rf.log[i].Term == reply.ConflictTerm{
+					reply.ConflictIdx = i
+					break
+				}
+			}
+		}
 	}
 	
 	reply.Term = rf.currentTerm
@@ -292,6 +305,7 @@ func (rf *Raft) handleCommitChannel(){
 		
 		// rf.mu.Lock()
 		// fmt.Println("Server " + strconv.Itoa(rf.me) + ", " + rf.state + ": ")
+		// fmt.Println()
 		// fmt.Print("Full Log: ")
 		// fmt.Println(rf.log)
 		// fmt.Println()
@@ -515,7 +529,7 @@ func (rf *Raft) Leader() {
 	// print(rf.me)
 	// print("\n")
 	
-	// fmt.Print(strconv.Itoa(rf.me) + "is new leader \n")
+	fmt.Print(strconv.Itoa(rf.me) + "is new leader \n")
 	heartbeat := time.Duration(100) * time.Millisecond
 	//send heartbeat to take tell others you are leader
 	rf.mu.Lock()
@@ -661,8 +675,21 @@ func (rf *Raft) sendHeartbeat(){
 								rf.commitReadyChan <- struct{}{}
 							}
 						} else {
-							if rf.nextIndex[id] > 0{
-								rf.nextIndex[id] = rf.nextIndex[id] -1
+							if resp.ConflictTerm >= 0 {
+								lastValid := -1
+								for i := len(rf.log) - 1; i >= 0; i-- {
+									if rf.log[i].Term == resp.ConflictTerm {
+										lastValid = i
+										break
+									}
+								}
+								if lastValid >= 0 {
+									rf.nextIndex[id] = lastValid + 1
+								} else {
+									rf.nextIndex[id] = resp.ConflictIdx
+								}
+							} else {
+								rf.nextIndex[id] = resp.ConflictIdx
 							}
 							 
 						}

mp2/src/raft/test_test.go

@@ -347,7 +347,6 @@ func TestRejoin2B(t *testing.T) {
 	cfg.rafts[leader1].Start(103)
 	cfg.rafts[leader1].Start(104)
 
-	
 	// new leader commits, also for index=2
 	cfg.one(103, 2, true)