-
Notifications
You must be signed in to change notification settings - Fork 9.6k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
raft: add learner #8605
raft: add learner #8605
Changes from 4 commits
a91a51d
c098dbb
b159733
ee003a5
aff1c50
abb497a
2d2c6b9
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -116,6 +116,8 @@ type Config struct { | |
// used for testing right now. | ||
peers []uint64 | ||
|
||
learners []uint64 | ||
|
||
// ElectionTick is the number of Node.Tick invocations that must pass between | ||
// elections. That is, if a follower does not receive any message from the | ||
// leader of current term before ElectionTick has elapsed, it will become | ||
|
@@ -238,6 +240,8 @@ type raft struct { | |
|
||
state StateType | ||
|
||
isLearner bool | ||
|
||
votes map[uint64]bool | ||
|
||
msgs []pb.Message | ||
|
@@ -288,19 +292,22 @@ func newRaft(c *Config) *raft { | |
if err != nil { | ||
panic(err) // TODO(bdarnell) | ||
} | ||
peers := c.peers | ||
if len(cs.Nodes) > 0 { | ||
if len(peers) > 0 { | ||
voters := c.peers | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. use peers instead of voters. i do not want to create a new term for now. |
||
learners := c.learners | ||
if len(cs.Nodes) > 0 || len(cs.Learners) > 0 { | ||
if len(voters) > 0 || len(learners) > 0 { | ||
// TODO(bdarnell): the peers argument is always nil except in | ||
// tests; the argument should be removed and these tests should be | ||
// updated to specify their nodes through a snapshot. | ||
panic("cannot specify both newRaft(peers) and ConfState.Nodes)") | ||
panic("cannot specify both newRaft(peers, learners) and ConfState.(Nodes, Learners)") | ||
} | ||
peers = cs.Nodes | ||
voters = cs.Nodes | ||
learners = cs.Learners | ||
} | ||
r := &raft{ | ||
id: c.ID, | ||
lead: None, | ||
isLearner: false, | ||
raftLog: raftlog, | ||
maxMsgSize: c.MaxSizePerMsg, | ||
maxInflight: c.MaxInflightMsgs, | ||
|
@@ -313,8 +320,17 @@ func newRaft(c *Config) *raft { | |
readOnly: newReadOnly(c.ReadOnlyOption), | ||
disableProposalForwarding: c.DisableProposalForwarding, | ||
} | ||
for _, p := range peers { | ||
r.prs[p] = &Progress{Next: 1, ins: newInflights(r.maxInflight)} | ||
for _, n := range voters { | ||
r.prs[n] = &Progress{Next: 1, ins: newInflights(r.maxInflight), isLearner: false} | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. revert this change. bool is false by default. |
||
} | ||
for _, n := range learners { | ||
if _, has := r.prs[n]; has { | ||
panic(fmt.Sprintf("cannot specify both Voter and Learner for node: %x", n)) | ||
} | ||
r.prs[n] = &Progress{Next: 1, ins: newInflights(r.maxInflight), isLearner: true} | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. maybe it is easier to have a separate prs map for learners. |
||
if r.id == n { | ||
r.isLearner = true | ||
} | ||
} | ||
if !isHardStateEqual(hs, emptyState) { | ||
r.loadState(hs) | ||
|
@@ -346,7 +362,18 @@ func (r *raft) hardState() pb.HardState { | |
} | ||
} | ||
|
||
func (r *raft) quorum() int { return len(r.prs)/2 + 1 } | ||
// TODO(shuaili): optimize, maybe record voterCount in raft | ||
func (r *raft) voterCount() int { | ||
count := 0 | ||
for _, p := range r.prs { | ||
if !p.isLearner { | ||
count++ | ||
} | ||
} | ||
return count | ||
} | ||
|
||
func (r *raft) quorum() int { return r.voterCount()/2 + 1 } | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This function is called frequently, maybe now we can use a variable to cache the quorum. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. agreed |
||
|
||
func (r *raft) nodes() []uint64 { | ||
nodes := make([]uint64, 0, len(r.prs)) | ||
|
@@ -504,9 +531,11 @@ func (r *raft) bcastHeartbeatWithCtx(ctx []byte) { | |
// r.bcastAppend). | ||
func (r *raft) maybeCommit() bool { | ||
// TODO(bmizerany): optimize.. Currently naive | ||
mis := make(uint64Slice, 0, len(r.prs)) | ||
for id := range r.prs { | ||
mis = append(mis, r.prs[id].Match) | ||
mis := make(uint64Slice, 0, r.voterCount()) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. seem we can still use |
||
for _, p := range r.prs { | ||
if !p.isLearner { | ||
mis = append(mis, p.Match) | ||
} | ||
} | ||
sort.Sort(sort.Reverse(mis)) | ||
mci := mis[r.quorum()-1] | ||
|
@@ -528,7 +557,7 @@ func (r *raft) reset(term uint64) { | |
|
||
r.votes = make(map[uint64]bool) | ||
for id := range r.prs { | ||
r.prs[id] = &Progress{Next: r.raftLog.lastIndex() + 1, ins: newInflights(r.maxInflight)} | ||
r.prs[id] = &Progress{Next: r.raftLog.lastIndex() + 1, ins: newInflights(r.maxInflight), isLearner: r.prs[id].isLearner} | ||
if id == r.id { | ||
r.prs[id].Match = r.raftLog.lastIndex() | ||
} | ||
|
@@ -676,6 +705,9 @@ func (r *raft) campaign(t CampaignType) { | |
if id == r.id { | ||
continue | ||
} | ||
if r.prs[id].isLearner { | ||
continue | ||
} | ||
r.logger.Infof("%x [logterm: %d, index: %d] sent %s request to %x at term %d", | ||
r.id, r.raftLog.lastTerm(), r.raftLog.lastIndex(), voteMsg, id, r.Term) | ||
|
||
|
@@ -990,6 +1022,10 @@ func stepLeader(r *raft, m pb.Message) { | |
} | ||
r.logger.Debugf("%x failed to send message to %x because it is unreachable [%s]", r.id, m.From, pr) | ||
case pb.MsgTransferLeader: | ||
if pr.isLearner { | ||
r.logger.Debugf("%x is Learner. Ignored transferring leadership", r.id) | ||
return | ||
} | ||
leadTransferee := m.From | ||
lastLeadTransferee := r.leadTransferee | ||
if lastLeadTransferee != None { | ||
|
@@ -1171,33 +1207,58 @@ func (r *raft) restore(s pb.Snapshot) bool { | |
|
||
r.raftLog.restore(s) | ||
r.prs = make(map[uint64]*Progress) | ||
for _, n := range s.Metadata.ConfState.Nodes { | ||
r.restoreNode(s.Metadata.ConfState.Nodes, false) | ||
r.restoreNode(s.Metadata.ConfState.Learners, true) | ||
return true | ||
} | ||
|
||
func (r *raft) restoreNode(nodes []uint64, isLearner bool) { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. probably restoreNode can take both nodes and learnerNodes as args. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. +1 |
||
for _, n := range nodes { | ||
match, next := uint64(0), r.raftLog.lastIndex()+1 | ||
if n == r.id { | ||
match = next - 1 | ||
r.isLearner = isLearner | ||
} | ||
r.setProgress(n, match, next) | ||
r.setProgress(n, match, next, isLearner) | ||
r.logger.Infof("%x restored progress of %x [%s]", r.id, n, r.prs[n]) | ||
} | ||
return true | ||
} | ||
|
||
// promotable indicates whether state machine can be promoted to leader, | ||
// which is true when its own id is in progress list. | ||
// which is true when its own id is in progress list and its not learner. | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. it is not a learner |
||
func (r *raft) promotable() bool { | ||
_, ok := r.prs[r.id] | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. if r.isLearer {
return false
}
... is cleaner |
||
return ok | ||
return ok && !r.isLearner | ||
} | ||
|
||
func (r *raft) addNode(id uint64) { | ||
r.addLearnerNode(id, false) | ||
} | ||
|
||
func (r *raft) addLearner(id uint64) { | ||
r.addLearnerNode(id, true) | ||
} | ||
|
||
func (r *raft) addLearnerNode(id uint64, isLearner bool) { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. addNodeOrLearnerNode. or probably just duplicate code a little bit, and remove this func. |
||
r.pendingConf = false | ||
if _, ok := r.prs[id]; ok { | ||
// Ignore any redundant addNode calls (which can happen because the | ||
// initial bootstrapping entries are applied twice). | ||
return | ||
if r.prs[id].isLearner == isLearner { | ||
// Ignore any redundant addNode calls (which can happen because the | ||
// initial bootstrapping entries are applied twice). | ||
return | ||
} | ||
if isLearner { | ||
// can only change Learner to Voter | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. log here. |
||
return | ||
} | ||
r.prs[id].isLearner = isLearner | ||
} else { | ||
r.setProgress(id, 0, r.raftLog.lastIndex()+1, isLearner) | ||
} | ||
|
||
r.setProgress(id, 0, r.raftLog.lastIndex()+1) | ||
if r.id == id { | ||
r.isLearner = isLearner | ||
} | ||
// When a node is first added, we should mark it as recently active. | ||
// Otherwise, CheckQuorum may cause us to step down if it is invoked | ||
// before the added node has a chance to communicate with us. | ||
|
@@ -1226,8 +1287,8 @@ func (r *raft) removeNode(id uint64) { | |
|
||
func (r *raft) resetPendingConf() { r.pendingConf = false } | ||
|
||
func (r *raft) setProgress(id, match, next uint64) { | ||
r.prs[id] = &Progress{Next: next, Match: match, ins: newInflights(r.maxInflight)} | ||
func (r *raft) setProgress(id, match, next uint64, isLearner bool) { | ||
r.prs[id] = &Progress{Next: next, Match: match, ins: newInflights(r.maxInflight), isLearner: isLearner} | ||
} | ||
|
||
func (r *raft) delProgress(id uint64) { | ||
|
@@ -1267,7 +1328,7 @@ func (r *raft) checkQuorumActive() bool { | |
continue | ||
} | ||
|
||
if r.prs[id].RecentActive { | ||
if r.prs[id].RecentActive && !r.prs[id].isLearner { | ||
act++ | ||
} | ||
|
||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
doc string?