Untitled
unknown
golang
2 years ago
24 kB
24
Indexable
// this version can pass all up to basic agree 2B. But cannot pass fail agree 2B
// after reconnection, it fails
//
// raft.go
// =======
// Write your code in this file
// We will use the original version of all other
// files for testing
//
package raft
//
// API
// ===
// This is an outline of the API that your raft implementation should
// expose.
//
// rf = NewPeer(...)
// Create a new Raft server.
//
// rf.PutCommand(command interface{}) (index, term, isleader)
// PutCommand agreement on a new log entry
//
// rf.GetState() (me, term, isLeader)
// Ask a Raft peer for "me", its current term, and whether it thinks it
// is a leader
//
// ApplyCommand
// Each time a new entry is committed to the log, each Raft peer
// should send an ApplyCommand to the service (e.g. tester) on the
// same server, via the applyCh channel passed to NewPeer()
//
import (
"fmt"
"io/ioutil"
"log"
"math"
"math/rand"
"os"
"sync"
"time"
"github.com/cmu440/rpc"
)
// Set to false to disable debug logs completely
// Make sure to set kEnableDebugLogs to false before submitting
const kEnableDebugLogs = false
// Set to true to log to stdout instead of file
const kLogToStdout = true
// Change this to output logs to a different directory
const kLogOutputDir = "./raftlogs/"
//
// ApplyCommand
// ========
//
// As each Raft peer becomes aware that successive log entries are
// committed, the peer should send an ApplyCommand to the service (or
// tester) on the same server, via the applyCh passed to NewPeer()
//
type ApplyCommand struct {
Index int
Command interface{}
}
type LogEntry struct {
Index int
Term int
Command interface{}
// AppendCount int
Committed bool
}
//
// Raft struct
// ===========
//
// A Go object implementing a single Raft peer
//
type Raft struct {
mux sync.Mutex // Lock to protect shared access to this peer's state
peers []*rpc.ClientEnd // RPC end points of all peers
me int // this peer's index into peers[]
// You are expected to create reasonably clear log files before asking a
// debugging question on Piazza or OH. Use of this logger is optional, and
// you are free to remove it completely.
logger *log.Logger // We provide you with a separate logger per peer.
// Your data here (2A, 2B).
// Look at the Raft paper's Figure 2 for a description of what
// state a Raft peer should maintain
role int // 1 for leader, 2 for follower, and 3 for candidate
currentTerm int
votedFor int // -1 if not yet voted
receivedVoteCount int // receive how many grants from all servers
becomesLeader chan bool
othersElected chan bool
log []LogEntry // list or slice or other data structures?
commitIndex int
lastApplied int
nextIndex []int // ?? leader maintains a nextIndex for each follower, which is the index of the next log entry the leader will send to the follower
matchIndex []int // ??
closeRoutine chan bool
votedResetTicker chan bool
applyCh chan ApplyCommand
}
//
// GetState()
// ==========
//
// Return "me", current term and whether this peer
// believes it is the leader
//
func (rf *Raft) GetState() (int, int, bool) {
var me int
var term int
var isleader bool
// Your code here (2A)
// fmt.Println("In GetState!")
rf.mux.Lock() // deadlock here!
me = rf.me
term = rf.currentTerm
if rf.role == 1 {
isleader = true
} else {
isleader = false
}
rf.mux.Unlock()
// fmt.Println("GetState returns! me = ", me, ", term = ", term, ", isleader = ", isleader)
return me, term, isleader
}
//
// RequestVoteArgs
// ===============
//
// Example RequestVote RPC arguments structure
//
// Please note
// ===========
// Field names must start with capital letters!
//
type RequestVoteArgs struct {
// Your data here (2A, 2B)
Term int
CandidateId int
LastLogIndex int
LastLogTerm int
}
//
// RequestVoteReply
// ================
//
// Example RequestVote RPC reply structure.
//
// Please note
// ===========
// Field names must start with capital letters!
//
//
type RequestVoteReply struct {
// Your data here (2A)
Term int
VoteGranted bool
}
//
// RequestVote
// ===========
//
// Example RequestVote RPC handler
//
func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) { // followers handles an incoming RPC
// Your code here (2A, 2B)
rf.mux.Lock()
if args.Term < rf.currentTerm {
reply.Term = rf.currentTerm
// fmt.Println("1: after reply term = ", reply.Term)
reply.VoteGranted = false
rf.mux.Unlock()
return
} else {
if (args.Term > rf.currentTerm) { // MODIFIED!!! ERROR means this follower needs to be updated (not in the current term yet)
rf.votedFor = -1
// fmt.Println("case 1: args.Term = ", args.Term, ", rf.currentTerm = ", rf.currentTerm, ", ", rf.me, " changes back to follower")
rf.currentTerm = args.Term
rf.role = 2
rf.mux.Unlock()
} else {
rf.mux.Unlock()
}
rf.mux.Lock()
// TODO: see whether the candidate's log is up-to-date
upToDate := false
if args.LastLogTerm > rf.log[len(rf.log)-1].Term {
upToDate = true
}
if (args.LastLogTerm == rf.log[len(rf.log)-1].Term) && args.LastLogIndex >= rf.log[len(rf.log)-1].Index {
upToDate = true
}
if (rf.votedFor == -1 || rf.votedFor == args.CandidateId) && upToDate {
// fmt.Println(rf.me, " votes true in the leader election === , it votes for ", args.CandidateId)
reply.Term = args.Term
// fmt.Println("2: after reply term = ", reply.Term)
reply.VoteGranted = true
rf.votedFor = args.CandidateId
rf.role = 2
rf.mux.Unlock()
rf.votedResetTicker <- true
} else {
reply.Term = args.Term
// fmt.Println("3: after reply term = ", reply.Term)
// reply.Term = rf.currentTerm // ???????????
reply.VoteGranted = false
rf.mux.Unlock()
return
}
}
}
// sendRequestVote
// ===============
//
// Example code to send a RequestVote RPC to a server
//
// server int -- index of the target server in
// rf.peers[]
//
// args *RequestVoteArgs -- RPC arguments in args
//
// reply *RequestVoteReply -- RPC reply
//
// The types of args and reply passed to Call() must be
// the same as the types of the arguments declared in the
// handler function (including whether they are pointers)
//
// The rpc package simulates a lossy network, in which servers
// may be unreachable, and in which requests and replies may be lost
//
// Call() sends a request and waits for a reply
//
// If a reply arrives within a timeout interval, Call() returns true;
// otherwise Call() returns false
//
// Thus Call() may not return for a while
//
// A false return can be caused by a dead server, a live server that
// can't be reached, a lost request, or a lost reply
//
// Call() is guaranteed to return (perhaps after a delay)
// *except* if the handler function on the server side does not return
//
// Thus there
// is no need to implement your own timeouts around Call()
//
// Please look at the comments and documentation in ../rpc/rpc.go
// for more details
//
// If you are having trouble getting RPC to work, check that you have
// capitalized all field names in the struct passed over RPC, and
// that the caller passes the address of the reply struct with "&",
// not the struct itself
func (rf *Raft) sendRequestVote(server int, args *RequestVoteArgs, reply *RequestVoteReply) bool { // sends an RPC example (the candidate sends request to all other servers)
ok := rf.peers[server].Call("Raft.RequestVote", args, reply)
rf.mux.Lock()
if ok == true {
if (reply.VoteGranted == true) && (rf.currentTerm == reply.Term) {
rf.receivedVoteCount += 1
if (rf.receivedVoteCount > len(rf.peers)/2) && (rf.role == 3) { // win election
// notify the main routine that it wins the election
// fmt.Println(rf.me, " -----------------------------WIN--------------------------")
if reply.Term == rf.currentTerm { // means that the grant is not out of date
rf.role = 1
rf.mux.Unlock()
rf.becomesLeader <- true
} else {
rf.mux.Unlock()
}
} else {
rf.mux.Unlock()
}
} else { // did not get granted
if reply.Term > rf.currentTerm {
rf.currentTerm = reply.Term
rf.role = 2 // back to follower
// fmt.Println("case 3:")
rf.votedFor = -1
rf.mux.Unlock()
} else {
rf.mux.Unlock()
}
}
} else {
rf.mux.Unlock()
}
return ok
}
// self-added
type AppendEntriesArgs struct {
// Your data here (2A, 2B)
Term int
LeaderId int
PrevLogIndex int
PrevLogTerm int
Entries []LogEntry
LeaderCommitIndex int
}
// self-added
type AppendEntriesReply struct {
// Your data here (2A, 2B)
Term int
Success bool
}
// self-added
func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
select {
default:
if args != nil {
// 1.
rf.mux.Lock()
if (args.Term < rf.currentTerm) { // ERROR!!! if leader is out of date // --------------
reply.Term = rf.currentTerm
// fmt.Println("4: after reply term = ", reply.Term, ", rf.currentTerm = ", rf.currentTerm)
// fmt.Println("rf.commitIndex = ", rf.commitIndex, ", args.LeaderCommitIndex = ", args.LeaderCommitIndex)
reply.Success = false
rf.mux.Unlock()
return
}
if (rf.currentTerm < args.Term) { // ERROR!!!
rf.currentTerm = args.Term
rf.role = 2
// fmt.Println("case 4:")
rf.votedFor = -1
}
rf.mux.Unlock()
if len(args.Entries) == 0 { // empty heartbeat
// fmt.Println(rf.me, " $$$$$$$$$$$$$$$$ receive empty heartbeat")
reply.Term = args.Term
// fmt.Println("5: after reply term = ", reply.Term)
rf.mux.Lock()
rf.commitIndex = args.PrevLogIndex
rf.mux.Unlock()
reply.Success = true
rf.othersElected <- true
return
}
// ordinary appendEntries
// fmt.Println("receive appendEntries ===")
if (rf.log[len(rf.log)-1].Index != args.PrevLogIndex) || (rf.log[len(rf.log)-1].Term != args.PrevLogTerm){ // log does not match
reply.Term = args.Term
// fmt.Println("6: after reply term = ", reply.Term)
reply.Success = false
// TODO: Delete the logs that does not match (delete the last entry in the log)
rf.mux.Lock()
if len(rf.log) > 1 { // should not delete the default log ??????????????
rf.log = rf.log[:len(rf.log)-1] // delete the last entry in the log???
}
// rf.log = rf.log[:len(rf.log)-1] // delete the last entry in the log???
// rf.currentTerm = args.Term // need this? ERROR ???
rf.mux.Unlock()
} else { // leader and follower's previous log match, so just append the log directly
rf.mux.Lock()
// fmt.Println("^^^^^^^^^^^^^^^ before len of log = ", len(rf.log))
for i := 0; i < len(args.Entries); i++ {
rf.log = append(rf.log, args.Entries[i])
}
// fmt.Println("^^^^^^^^^^^^^^^ after len of log = ", len(rf.log))
reply.Term = args.Term
// fmt.Println("7: after reply term = ", reply.Term)
reply.Success = true
// rf.currentTerm = args.Term // need this? ERROR ???
rf.mux.Unlock()
}
// TODO: receiver's rule 5: update commitIndex ????????
rf.mux.Lock()
if args.LeaderCommitIndex > rf.commitIndex {
// fmt.Println(rf.me, ", len of log = ", len(rf.log))
if args.LeaderCommitIndex <= rf.log[len(rf.log)-1].Index {
rf.commitIndex = args.LeaderCommitIndex
} else {
rf.commitIndex = rf.log[len(rf.log)-1].Index
}
}
// fmt.Println(rf.me, " ^^^^^^^^^^^^ commitIndex update to = ", rf.commitIndex, ", lastApplied = ", rf.lastApplied)
// for i := 0; i < len(rf.log); i++ {
// fmt.Println("log [", i, "] = ", rf.log[i].Command)
// }
rf.mux.Unlock()
}
}
}
func (rf *Raft) sendAppendEntries(server int, args *AppendEntriesArgs, reply *AppendEntriesReply) bool {
rf.mux.Lock()
if rf.role != 1 { // --------------
rf.mux.Unlock() // !!!!!!!!!!! ERROR
return true
}
rf.mux.Unlock()
ok := rf.peers[server].Call("Raft.AppendEntries", args, reply)
rf.mux.Lock()
if ok == true {
if rf.role != 1 { // is not leader anymore, should not send
rf.mux.Unlock()
return ok
}
if (reply.Term > args.Term) || (reply.Term > rf.currentTerm) { // out of date
rf.currentTerm = reply.Term
rf.role = 2
rf.votedFor = -1
// fmt.Println("case 1: reply.Term = ", reply.Term, ", args.Term = ", args.Term, ", rf.currentTerm = ", rf.currentTerm)
rf.mux.Unlock()
return false
}
if rf.currentTerm != args.Term { // already out of date
rf.role = 2
// fmt.Println("case 6:")
rf.votedFor = -1
// fmt.Println("case 2: rf.currentTerm = ", rf.currentTerm, ", args.Term = ", args.Term)
rf.mux.Unlock()
return false
}
// TODO
if reply.Success == false { // means the log is not match, need to update the nextIndex
// fmt.Println("@@@@@@@@@@@@ log does not match, so need to decrement by 1")
rf.nextIndex[server] -= 1
rf.matchIndex[server] -= 1 // ERROR ?????
rf.mux.Unlock()
// re-send appendEntries??????
var args_2 AppendEntriesArgs // ERROR all added!!!
args_2.Term = rf.currentTerm
args_2.LeaderId = rf.me
args_2.PrevLogIndex = rf.log[rf.nextIndex[server]-1].Index
args_2.PrevLogTerm = rf.log[rf.nextIndex[server]-1].Term
args_2.Entries = rf.log[rf.nextIndex[server]:]
args_2.LeaderCommitIndex = rf.commitIndex
var reply_2 AppendEntriesReply
go rf.sendAppendEntries(server, &args_2, &reply_2)
return false
} else { // reply success
// rf.mux.Lock()
if len(args.Entries) != 0 { // if not heartbeat
// TODO: update matchIndex and nextIndex ???
rf.matchIndex[server] = args.PrevLogIndex+len(args.Entries)
rf.nextIndex[server] = args.PrevLogIndex+len(args.Entries)+1
// for i := 0; i < len(args.Entries); i++ {
// rf.log[args.PrevLogIndex+1+i].AppendCount += 1
// // if rf.log[args.PrevLogIndex+1+i].AppendCount > len(rf.peers) { // ? len(rf.peers)/2 ??? NEED THIS?
// // rf.log[args.PrevLogIndex+1+i].Committed = true
// // }
// }
// TODO: see whether commitIndex can be updated (leaders' last rule in fig 2)
for i := rf.commitIndex+1; i <= rf.log[len(rf.log)-1].Index; i++ {
count := 0
for j := 0; j < len(rf.peers); j++ {
if rf.matchIndex[j] >= i {
count += 1
}
}
// count += 1 // ????????
// fmt.Println("count (how many servers have appended this log) = ", count)
if (count > len(rf.peers)/2) && (rf.log[i].Term == rf.currentTerm) {
rf.commitIndex = i
} else {
break
}
}
rf.mux.Unlock()
return true
} else { // if heartbeat
rf.mux.Unlock()
return true
}
}
} else {
rf.mux.Unlock()
}
// rf.mux.Unlock()
return ok
}
//
// PutCommand
// =====
//
// The service using Raft (e.g. a k/v server) wants to start
// agreement on the next command to be appended to Raft's log
//
// If this server is not the leader, return false
//
// Otherwise start the agreement and return immediately
//
// There is no guarantee that this command will ever be committed to
// the Raft log, since the leader may fail or lose an election
//
// The first return value is the index that the command will appear at
// if it is ever committed
//
// The second return value is the current term
//
// The third return value is true if this server believes it is
// the leader
//
func (rf *Raft) PutCommand(command interface{}) (int, int, bool) {
rf.mux.Lock()
// index := len(rf.log)-1
index := len(rf.log)
term := rf.currentTerm
isLeader := false
if rf.role == 1 {
isLeader = true
}
rf.mux.Unlock()
// Your code here (2B)
if isLeader == true {
// Append command to its log
rf.mux.Lock()
var newEntry LogEntry
newEntry.Index = len(rf.log)
newEntry.Term = rf.currentTerm
newEntry.Command = command
rf.log = append(rf.log, newEntry)
rf.mux.Unlock()
// fmt.Println("leader: ", rf.me, " appends ", newEntry.Command)
// sendAppendEntries to all followers
for i := 0; i < len(rf.peers); i++{
rf.mux.Lock()
var args AppendEntriesArgs
args.Term = rf.currentTerm
// fmt.Println("########### currentTerm in PutCommand = ", rf.currentTerm)
args.LeaderId = rf.me
args.PrevLogIndex = rf.log[rf.nextIndex[i]-1].Index
args.PrevLogTerm = rf.log[rf.nextIndex[i]-1].Term
args.Entries = rf.log[rf.nextIndex[i]:]
// fmt.Println("i = ", i, ", rf.nextIndex[i] = ", rf.nextIndex[i])
// fmt.Println("### len of args.Entries = ", len(args.Entries))
// fmt.Println("i = ", i, ", args.PrevLogIndex = ", args.PrevLogIndex, ", args.PrevLogTerm = ", args.PrevLogTerm)
args.LeaderCommitIndex = rf.commitIndex
var reply AppendEntriesReply
rf.mux.Unlock()
if i == rf.me { // do not need to send to itself
rf.mux.Lock()
rf.nextIndex[rf.me] += 1 // ?? added
rf.matchIndex[rf.me] += 1 // ?? added ERROR!!!
rf.mux.Unlock()
continue
} else {
go rf.sendAppendEntries(i, &args, &reply) // another goroutines
}
}
}
return index, term, isLeader
}
//
// Stop
// ====
//
// The tester calls Stop() when a Raft instance will not
// be needed again
//
// You are not required to do anything
// in Stop(), but it might be convenient to (for example)
// turn off debug output from this instance
//
func (rf *Raft) Stop() {
// Your code here, if desired
rf.closeRoutine <- true
}
//
// NewPeer
// ====
//
// The service or tester wants to create a Raft server
//
// The port numbers of all the Raft servers (including this one)
// are in peers[]
//
// This server's port is peers[me]
//
// All the servers' peers[] arrays have the same order
//
// applyCh
// =======
//
// applyCh is a channel on which the tester or service expects
// Raft to send ApplyCommand messages
//
// NewPeer() must return quickly, so it should start Goroutines
// for any long-running work
//
func NewPeer(peers []*rpc.ClientEnd, me int, applyCh chan ApplyCommand) *Raft {
rf := &Raft{}
rf.mux.Lock()
rf.peers = peers
rf.me = me
if kEnableDebugLogs {
peerName := peers[me].String()
logPrefix := fmt.Sprintf("%s ", peerName)
if kLogToStdout {
rf.logger = log.New(os.Stdout, peerName, log.Lmicroseconds|log.Lshortfile)
} else {
err := os.MkdirAll(kLogOutputDir, os.ModePerm)
if err != nil {
panic(err.Error())
}
logOutputFile, err := os.OpenFile(fmt.Sprintf("%s/%s.txt", kLogOutputDir, logPrefix), os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0755)
if err != nil {
panic(err.Error())
}
rf.logger = log.New(logOutputFile, logPrefix, log.Lmicroseconds|log.Lshortfile)
}
rf.logger.Println("logger initialized")
} else {
rf.logger = log.New(ioutil.Discard, "", 0)
}
// Your initialization code here (2A, 2B)
rf.role = 2 // default should be a follower
rf.currentTerm = 0 // init to 0
rf.votedFor = -1 // not yet voted
rf.receivedVoteCount = 0
rf.becomesLeader = make(chan bool)
rf.othersElected = make(chan bool)
rf.log = []LogEntry{}
var emptyEntry LogEntry // make the log initially with size 1 (at index 0)
emptyEntry.Index = 0
emptyEntry.Term = 0
// emptyEntry.Command = interface{} // ???
rf.log = append(rf.log, emptyEntry)
rf.commitIndex = 0
rf.lastApplied = 0
rf.nextIndex = make([]int, len(rf.peers))
rf.matchIndex = make([]int, len(rf.peers))
rf.closeRoutine = make(chan bool)
rf.votedResetTicker = make(chan bool)
rf.mux.Unlock()
rf.applyCh = applyCh
go rf.main_routine()
go rf.commit_routine()
return rf
}
func (rf *Raft) main_routine() {
for {
rf.mux.Lock()
curRole := rf.role
rf.mux.Unlock()
if curRole == 1 { // leader
// rf.mux.Lock()
// for i := 0; i < len(rf.peers); i++ {
// fmt.Println("rf.nextIndex[", i, "] = ", rf.nextIndex[i])
// fmt.Println("rf.matchIndex[", i, "] = ", rf.matchIndex[i])
// }
// rf.mux.Unlock()
select {
default:
for i := 0; i < len(rf.peers); i++ {
rf.mux.Lock()
var args_1 AppendEntriesArgs // empty heartbeat
args_1.Term = rf.currentTerm
args_1.LeaderId = rf.me
args_1.PrevLogIndex = 0
args_1.PrevLogTerm = 0
args_1.Entries = make([]LogEntry, 0) // change from string to LogEntry
args_1.LeaderCommitIndex = rf.commitIndex // ERROR !!!!!!!!!!!!!!!!!!!
rf.mux.Unlock()
if i == rf.me { // do not need to send to itself
continue
} else {
var reply AppendEntriesReply
go rf.sendAppendEntries(i, &args_1, &reply) // another goroutines
}
}
time.Sleep(100 * time.Millisecond) // every n time, send appendEntries messages
case <-rf.closeRoutine:
return
}
} else if curRole == 2 { // follower
randomVal := math.Abs(rand.NormFloat64())
testTicker := time.NewTicker(250*time.Millisecond + 100*time.Millisecond*time.Duration(randomVal))
select {
case <-rf.othersElected:
testTicker = time.NewTicker(250*time.Millisecond + 100*time.Millisecond*time.Duration(randomVal)) // --------------
case <-testTicker.C:
rf.mux.Lock() // added here to avoid race condition
rf.role = 3
rf.mux.Unlock()
break
case <-rf.votedResetTicker:
testTicker = time.NewTicker(250*time.Millisecond + 100*time.Millisecond*time.Duration(randomVal))
case <-rf.closeRoutine:
return
}
} else if curRole == 3 { // candidate
rf.mux.Lock()
rf.currentTerm += 1
rf.votedFor = rf.me
rf.receivedVoteCount = 1 // reset
rf.mux.Unlock()
candidateTicker := time.NewTicker(250 * time.Millisecond)
var args RequestVoteArgs
rf.mux.Lock()
args.Term = rf.currentTerm
args.CandidateId = rf.me
args.LastLogIndex = rf.log[len(rf.log)-1].Index // !
args.LastLogTerm = rf.log[len(rf.log)-1].Term // !
rf.mux.Unlock()
for i := 0; i < len(rf.peers); i++ {
if i == rf.me { // do not need to send to itself
continue
} else {
var reply RequestVoteReply
go rf.sendRequestVote(i, &args, &reply) // a separate go routine
}
}
stop := false
for stop == false {
select {
case <-rf.becomesLeader: // this candidate wins the election
rf.mux.Lock()
fmt.Println(rf.me, " ******* becomes leader")
rf.role = 1 // becomes the leader
// TODO: init nextIndex[] and matchIndex[]
for i := 0; i < len(rf.peers); i++ {
rf.nextIndex[i] = rf.log[len(rf.log)-1].Index + 1
rf.matchIndex[i] = 0
// rf.matchIndex[i] = rf.nextIndex[i] - 1 // ?????????????? ERROR ?
}
rf.mux.Unlock()
stop = true
case <-rf.othersElected: // others win the election
rf.mux.Lock()
rf.role = 2 // change back to followers state
// fmt.Println("case 7:")
rf.votedFor = -1
rf.mux.Unlock()
stop = true
break
case <-candidateTicker.C:
rf.mux.Lock()
rf.role = 2
// fmt.Println("case 8:")
rf.votedFor = -1
rf.mux.Unlock()
stop = true
case <-rf.closeRoutine:
return
}
}
}
}
}
func (rf *Raft) commit_routine() { // ensure that each server commits in order
for {
rf.mux.Lock()
// fmt.Println(rf.me, "--------------------------------------------------------------")
// fmt.Println("-------rf.commitIndex = ", rf.commitIndex, ", rf.lastApplied = ", rf.lastApplied, ", len(rf.log) = ", len(rf.log))
if (rf.commitIndex > rf.lastApplied) && (len(rf.log) > rf.lastApplied+1) { // ERROR HERE!!!!!! commitIndex >= and >
// fmt.Println(rf.me, "================================================")
var commitCommand ApplyCommand
commitCommand.Index = rf.log[rf.lastApplied+1].Index
commitCommand.Command = rf.log[rf.lastApplied+1].Command
// rf.log[rf.lastApplied+1].Committed = true // ??? ERROR
rf.lastApplied += 1
rf.mux.Unlock()
// fmt.Println(rf.me, " committed, Index = ", commitCommand.Index, ", Command = ", commitCommand.Command)
rf.applyCh <- commitCommand
fmt.Println(rf.me, " committed, Index = ", commitCommand.Index, ", Command = ", commitCommand.Command)
} else {
rf.mux.Unlock()
}
time.Sleep(50 * time.Millisecond)
}
}
Editor is loading...