log.go 4.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206
  1. package raft
  2. import "fmt"
  3. const (
  4. Normal int64 = iota
  5. ClusterInit
  6. AddNode
  7. RemoveNode
  8. )
  9. const (
  10. defaultCompactThreshold = 10000
  11. )
  12. func (e *Entry) isConfig() bool {
  13. return e.Type == AddNode || e.Type == RemoveNode
  14. }
  15. type raftLog struct {
  16. ents []Entry
  17. unstable int64
  18. committed int64
  19. applied int64
  20. offset int64
  21. // want a compact after the number of entries exceeds the threshold
  22. // TODO(xiangli) size might be a better criteria
  23. compactThreshold int64
  24. }
  25. func newLog() *raftLog {
  26. return &raftLog{
  27. ents: make([]Entry, 1),
  28. unstable: 1,
  29. committed: 0,
  30. applied: 0,
  31. compactThreshold: defaultCompactThreshold,
  32. }
  33. }
  34. func (l *raftLog) isEmpty() bool {
  35. return l.offset == 0 && len(l.ents) == 1
  36. }
  37. func (l *raftLog) String() string {
  38. return fmt.Sprintf("offset=%d committed=%d applied=%d len(ents)=%d", l.offset, l.committed, l.applied, len(l.ents))
  39. }
  40. func (l *raftLog) maybeAppend(index, logTerm, committed int64, ents ...Entry) bool {
  41. if l.matchTerm(index, logTerm) {
  42. from := index + 1
  43. ci := l.findConflict(from, ents)
  44. switch {
  45. case ci == -1:
  46. case ci <= l.committed:
  47. panic("conflict with committed entry")
  48. default:
  49. l.append(ci-1, ents[ci-from:]...)
  50. }
  51. if l.committed < committed {
  52. l.committed = min(committed, l.lastIndex())
  53. }
  54. return true
  55. }
  56. return false
  57. }
  58. func (l *raftLog) append(after int64, ents ...Entry) int64 {
  59. l.ents = append(l.slice(l.offset, after+1), ents...)
  60. l.unstable = min(l.unstable, after+1)
  61. return l.lastIndex()
  62. }
  63. func (l *raftLog) findConflict(from int64, ents []Entry) int64 {
  64. for i, ne := range ents {
  65. if oe := l.at(from + int64(i)); oe == nil || oe.Term != ne.Term {
  66. return from + int64(i)
  67. }
  68. }
  69. return -1
  70. }
  71. func (l *raftLog) unstableEnts() []Entry {
  72. ents := l.entries(l.unstable)
  73. l.unstable = l.lastIndex() + 1
  74. return ents
  75. }
  76. func (l *raftLog) lastIndex() int64 {
  77. return int64(len(l.ents)) - 1 + l.offset
  78. }
  79. func (l *raftLog) term(i int64) int64 {
  80. if e := l.at(i); e != nil {
  81. return e.Term
  82. }
  83. return -1
  84. }
  85. func (l *raftLog) entries(i int64) []Entry {
  86. // never send out the first entry
  87. // first entry is only used for matching
  88. // prevLogTerm
  89. if i == l.offset {
  90. panic("cannot return the first entry in log")
  91. }
  92. return l.slice(i, l.lastIndex()+1)
  93. }
  94. func (l *raftLog) isUpToDate(i, term int64) bool {
  95. e := l.at(l.lastIndex())
  96. return term > e.Term || (term == e.Term && i >= l.lastIndex())
  97. }
  98. func (l *raftLog) matchTerm(i, term int64) bool {
  99. if e := l.at(i); e != nil {
  100. return e.Term == term
  101. }
  102. return false
  103. }
  104. func (l *raftLog) maybeCommit(maxIndex, term int64) bool {
  105. if maxIndex > l.committed && l.term(maxIndex) == term {
  106. l.committed = maxIndex
  107. return true
  108. }
  109. return false
  110. }
  111. // nextEnts returns all the available entries for execution.
  112. // all the returned entries will be marked as applied.
  113. func (l *raftLog) nextEnts() (ents []Entry) {
  114. if l.committed > l.applied {
  115. ents = l.slice(l.applied+1, l.committed+1)
  116. l.applied = l.committed
  117. }
  118. return ents
  119. }
  120. // compact compacts all log entries until i.
  121. // It removes the log entries before i, exclusive.
  122. // i must be not smaller than the index of the first entry
  123. // and not greater than the index of the last entry.
  124. // the number of entries after compaction will be returned.
  125. func (l *raftLog) compact(i int64) int64 {
  126. if l.isOutOfBounds(i) {
  127. panic(fmt.Sprintf("compact %d out of bounds [%d:%d]", i, l.offset, l.lastIndex()))
  128. }
  129. l.ents = l.slice(i, l.lastIndex()+1)
  130. l.unstable = max(i+1, l.unstable)
  131. l.offset = i
  132. return int64(len(l.ents))
  133. }
  134. func (l *raftLog) shouldCompact() bool {
  135. return (l.applied - l.offset) > l.compactThreshold
  136. }
  137. func (l *raftLog) restore(index, term int64) {
  138. l.ents = []Entry{{Term: term}}
  139. l.unstable = index + 1
  140. l.committed = index
  141. l.applied = index
  142. l.offset = index
  143. }
  144. func (l *raftLog) at(i int64) *Entry {
  145. if l.isOutOfBounds(i) {
  146. return nil
  147. }
  148. return &l.ents[i-l.offset]
  149. }
  150. // slice get a slice of log entries from lo through hi-1, inclusive.
  151. func (l *raftLog) slice(lo int64, hi int64) []Entry {
  152. if lo >= hi {
  153. return nil
  154. }
  155. if l.isOutOfBounds(lo) || l.isOutOfBounds(hi-1) {
  156. return nil
  157. }
  158. return l.ents[lo-l.offset : hi-l.offset]
  159. }
  160. func (l *raftLog) isOutOfBounds(i int64) bool {
  161. if i < l.offset || i > l.lastIndex() {
  162. return true
  163. }
  164. return false
  165. }
  166. func min(a, b int64) int64 {
  167. if a > b {
  168. return b
  169. }
  170. return a
  171. }
  172. func max(a, b int64) int64 {
  173. if a > b {
  174. return a
  175. }
  176. return b
  177. }