Skip to content

Commit 34d67b7

Browse files
pav-kvpav-kv
committed
tracker: rename the paused probes flow field
Signed-off-by: Pavel Kalinnikov <[email protected]>
1 parent b560eab commit 34d67b7

File tree

5 files changed

+54
-49
lines changed

5 files changed

+54
-49
lines changed

‎raft.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1561,7 +1561,7 @@ func stepLeader(r *raft, m pb.Message) error {
15611561
// If snapshot finish, wait for the MsgAppResp from the remote node before sending
15621562
// out the next MsgApp.
15631563
// If snapshot failure, wait for a heartbeat interval before next try
1564-
pr.MsgAppFlowPaused = true
1564+
pr.PauseMsgAppProbes(true)
15651565
case pb.MsgUnreachable:
15661566
// During optimistic replication, if the remote becomes unreachable,
15671567
// there is huge probability that a MsgApp is lost.

‎raft_snap_test.go‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -83,8 +83,8 @@ func TestSnapshotFailure(t *testing.T) {
8383
if sm.trk.Progress[2].Next != 1 {
8484
t.Fatalf("Next = %d, want 1", sm.trk.Progress[2].Next)
8585
}
86-
if !sm.trk.Progress[2].MsgAppFlowPaused {
87-
t.Errorf("MsgAppFlowPaused = %v, want true", sm.trk.Progress[2].MsgAppFlowPaused)
86+
if !sm.trk.Progress[2].MsgAppProbesPaused {
87+
t.Errorf("msgAppProbesPaused = %v, want true", sm.trk.Progress[2].MsgAppProbesPaused)
8888
}
8989
}
9090

@@ -106,8 +106,8 @@ func TestSnapshotSucceed(t *testing.T) {
106106
if sm.trk.Progress[2].Next != 12 {
107107
t.Fatalf("Next = %d, want 12", sm.trk.Progress[2].Next)
108108
}
109-
if !sm.trk.Progress[2].MsgAppFlowPaused {
110-
t.Errorf("MsgAppFlowPaused = %v, want true", sm.trk.Progress[2].MsgAppFlowPaused)
109+
if !sm.trk.Progress[2].MsgAppProbesPaused {
110+
t.Errorf("MsgAppProbesPaused = %v, want true", sm.trk.Progress[2].MsgAppProbesPaused)
111111
}
112112
}
113113

‎raft_test.go‎

Lines changed: 14 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -128,21 +128,21 @@ func TestProgressResumeByHeartbeatResp(t *testing.T) {
128128
r.becomeCandidate()
129129
r.becomeLeader()
130130

131-
r.trk.Progress[2].MsgAppFlowPaused = true
131+
r.trk.Progress[2].PauseMsgAppProbes(true)
132132

133133
r.Step(pb.Message{From: 1, To: 1, Type: pb.MsgBeat})
134-
if !r.trk.Progress[2].MsgAppFlowPaused {
135-
t.Errorf("paused = %v, want true", r.trk.Progress[2].MsgAppFlowPaused)
134+
if !r.trk.Progress[2].MsgAppProbesPaused {
135+
t.Errorf("paused = %v, want true", r.trk.Progress[2].MsgAppProbesPaused)
136136
}
137137

138138
r.trk.Progress[2].BecomeReplicate()
139-
if r.trk.Progress[2].MsgAppFlowPaused {
140-
t.Errorf("paused = %v, want false", r.trk.Progress[2].MsgAppFlowPaused)
139+
if r.trk.Progress[2].MsgAppProbesPaused {
140+
t.Errorf("paused = %v, want false", r.trk.Progress[2].MsgAppProbesPaused)
141141
}
142-
r.trk.Progress[2].MsgAppFlowPaused = true
142+
r.trk.Progress[2].PauseMsgAppProbes(true)
143143
r.Step(pb.Message{From: 2, To: 1, Type: pb.MsgHeartbeatResp})
144-
if !r.trk.Progress[2].MsgAppFlowPaused {
145-
t.Errorf("paused = %v, want true", r.trk.Progress[2].MsgAppFlowPaused)
144+
if !r.trk.Progress[2].MsgAppProbesPaused {
145+
t.Errorf("paused = %v, want true", r.trk.Progress[2].MsgAppProbesPaused)
146146
}
147147
}
148148

@@ -2784,8 +2784,8 @@ func TestSendAppendForProgressProbe(t *testing.T) {
27842784
}
27852785
}
27862786

2787-
if !r.trk.Progress[2].MsgAppFlowPaused {
2788-
t.Errorf("paused = %v, want true", r.trk.Progress[2].MsgAppFlowPaused)
2787+
if !r.trk.Progress[2].MsgAppProbesPaused {
2788+
t.Errorf("paused = %v, want true", r.trk.Progress[2].MsgAppProbesPaused)
27892789
}
27902790
for j := 0; j < 10; j++ {
27912791
mustAppendEntry(r, pb.Entry{Data: []byte("somedata")})
@@ -2799,8 +2799,8 @@ func TestSendAppendForProgressProbe(t *testing.T) {
27992799
for j := 0; j < r.heartbeatTimeout; j++ {
28002800
r.Step(pb.Message{From: 1, To: 1, Type: pb.MsgBeat})
28012801
}
2802-
if !r.trk.Progress[2].MsgAppFlowPaused {
2803-
t.Errorf("paused = %v, want true", r.trk.Progress[2].MsgAppFlowPaused)
2802+
if !r.trk.Progress[2].MsgAppProbesPaused {
2803+
t.Errorf("paused = %v, want true", r.trk.Progress[2].MsgAppProbesPaused)
28042804
}
28052805

28062806
// consume the heartbeat
@@ -2822,8 +2822,8 @@ func TestSendAppendForProgressProbe(t *testing.T) {
28222822
if msg[0].Index != 0 {
28232823
t.Errorf("index = %d, want %d", msg[0].Index, 0)
28242824
}
2825-
if !r.trk.Progress[2].MsgAppFlowPaused {
2826-
t.Errorf("paused = %v, want true", r.trk.Progress[2].MsgAppFlowPaused)
2825+
if !r.trk.Progress[2].MsgAppProbesPaused {
2826+
t.Errorf("paused = %v, want true", r.trk.Progress[2].MsgAppProbesPaused)
28272827
}
28282828
}
28292829

‎tracker/progress.go‎

Lines changed: 18 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -94,13 +94,13 @@ type Progress struct {
9494
// This is always true on the leader.
9595
RecentActive bool
9696

97-
// MsgAppFlowPaused is used when the MsgApp flow to a node is throttled. This
98-
// happens in StateProbe, or StateReplicate with saturated Inflights. In both
99-
// cases, we need to continue sending MsgApp once in a while to guarantee
100-
// progress, but we only do so when MsgAppFlowPaused is false (it is reset on
101-
// receiving a heartbeat response), to not overflow the receiver. See
102-
// IsPaused().
103-
MsgAppFlowPaused bool
97+
// MsgAppProbesPaused set to true prevents sending "probe" MsgApp messages to
98+
// this follower. Used in StateProbe, or StateReplicate when all entries are
99+
// in-flight or the in-flight volume exceeds limits. See ShouldSendMsgApp().
100+
//
101+
// TODO(pav-kv): unexport this field. It is used by a few tests, but should be
102+
// replaced by PauseMsgAppProbes() and ShouldSendMsgApp().
103+
MsgAppProbesPaused bool
104104

105105
// Inflights is a sliding window for the inflight messages.
106106
// Each inflight message contains one or more log entries.
@@ -120,7 +120,7 @@ type Progress struct {
120120
IsLearner bool
121121
}
122122

123-
// ResetState moves the Progress into the specified State, resetting MsgAppFlowPaused,
123+
// ResetState moves the Progress into the specified State, resetting MsgAppProbesPaused,
124124
// PendingSnapshot, and Inflights.
125125
func (pr *Progress) ResetState(state StateType) {
126126
pr.PauseMsgAppProbes(false)
@@ -176,7 +176,7 @@ func (pr *Progress) SentEntries(entries int, bytes uint64) {
176176
// PauseMsgAppProbes pauses or unpauses empty MsgApp messages flow, depending on
177177
// the passed-in bool.
178178
func (pr *Progress) PauseMsgAppProbes(pause bool) {
179-
pr.MsgAppFlowPaused = pause
179+
pr.MsgAppProbesPaused = pause
180180
}
181181

182182
// CanSendEntries returns true if the flow control state allows sending at least
@@ -258,12 +258,17 @@ func (pr *Progress) MaybeDecrTo(rejected, matchHint uint64) bool {
258258
// operation, this is false. A throttled node will be contacted less frequently
259259
// until it has reached a state in which it's able to accept a steady stream of
260260
// log entries again.
261+
//
262+
// TODO(pav-kv): this method is deprecated, remove it. It is still used in tests
263+
// and String(), find a way to avoid this. The problem is that the actual flow
264+
// control state depends on the log size and commit index, which are not part of
265+
// this Progress struct - they are passed-in to methods like ShouldSendMsgApp().
261266
func (pr *Progress) IsPaused() bool {
262267
switch pr.State {
263268
case StateProbe:
264-
return pr.MsgAppFlowPaused
269+
return pr.MsgAppProbesPaused
265270
case StateReplicate:
266-
return pr.MsgAppFlowPaused && pr.Inflights.Full()
271+
return pr.MsgAppProbesPaused && pr.Inflights.Full()
267272
case StateSnapshot:
268273
return true
269274
default:
@@ -293,10 +298,10 @@ func (pr *Progress) IsPaused() bool {
293298
func (pr *Progress) ShouldSendMsgApp(last, commit uint64) bool {
294299
switch pr.State {
295300
case StateProbe:
296-
return !pr.MsgAppFlowPaused
301+
return !pr.MsgAppProbesPaused
297302
case StateReplicate:
298303
return pr.CanBumpCommit(commit) ||
299-
pr.Match < last && (!pr.MsgAppFlowPaused || pr.CanSendEntries(last))
304+
pr.Match < last && (!pr.MsgAppProbesPaused || pr.CanSendEntries(last))
300305
case StateSnapshot:
301306
return false
302307
default:

‎tracker/progress_test.go‎

Lines changed: 17 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -24,14 +24,14 @@ func TestProgressString(t *testing.T) {
2424
ins := NewInflights(1, 0)
2525
ins.Add(123, 1)
2626
pr := &Progress{
27-
Match: 1,
28-
Next: 2,
29-
State: StateSnapshot,
30-
PendingSnapshot: 123,
31-
RecentActive: false,
32-
MsgAppFlowPaused: true,
33-
IsLearner: true,
34-
Inflights: ins,
27+
Match: 1,
28+
Next: 2,
29+
State: StateSnapshot,
30+
PendingSnapshot: 123,
31+
RecentActive: false,
32+
MsgAppProbesPaused: true,
33+
IsLearner: true,
34+
Inflights: ins,
3535
}
3636
const exp = `StateSnapshot match=1 next=2 learner paused pendingSnap=123 inactive inflight=1[full]`
3737
assert.Equal(t, exp, pr.String())
@@ -53,29 +53,29 @@ func TestProgressIsPaused(t *testing.T) {
5353
}
5454
for i, tt := range tests {
5555
p := &Progress{
56-
State: tt.state,
57-
MsgAppFlowPaused: tt.paused,
58-
Inflights: NewInflights(256, 0),
56+
State: tt.state,
57+
MsgAppProbesPaused: tt.paused,
58+
Inflights: NewInflights(256, 0),
5959
}
6060
assert.Equal(t, tt.w, p.IsPaused(), i)
6161
}
6262
}
6363

64-
// TestProgressResume ensures that MaybeDecrTo resets MsgAppFlowPaused, and
64+
// TestProgressResume ensures that MaybeDecrTo resets MsgAppProbesPaused, and
6565
// MaybeUpdate does not.
6666
//
6767
// TODO(pav-kv): there is little sense in testing these micro-behaviours in the
6868
// struct. We should test the visible behaviour instead.
6969
func TestProgressResume(t *testing.T) {
7070
p := &Progress{
71-
Next: 2,
72-
MsgAppFlowPaused: true,
71+
Next: 2,
72+
MsgAppProbesPaused: true,
7373
}
7474
p.MaybeDecrTo(1, 1)
75-
assert.False(t, p.MsgAppFlowPaused)
76-
p.MsgAppFlowPaused = true
75+
assert.False(t, p.MsgAppProbesPaused)
76+
p.MsgAppProbesPaused = true
7777
p.MaybeUpdate(2)
78-
assert.True(t, p.MsgAppFlowPaused)
78+
assert.True(t, p.MsgAppProbesPaused)
7979
}
8080

8181
func TestProgressBecomeProbe(t *testing.T) {

0 commit comments

Comments
 (0)