forked from etcd-io/raft
-
Notifications
You must be signed in to change notification settings - Fork 0
/
log.go
574 lines (520 loc) · 19.3 KB
/
log.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
// Copyright 2015 The etcd Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package raft
import (
"fmt"
pb "go.etcd.io/raft/v3/raftpb"
)
type raftLog struct {
// storage contains all stable entries since the last snapshot.
storage Storage
// unstable contains all unstable entries and snapshot.
// they will be saved into storage.
unstable unstable
// committed is the highest log position that is known to be in
// stable storage on a quorum of nodes.
committed uint64
// applying is the highest log position that the application has
// been instructed to apply to its state machine. Some of these
// entries may be in the process of applying and have not yet
// reached applied.
// Use: The field is incremented when accepting a Ready struct.
// Invariant: applied <= applying && applying <= committed
applying uint64
// applied is the highest log position that the application has
// successfully applied to its state machine.
// Use: The field is incremented when advancing after the committed
// entries in a Ready struct have been applied (either synchronously
// or asynchronously).
// Invariant: applied <= committed
applied uint64
logger Logger
// maxApplyingEntsSize limits the outstanding byte size of the messages
// returned from calls to nextCommittedEnts that have not been acknowledged
// by a call to appliedTo.
maxApplyingEntsSize entryEncodingSize
// applyingEntsSize is the current outstanding byte size of the messages
// returned from calls to nextCommittedEnts that have not been acknowledged
// by a call to appliedTo.
applyingEntsSize entryEncodingSize
// applyingEntsPaused is true when entry application has been paused until
// enough progress is acknowledged.
applyingEntsPaused bool
}
// newLog returns log using the given storage and default options. It
// recovers the log to the state that it just commits and applies the
// latest snapshot.
func newLog(storage Storage, logger Logger) *raftLog {
return newLogWithSize(storage, logger, noLimit)
}
// newLogWithSize returns a log using the given storage and max
// message size.
func newLogWithSize(storage Storage, logger Logger, maxApplyingEntsSize entryEncodingSize) *raftLog {
firstIndex, err := storage.FirstIndex()
if err != nil {
panic(err) // TODO(bdarnell)
}
lastIndex, err := storage.LastIndex()
if err != nil {
panic(err) // TODO(bdarnell)
}
return &raftLog{
storage: storage,
unstable: unstable{
offset: lastIndex + 1,
offsetInProgress: lastIndex + 1,
logger: logger,
},
maxApplyingEntsSize: maxApplyingEntsSize,
// Initialize our committed and applied pointers to the time of the last compaction.
committed: firstIndex - 1,
applying: firstIndex - 1,
applied: firstIndex - 1,
logger: logger,
}
}
func (l *raftLog) String() string {
return fmt.Sprintf("committed=%d, applied=%d, applying=%d, unstable.offset=%d, unstable.offsetInProgress=%d, len(unstable.Entries)=%d",
l.committed, l.applied, l.applying, l.unstable.offset, l.unstable.offsetInProgress, len(l.unstable.entries))
}
// maybeAppend returns (0, false) if the entries cannot be appended. Otherwise,
// it returns (last index of new entries, true).
func (l *raftLog) maybeAppend(a logSlice, committed uint64) (lastnewi uint64, ok bool) {
if !l.matchTerm(a.prev) {
return 0, false
}
// TODO(pav-kv): propagate logSlice down the stack. It will be used all the
// way down in unstable, for safety checks, and for useful bookkeeping.
lastnewi = a.prev.index + uint64(len(a.entries))
ci := l.findConflict(a.entries)
switch {
case ci == 0:
case ci <= l.committed:
l.logger.Panicf("entry %d conflict with committed entry [committed(%d)]", ci, l.committed)
default:
offset := a.prev.index + 1
if ci-offset > uint64(len(a.entries)) {
l.logger.Panicf("index, %d, is out of range [%d]", ci-offset, len(a.entries))
}
l.append(a.entries[ci-offset:]...)
}
l.commitTo(min(committed, lastnewi))
return lastnewi, true
}
func (l *raftLog) append(ents ...pb.Entry) uint64 {
if len(ents) == 0 {
return l.lastIndex()
}
if after := ents[0].Index - 1; after < l.committed {
l.logger.Panicf("after(%d) is out of range [committed(%d)]", after, l.committed)
}
l.unstable.truncateAndAppend(ents)
return l.lastIndex()
}
// findConflict finds the index of the conflict.
// It returns the first pair of conflicting entries between the existing
// entries and the given entries, if there are any.
// If there is no conflicting entries, and the existing entries contains
// all the given entries, zero will be returned.
// If there is no conflicting entries, but the given entries contains new
// entries, the index of the first new entry will be returned.
// An entry is considered to be conflicting if it has the same index but
// a different term.
// The index of the given entries MUST be continuously increasing.
func (l *raftLog) findConflict(ents []pb.Entry) uint64 {
for i := range ents {
if id := pbEntryID(&ents[i]); !l.matchTerm(id) {
if id.index <= l.lastIndex() {
// TODO(pav-kv): can simply print %+v of the id. This will change the
// log format though.
l.logger.Infof("found conflict at index %d [existing term: %d, conflicting term: %d]",
id.index, l.zeroTermOnOutOfBounds(l.term(id.index)), id.term)
}
return id.index
}
}
return 0
}
// findConflictByTerm returns a best guess on where this log ends matching
// another log, given that the only information known about the other log is the
// (index, term) of its single entry.
//
// Specifically, the first returned value is the max guessIndex <= index, such
// that term(guessIndex) <= term or term(guessIndex) is not known (because this
// index is compacted or not yet stored).
//
// The second returned value is the term(guessIndex), or 0 if it is unknown.
//
// This function is used by a follower and leader to resolve log conflicts after
// an unsuccessful append to a follower, and ultimately restore the steady flow
// of appends.
func (l *raftLog) findConflictByTerm(index uint64, term uint64) (uint64, uint64) {
for ; index > 0; index-- {
// If there is an error (likely ErrCompacted or ErrUnavailable), we don't
// know whether it's a match or not, so assume a possible match and return
// the index, with 0 term indicating an unknown term.
if ourTerm, err := l.term(index); err != nil {
return index, 0
} else if ourTerm <= term {
return index, ourTerm
}
}
return 0, 0
}
// nextUnstableEnts returns all entries that are available to be written to the
// local stable log and are not already in-progress.
func (l *raftLog) nextUnstableEnts() []pb.Entry {
return l.unstable.nextEntries()
}
// hasNextUnstableEnts returns if there are any entries that are available to be
// written to the local stable log and are not already in-progress.
func (l *raftLog) hasNextUnstableEnts() bool {
return len(l.nextUnstableEnts()) > 0
}
// hasNextOrInProgressUnstableEnts returns if there are any entries that are
// available to be written to the local stable log or in the process of being
// written to the local stable log.
func (l *raftLog) hasNextOrInProgressUnstableEnts() bool {
return len(l.unstable.entries) > 0
}
// nextCommittedEnts returns all the available entries for execution.
// Entries can be committed even when the local raft instance has not durably
// appended them to the local raft log yet. If allowUnstable is true, committed
// entries from the unstable log may be returned; otherwise, only entries known
// to reside locally on stable storage will be returned.
func (l *raftLog) nextCommittedEnts(allowUnstable bool) (ents []pb.Entry) {
if l.applyingEntsPaused {
// Entry application outstanding size limit reached.
return nil
}
if l.hasNextOrInProgressSnapshot() {
// See comment in hasNextCommittedEnts.
return nil
}
lo, hi := l.applying+1, l.maxAppliableIndex(allowUnstable)+1 // [lo, hi)
if lo >= hi {
// Nothing to apply.
return nil
}
maxSize := l.maxApplyingEntsSize - l.applyingEntsSize
if maxSize <= 0 {
l.logger.Panicf("applying entry size (%d-%d)=%d not positive",
l.maxApplyingEntsSize, l.applyingEntsSize, maxSize)
}
ents, err := l.slice(lo, hi, maxSize)
if err != nil {
l.logger.Panicf("unexpected error when getting unapplied entries (%v)", err)
}
return ents
}
// hasNextCommittedEnts returns if there is any available entries for execution.
// This is a fast check without heavy raftLog.slice() in nextCommittedEnts().
func (l *raftLog) hasNextCommittedEnts(allowUnstable bool) bool {
if l.applyingEntsPaused {
// Entry application outstanding size limit reached.
return false
}
if l.hasNextOrInProgressSnapshot() {
// If we have a snapshot to apply, don't also return any committed
// entries. Doing so raises questions about what should be applied
// first.
return false
}
lo, hi := l.applying+1, l.maxAppliableIndex(allowUnstable)+1 // [lo, hi)
return lo < hi
}
// maxAppliableIndex returns the maximum committed index that can be applied.
// If allowUnstable is true, committed entries from the unstable log can be
// applied; otherwise, only entries known to reside locally on stable storage
// can be applied.
func (l *raftLog) maxAppliableIndex(allowUnstable bool) uint64 {
hi := l.committed
if !allowUnstable {
hi = min(hi, l.unstable.offset-1)
}
return hi
}
// nextUnstableSnapshot returns the snapshot, if present, that is available to
// be applied to the local storage and is not already in-progress.
func (l *raftLog) nextUnstableSnapshot() *pb.Snapshot {
return l.unstable.nextSnapshot()
}
// hasNextUnstableSnapshot returns if there is a snapshot that is available to
// be applied to the local storage and is not already in-progress.
func (l *raftLog) hasNextUnstableSnapshot() bool {
return l.unstable.nextSnapshot() != nil
}
// hasNextOrInProgressSnapshot returns if there is pending snapshot waiting for
// applying or in the process of being applied.
func (l *raftLog) hasNextOrInProgressSnapshot() bool {
return l.unstable.snapshot != nil
}
func (l *raftLog) snapshot() (pb.Snapshot, error) {
if l.unstable.snapshot != nil {
return *l.unstable.snapshot, nil
}
return l.storage.Snapshot()
}
func (l *raftLog) firstIndex() uint64 {
if i, ok := l.unstable.maybeFirstIndex(); ok {
return i
}
index, err := l.storage.FirstIndex()
if err != nil {
panic(err) // TODO(bdarnell)
}
return index
}
func (l *raftLog) lastIndex() uint64 {
if i, ok := l.unstable.maybeLastIndex(); ok {
return i
}
i, err := l.storage.LastIndex()
if err != nil {
panic(err) // TODO(bdarnell)
}
return i
}
func (l *raftLog) commitTo(tocommit uint64) {
// never decrease commit
if l.committed < tocommit {
if l.lastIndex() < tocommit {
l.logger.Panicf("tocommit(%d) is out of range [lastIndex(%d)]. Was the raft log corrupted, truncated, or lost?", tocommit, l.lastIndex())
}
l.committed = tocommit
}
}
func (l *raftLog) appliedTo(i uint64, size entryEncodingSize) {
if l.committed < i || i < l.applied {
l.logger.Panicf("applied(%d) is out of range [prevApplied(%d), committed(%d)]", i, l.applied, l.committed)
}
l.applied = i
l.applying = max(l.applying, i)
if l.applyingEntsSize > size {
l.applyingEntsSize -= size
} else {
// Defense against underflow.
l.applyingEntsSize = 0
}
l.applyingEntsPaused = l.applyingEntsSize >= l.maxApplyingEntsSize
}
func (l *raftLog) acceptApplying(i uint64, size entryEncodingSize, allowUnstable bool) {
if l.committed < i {
l.logger.Panicf("applying(%d) is out of range [prevApplying(%d), committed(%d)]", i, l.applying, l.committed)
}
l.applying = i
l.applyingEntsSize += size
// Determine whether to pause entry application until some progress is
// acknowledged. We pause in two cases:
// 1. the outstanding entry size equals or exceeds the maximum size.
// 2. the outstanding entry size does not equal or exceed the maximum size,
// but we determine that the next entry in the log will push us over the
// limit. We determine this by comparing the last entry returned from
// raftLog.nextCommittedEnts to the maximum entry that the method was
// allowed to return had there been no size limit. If these indexes are
// not equal, then the returned entries slice must have been truncated to
// adhere to the memory limit.
l.applyingEntsPaused = l.applyingEntsSize >= l.maxApplyingEntsSize ||
i < l.maxAppliableIndex(allowUnstable)
}
func (l *raftLog) stableTo(id entryID) { l.unstable.stableTo(id) }
func (l *raftLog) stableSnapTo(i uint64) { l.unstable.stableSnapTo(i) }
// acceptUnstable indicates that the application has started persisting the
// unstable entries in storage, and that the current unstable entries are thus
// to be marked as being in-progress, to avoid returning them with future calls
// to Ready().
func (l *raftLog) acceptUnstable() { l.unstable.acceptInProgress() }
// lastEntryID returns the ID of the last entry in the log.
func (l *raftLog) lastEntryID() entryID {
index := l.lastIndex()
t, err := l.term(index)
if err != nil {
l.logger.Panicf("unexpected error when getting the last term at %d: %v", index, err)
}
return entryID{term: t, index: index}
}
func (l *raftLog) term(i uint64) (uint64, error) {
// Check the unstable log first, even before computing the valid term range,
// which may need to access stable Storage. If we find the entry's term in
// the unstable log, we know it was in the valid range.
if t, ok := l.unstable.maybeTerm(i); ok {
return t, nil
}
// The valid term range is [firstIndex-1, lastIndex]. Even though the entry at
// firstIndex-1 is compacted away, its term is available for matching purposes
// when doing log appends.
if i+1 < l.firstIndex() {
return 0, ErrCompacted
}
if i > l.lastIndex() {
return 0, ErrUnavailable
}
t, err := l.storage.Term(i)
if err == nil {
return t, nil
}
if err == ErrCompacted || err == ErrUnavailable {
return 0, err
}
panic(err) // TODO(bdarnell)
}
func (l *raftLog) entries(i uint64, maxSize entryEncodingSize) ([]pb.Entry, error) {
if i > l.lastIndex() {
return nil, nil
}
return l.slice(i, l.lastIndex()+1, maxSize)
}
// allEntries returns all entries in the log.
func (l *raftLog) allEntries() []pb.Entry {
ents, err := l.entries(l.firstIndex(), noLimit)
if err == nil {
return ents
}
if err == ErrCompacted { // try again if there was a racing compaction
return l.allEntries()
}
// TODO (xiangli): handle error?
panic(err)
}
// isUpToDate determines if a log with the given last entry is more up-to-date
// by comparing the index and term of the last entries in the existing logs.
//
// If the logs have last entries with different terms, then the log with the
// later term is more up-to-date. If the logs end with the same term, then
// whichever log has the larger lastIndex is more up-to-date. If the logs are
// the same, the given log is up-to-date.
func (l *raftLog) isUpToDate(their entryID) bool {
our := l.lastEntryID()
return their.term > our.term || their.term == our.term && their.index >= our.index
}
func (l *raftLog) matchTerm(id entryID) bool {
t, err := l.term(id.index)
if err != nil {
return false
}
return t == id.term
}
func (l *raftLog) maybeCommit(at entryID) bool {
// NB: term should never be 0 on a commit because the leader campaigned at
// least at term 1. But if it is 0 for some reason, we don't consider this a
// term match.
if at.term != 0 && at.index > l.committed && l.matchTerm(at) {
l.commitTo(at.index)
return true
}
return false
}
func (l *raftLog) restore(s pb.Snapshot) {
l.logger.Infof("log [%s] starts to restore snapshot [index: %d, term: %d]", l, s.Metadata.Index, s.Metadata.Term)
l.committed = s.Metadata.Index
l.unstable.restore(s)
}
// scan visits all log entries in the [lo, hi) range, returning them via the
// given callback. The callback can be invoked multiple times, with consecutive
// sub-ranges of the requested range. Returns up to pageSize bytes worth of
// entries at a time. May return more if a single entry size exceeds the limit.
//
// The entries in [lo, hi) must exist, otherwise scan() eventually returns an
// error (possibly after passing some entries through the callback).
//
// If the callback returns an error, scan terminates and returns this error
// immediately. This can be used to stop the scan early ("break" the loop).
func (l *raftLog) scan(lo, hi uint64, pageSize entryEncodingSize, v func([]pb.Entry) error) error {
for lo < hi {
ents, err := l.slice(lo, hi, pageSize)
if err != nil {
return err
} else if len(ents) == 0 {
return fmt.Errorf("got 0 entries in [%d, %d)", lo, hi)
}
if err := v(ents); err != nil {
return err
}
lo += uint64(len(ents))
}
return nil
}
// slice returns a slice of log entries from lo through hi-1, inclusive.
func (l *raftLog) slice(lo, hi uint64, maxSize entryEncodingSize) ([]pb.Entry, error) {
if err := l.mustCheckOutOfBounds(lo, hi); err != nil {
return nil, err
}
if lo == hi {
return nil, nil
}
if lo >= l.unstable.offset {
ents := limitSize(l.unstable.slice(lo, hi), maxSize)
// NB: use the full slice expression to protect the unstable slice from
// appends to the returned ents slice.
return ents[:len(ents):len(ents)], nil
}
cut := min(hi, l.unstable.offset)
ents, err := l.storage.Entries(lo, cut, uint64(maxSize))
if err == ErrCompacted {
return nil, err
} else if err == ErrUnavailable {
l.logger.Panicf("entries[%d:%d) is unavailable from storage", lo, cut)
} else if err != nil {
panic(err) // TODO(pavelkalinnikov): handle errors uniformly
}
if hi <= l.unstable.offset {
return ents, nil
}
// Fast path to check if ents has reached the size limitation. Either the
// returned slice is shorter than requested (which means the next entry would
// bring it over the limit), or a single entry reaches the limit.
if uint64(len(ents)) < cut-lo {
return ents, nil
}
// Slow path computes the actual total size, so that unstable entries are cut
// optimally before being copied to ents slice.
size := entsSize(ents)
if size >= maxSize {
return ents, nil
}
unstable := limitSize(l.unstable.slice(l.unstable.offset, hi), maxSize-size)
// Total size of unstable may exceed maxSize-size only if len(unstable) == 1.
// If this happens, ignore this extra entry.
if len(unstable) == 1 && size+entsSize(unstable) > maxSize {
return ents, nil
}
// Otherwise, total size of unstable does not exceed maxSize-size, so total
// size of ents+unstable does not exceed maxSize. Simply concatenate them.
return extend(ents, unstable), nil
}
// l.firstIndex <= lo <= hi <= l.firstIndex + len(l.entries)
func (l *raftLog) mustCheckOutOfBounds(lo, hi uint64) error {
if lo > hi {
l.logger.Panicf("invalid slice %d > %d", lo, hi)
}
fi := l.firstIndex()
if lo < fi {
return ErrCompacted
}
length := l.lastIndex() + 1 - fi
if hi > fi+length {
l.logger.Panicf("slice[%d,%d) out of bound [%d,%d]", lo, hi, fi, l.lastIndex())
}
return nil
}
func (l *raftLog) zeroTermOnOutOfBounds(t uint64, err error) uint64 {
if err == nil {
return t
}
if err == ErrCompacted || err == ErrUnavailable {
return 0
}
l.logger.Panicf("unexpected error (%v)", err)
return 0
}