Mirroristas/etcd
2
0
Fork 0
mirror of https://github.com/etcd-io/etcd.git synced 2024-09-27 06:25:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

raft: cleanup maybeSendAppend method

Browse Source 
- avoid large indented blocks, leave the main block unindented
- declare pb.Message inlined in the sending call

Signed-off-by: Pavel Kalinnikov <pavel@cockroachlabs.com>
This commit is contained in:
Pavel Kalinnikov 2022-11-01 21:45:10 +00:00
parent 5619953f33
commit d5ac7b833f
1 changed files with 27 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

50
raft/raft.go
View File

@ -437,8 +437,6 @@ func (r *raft) maybeSendAppend(to uint64, sendIfEmpty bool) bool {
if pr.IsPaused() { if pr.IsPaused() {
return false return false
} }
m := pb.Message{}
m.To = to
term, errt := r.raftLog.term(pr.Next - 1) term, errt := r.raftLog.term(pr.Next - 1)
ents, erre := r.raftLog.entries(pr.Next, r.maxMsgSize) ents, erre := r.raftLog.entries(pr.Next, r.maxMsgSize)
@ -452,7 +450,6 @@ func (r *raft) maybeSendAppend(to uint64, sendIfEmpty bool) bool {
return false return false
} }
m.Type = pb.MsgSnap
snapshot, err := r.raftLog.snapshot() snapshot, err := r.raftLog.snapshot()
if err != nil { if err != nil {
if err == ErrSnapshotTemporarilyUnavailable { if err == ErrSnapshotTemporarilyUnavailable {
@ -464,33 +461,40 @@ func (r *raft) maybeSendAppend(to uint64, sendIfEmpty bool) bool {
if IsEmptySnap(snapshot) { if IsEmptySnap(snapshot) {
panic("need non-empty snapshot") panic("need non-empty snapshot")
} }
m.Snapshot = snapshot
sindex, sterm := snapshot.Metadata.Index, snapshot.Metadata.Term sindex, sterm := snapshot.Metadata.Index, snapshot.Metadata.Term
r.logger.Debugf("%x [firstindex: %d, commit: %d] sent snapshot[index: %d, term: %d] to %x [%s]", r.logger.Debugf("%x [firstindex: %d, commit: %d] sent snapshot[index: %d, term: %d] to %x [%s]",
r.id, r.raftLog.firstIndex(), r.raftLog.committed, sindex, sterm, to, pr) r.id, r.raftLog.firstIndex(), r.raftLog.committed, sindex, sterm, to, pr)
pr.BecomeSnapshot(sindex) pr.BecomeSnapshot(sindex)
r.logger.Debugf("%x paused sending replication messages to %x [%s]", r.id, to, pr) r.logger.Debugf("%x paused sending replication messages to %x [%s]", r.id, to, pr)
} else {
m.Type = pb.MsgApp r.send(pb.Message{To: to, Type: pb.MsgSnap, Snapshot: snapshot})
m.Index = pr.Next - 1 return true
m.LogTerm = term }
m.Entries = ents
m.Commit = r.raftLog.committed // Send the actual MsgApp otherwise, and update the progress accordingly.
if n := len(m.Entries); n != 0 { // TODO(pavelkalinnikov): factor out the Progress update to a method
switch pr.State { next := pr.Next // save Next for later, as the progress update can change it
// optimistically increase the next when in StateReplicate if n := len(ents); n != 0 {
case tracker.StateReplicate: switch pr.State {
last := m.Entries[n-1].Index // optimistically increase the next when in StateReplicate
pr.OptimisticUpdate(last) case tracker.StateReplicate:
pr.Inflights.Add(last) last := ents[n-1].Index
case tracker.StateProbe: pr.OptimisticUpdate(last)
pr.ProbeSent = true pr.Inflights.Add(last)
default: case tracker.StateProbe:
r.logger.Panicf("%x is sending append in unhandled state %s", r.id, pr.State) pr.ProbeSent = true
} default:
r.logger.Panicf("%x is sending append in unhandled state %s", r.id, pr.State)
} }
} }
r.send(m) r.send(pb.Message{
To: to,
Type: pb.MsgApp,
Index: next - 1,
LogTerm: term,
Entries: ents,
Commit: r.raftLog.committed,
})
return true return true
} }