Skip to content
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.

Sign up for GitHub

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

WIP: Durability issue fix in RAFT layer. #14406

Closed
wants to merge 2 commits into from
Closed

WIP: Durability issue fix in RAFT layer. #14406

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a1de5fc
RAFT considerd leaders entries 'persisted' on ready advance.
ptabor Aug 30, 2022
9c8d974
Tempary Logging for test-fixes
ptabor Aug 30, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions raft/log.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -272,11 +272,13 @@ func (l *raftLog) term(i uint64) (uint64, error) {
// TODO: return an error instead?
return 0, nil
}

log.Printf("UNSTABLE=%+v", l.unstable.entries)
if t, ok := l.unstable.maybeTerm(i); ok {
log.Printf("Found term !!!")
return t, nil
}

li, _ := l.storage.LastIndex()
log.Printf("STABLE=%v", li)
t, err := l.storage.Term(i)
if err == nil {
return t, nil
Expand Down Expand Up @@ -326,6 +328,7 @@ func (l *raftLog) matchTerm(i, term uint64) bool {
}

func (l *raftLog) maybeCommit(maxIndex, term uint64) bool {
log.Println("maxIndex=", maxIndex)
if maxIndex > l.committed && l.zeroTermOnErrCompacted(l.term(maxIndex)) == term {
l.commitTo(maxIndex)
return true
Expand Down
2 changes: 2 additions & 0 deletions raft/node.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -165,6 +165,8 @@ type Node interface {
// commands. For example. when the last Ready contains a snapshot, the application might take
// a long time to apply the snapshot data. To continue receiving Ready without blocking raft
// progress, it can call Advance before finishing applying the last ready.
// Advance must not be called if Entries has not been stored to the stable log storage (WAL).
//
Advance()
// ApplyConfChange applies a config change (previously passed to
// ProposeConfChange) to the node. This must be called whenever a config
Expand Down
26 changes: 21 additions & 5 deletions raft/node_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -373,9 +373,9 @@ func TestNodeProposeAddDuplicateNode(t *testing.T) {
n.Tick()
case rd := <-n.Ready():
s.Append(rd.Entries)
rdyEntries = append(rdyEntries, rd.Entries...)
applied := false
for _, e := range rd.Entries {
rdyEntries = append(rdyEntries, e)
for _, e := range rd.CommittedEntries {
switch e.Type {
case raftpb.EntryNormal:
case raftpb.EntryConfChange:
Expand All @@ -385,8 +385,11 @@ func TestNodeProposeAddDuplicateNode(t *testing.T) {
applied = true
}
}
t.Logf("PreA !!!")
n.Advance()
t.Logf("PostA !!!")
if applied {
t.Logf("Applied !!!")
applyConfChan <- struct{}{}
}
}
Expand Down Expand Up @@ -600,11 +603,17 @@ func TestNodeStart(t *testing.T) {
MustSync: true,
},
{
HardState: raftpb.HardState{Term: 2, Commit: 3, Vote: 1},
HardState: raftpb.HardState{Term: 2, Commit: 2, Vote: 1},
Entries: []raftpb.Entry{{Term: 2, Index: 3, Data: []byte("foo")}},
CommittedEntries: []raftpb.Entry{{Term: 2, Index: 3, Data: []byte("foo")}},
CommittedEntries: []raftpb.Entry{{Term: 2, Index: 2, Data: nil}},
MustSync: true,
},
{
HardState: raftpb.HardState{Term: 2, Commit: 3, Vote: 1},
Entries: nil,
CommittedEntries: []raftpb.Entry{{Term: 2, Index: 3, Data: []byte("foo")}},
MustSync: false,
},
}
storage := NewMemoryStorage()
c := &Config{
Expand Down Expand Up @@ -640,6 +649,13 @@ func TestNodeStart(t *testing.T) {
n.Advance()
}

if g3 := <-n.Ready(); !reflect.DeepEqual(g3, wants[2]) {
t.Errorf("#%d!: g = %+v,\n w = %+v", 3, g3, wants[2])
} else {
storage.Append(g3.Entries)
n.Advance()
}

select {
case rd := <-n.Ready():
t.Errorf("unexpected Ready: %+v", rd)
Expand Down Expand Up @@ -759,7 +775,7 @@ func TestNodeAdvance(t *testing.T) {
n.Advance()

n.Campaign(ctx)
<-n.Ready()
rd = <-n.Ready()

n.Propose(ctx, []byte("foo"))
select {
Expand Down
9 changes: 6 additions & 3 deletions raft/raft.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -573,6 +573,11 @@ func (r *raft) advance(rd Ready) {
if len(rd.Entries) > 0 {
e := rd.Entries[len(rd.Entries)-1]
r.raftLog.stableTo(e.Index, e.Term)
if r.prs.Progress[r.id] != nil {
r.logger.Infof("Updating progress: %v", e.Index)
r.prs.Progress[r.id].MaybeUpdate(e.Index)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This isn't thread(goroutine) safe either, because it might trigger the map data-race error.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You might need to construct a pb.MsgAppResp message and call (*node)Step()

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agree. Was looking for a way to do this within 'raft' - without exiting to the 'node' level,
but calling r.Step() would be risky.

}
r.maybeCommit() // TODO: Consider moving to stepLeader loop...
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This isn't thread(goroutine) safe, because the raft may do it concurrently.

Copy link
Member

@ahrtr ahrtr Aug 31, 2022 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No need to call r.maybeCommit() if you follow comment

}
if !IsEmptySnap(rd.Snapshot) {
r.raftLog.stableSnapTo(rd.Snapshot.Metadata.Index)
Expand Down Expand Up @@ -633,9 +638,7 @@ func (r *raft) appendEntry(es ...pb.Entry) (accepted bool) {
// Drop the proposal.
return false
}
// use latest "last" index after truncate/append
li = r.raftLog.append(es...)
r.prs.Progress[r.id].MaybeUpdate(li)
r.raftLog.append(es...)
// Regardless of maybeCommit's return, our caller will call bcastAppend.
r.maybeCommit()
return true
Expand Down