raft/state.go at main · sumimakito/raft · GitHub

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
package raft

import (
	"sync/atomic"
)

type ServerRole uint32

const (
	Leader ServerRole = 1 + iota
	Candidate
	Follower
)

func (r ServerRole) String() string {
	switch r {
	case Leader:
		return "Leader"
	case Follower:
		return "Follower"
	case Candidate:
		return "Candidate"
	}
	return "Unknown"
}

type voteSummary struct {
	term      uint64
	candidate string
}

// nilVoteSummary is equivalent to the "unvoted" state due to its zero term
var nilVoteSummary = voteSummary{term: 0, candidate: ""}

type serverState struct {
	noCopy

	stateRole            ServerRole   // volatile
	stateCurrentTerm     uint64       // persistent
	stateFirstLogIndex   uint64       // volatile
	stateLastLogIndex    uint64       // volatile
	stateLastVoteSummary atomic.Value // voteSummary persistent
	stateShutdownState   uint32       // volatile
}

func (s *Server) restoreStates() error {
	atomic.StoreUint64(&s.serverState.stateCurrentTerm, Must2(s.stableStore.CurrentTerm()))
	atomic.StoreUint64(&s.serverState.stateFirstLogIndex, Must2(s.logStore.FirstIndex()))
	atomic.StoreUint64(&s.serverState.stateLastLogIndex, Must2(s.logStore.LastIndex()))
	s.serverState.stateLastVoteSummary.Store(Must2(s.stableStore.LastVote()))
	return nil
}

func (s *Server) role() ServerRole {
	return ServerRole(atomic.LoadUint32((*uint32)(&s.serverState.stateRole)))
}

func (s *Server) setRole(role ServerRole) {
	atomic.StoreUint32((*uint32)(&s.serverState.stateRole), uint32(role))
}

func (s *Server) currentTerm() uint64 {
	return atomic.LoadUint64(&s.serverState.stateCurrentTerm)
}

func (s *Server) setCurrentTerm(currentTerm uint64) {
	Must1(s.stableStore.SetCurrentTerm(currentTerm))
	atomic.StoreUint64(&s.serverState.stateCurrentTerm, currentTerm)
}

func (s *Server) firstLogIndex() uint64 {
	return atomic.LoadUint64(&s.serverState.stateFirstLogIndex)
}

func (s *Server) setFirstLogIndex(firstLogIndex uint64) {
	atomic.StoreUint64(&s.serverState.stateFirstLogIndex, firstLogIndex)
}

func (s *Server) lastLogIndex() uint64 {
	return atomic.LoadUint64(&s.serverState.stateLastLogIndex)
}

func (s *Server) setLastLogIndex(lastLogIndex uint64) {
	atomic.StoreUint64(&s.serverState.stateLastLogIndex, lastLogIndex)
}

func (s *Server) lastVoteSummary() voteSummary {
	if v := s.serverState.stateLastVoteSummary.Load(); v != nil {
		return v.(voteSummary)
	}
	return nilVoteSummary
}

func (s *Server) setLastVoteSummary(term uint64, candidate string) {
	summary := voteSummary{term: term, candidate: candidate}
	Must1(s.stableStore.SetLastVote(summary))
	s.serverState.stateLastVoteSummary.Store(summary)
}

func (server *Server) shutdownState() bool {
	return atomic.LoadUint32(&server.serverState.stateShutdownState) != 0
}

func (server *Server) setShutdownState() bool {
	return atomic.CompareAndSwapUint32(&server.serverState.stateShutdownState, 0, 1)
}

type lastAppliedTuple struct {
	Index uint64
	Term  uint64
}

var nilLastAppliedTuple = lastAppliedTuple{Index: 0, Term: 0}

type commitState struct {
	noCopy

	aCommitIndex uint64
	aLastApplied atomic.Value // lastAppliedTuple
}

func (state *commitState) commitIndex() uint64 {
	return atomic.LoadUint64(&state.aCommitIndex)
}

func (state *commitState) setCommitIndex(commitIndex uint64) {
	atomic.StoreUint64(&state.aCommitIndex, commitIndex)
}

func (state *commitState) lastApplied() lastAppliedTuple {
	if v := state.aLastApplied.Load(); v != nil {
		return v.(lastAppliedTuple)
	}
	return nilLastAppliedTuple
}

func (state *commitState) setLastApplied(index, term uint64) {
	state.aLastApplied.Store(lastAppliedTuple{Index: index, Term: term})
}

// StateStore defines the interface to save and restore the persistent
// server states from a stable store.
type StateStore interface {
	CurrentTerm() (uint64, error)
	SetCurrentTerm(term uint64) error
	LastVote() (voteSummary, error)
	SetLastVote(summary voteSummary) error
}