/
replica_state.go
225 lines (201 loc) · 6.49 KB
/
replica_state.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
package main
import (
"golang.org/x/net/context"
"encoding/binary"
"fmt"
"github.com/coreos/etcd/raft/raftpb"
"github.com/journeymidnight/badger"
"github.com/journeymidnight/nentropy/helper"
"github.com/journeymidnight/nentropy/osd/keys"
"github.com/journeymidnight/nentropy/osd/multiraftbase"
"github.com/journeymidnight/nentropy/storage/engine"
"github.com/journeymidnight/nentropy/util/protoutil"
)
type replicaStateLoader struct {
keys.GroupIDPrefixBuf
}
func makeReplicaStateLoader(groupID multiraftbase.GroupID) replicaStateLoader {
return replicaStateLoader{
GroupIDPrefixBuf: keys.MakeGroupIDPrefixBuf(groupID),
}
}
func (rsl replicaStateLoader) loadHardState(
ctx context.Context, reader engine.Reader,
) (raftpb.HardState, error) {
var data []byte
data, err := reader.Get(rsl.RaftHardStateKey())
if err != nil && err != badger.ErrKeyNotFound {
return raftpb.HardState{}, err
}
var hs raftpb.HardState
err = hs.Unmarshal(data)
if err != nil {
return raftpb.HardState{}, err
}
return hs, nil
}
func (rsl replicaStateLoader) loadTruncatedState(
ctx context.Context, reader engine.Reader,
) (multiraftbase.RaftTruncatedState, error) {
var truncState multiraftbase.RaftTruncatedState
var data []byte
data, err := reader.Get(rsl.RaftTruncatedStateKey())
if err != nil && err != badger.ErrKeyNotFound {
return multiraftbase.RaftTruncatedState{}, err
}
err = truncState.Unmarshal(data)
if err != nil {
return multiraftbase.RaftTruncatedState{}, err
}
return truncState, nil
}
func (rsl replicaStateLoader) setTruncatedState(
ctx context.Context,
eng engine.Writer,
truncState *multiraftbase.RaftTruncatedState,
) error {
if (*truncState == multiraftbase.RaftTruncatedState{}) {
return nil
}
val, err := truncState.Marshal()
if err != nil {
}
return eng.Put(rsl.RaftTruncatedStateKey(), val)
}
func loadTruncatedState(
ctx context.Context, reader engine.Reader, groupID multiraftbase.GroupID,
) (multiraftbase.RaftTruncatedState, error) {
rsl := makeReplicaStateLoader(groupID)
return rsl.loadTruncatedState(ctx, reader)
}
func (rsl replicaStateLoader) setHardState(
ctx context.Context, batch engine.Writer, st raftpb.HardState,
) error {
data, err := st.Marshal()
if err != nil {
return err
}
return batch.Put(rsl.RaftHardStateKey(), data)
}
// loadAppliedIndex returns the Raft applied index and the lease applied index.
func (rsl replicaStateLoader) loadAppliedIndex(
ctx context.Context, reader engine.Reader,
) (uint64, error) {
var appliedIndex uint64
v, err := reader.Get(rsl.RaftAppliedIndexKey())
if err != nil && err != badger.ErrKeyNotFound {
return 0, err
}
if v != nil {
int64AppliedIndex, n := binary.Varint(v)
if n <= 0 {
return 0, fmt.Errorf("int64 varint decoding failed: %d", n)
}
appliedIndex = uint64(int64AppliedIndex)
}
return appliedIndex, nil
}
// setAppliedIndex sets the {raft,lease} applied index values, properly
// accounting for existing keys in the returned stats.
func (rsl replicaStateLoader) setAppliedIndex(
ctx context.Context,
eng engine.Writer,
appliedIndex uint64,
) error {
data := make([]byte, binary.MaxVarintLen64)
binary.PutVarint(data, int64(appliedIndex))
if err := eng.Put(rsl.RaftAppliedIndexKey(),
data); err != nil {
return err
}
return nil
}
// loadState loads a ReplicaState from disk. The exception is the Desc field,
// which is updated transactionally, and is populated from the supplied
// GroupDescriptor under the convention that that is the latest committed
// version.
func (rsl replicaStateLoader) load(
ctx context.Context, reader engine.Reader, desc *multiraftbase.GroupDescriptor,
) (multiraftbase.ReplicaState, error) {
var s multiraftbase.ReplicaState
// TODO(tschottdorf): figure out whether this is always synchronous with
// on-disk state (likely iffy during Split/ChangeReplica triggers).
s.Desc = protoutil.Clone(desc).(*multiraftbase.GroupDescriptor)
var err error
if s.RaftAppliedIndex, err = rsl.loadAppliedIndex(ctx, reader); err != nil {
return multiraftbase.ReplicaState{}, err
}
// The truncated state should not be optional (i.e. the pointer is
// pointless), but it is and the migration is not worth it.
truncState, err := rsl.loadTruncatedState(ctx, reader)
if err != nil {
return multiraftbase.ReplicaState{}, err
}
s.TruncatedState = &truncState
helper.Println(5, "replicaStateLoader load RaftAppliedIndex:", s.RaftAppliedIndex)
helper.Println(5, "replicaStateLoader load TruncatedState Index:", s.TruncatedState.Index, " Term:", s.TruncatedState.Term)
return s, nil
}
func (rsl replicaStateLoader) save(
ctx context.Context, eng engine.Writer, state multiraftbase.ReplicaState,
) error {
if err := rsl.setAppliedIndex(
ctx, eng, state.RaftAppliedIndex,
); err != nil {
return err
}
if err := rsl.setTruncatedState(ctx, eng, state.TruncatedState); err != nil {
return err
}
return nil
}
// The rest is not technically part of ReplicaState.
// TODO(tschottdorf): more consolidation of ad-hoc structures: last index and
// hard state. These are closely coupled with ReplicaState (and in particular
// with its TruncatedState) but are different in that they are not consistently
// updated through Raft.
func loadLastIndex(
ctx context.Context, reader engine.Reader, groupID multiraftbase.GroupID,
) (uint64, error) {
rsl := makeReplicaStateLoader(groupID)
return rsl.loadLastIndex(ctx, reader)
}
func (rsl replicaStateLoader) loadLastIndex(
ctx context.Context, reader engine.Reader,
) (uint64, error) {
var lastIndex uint64
v, err := reader.Get(rsl.RaftLastIndexKey())
if err != nil && err != badger.ErrKeyNotFound {
return 0, err
}
if v != nil {
int64LastIndex, n := binary.Varint(v)
if n <= 0 {
return 0, fmt.Errorf("int64 varint decoding failed: %d", n)
}
lastIndex = uint64(int64LastIndex)
} else {
// The log is empty, which means we are either starting from scratch
// or the entire log has been truncated away.
lastEnt, err := rsl.loadTruncatedState(ctx, reader)
if err != nil {
return 0, err
}
lastIndex = lastEnt.Index
}
return lastIndex, nil
}
func (rsl replicaStateLoader) setLastIndex(
ctx context.Context, writer engine.Writer, lastIndex uint64,
) error {
data := make([]byte, binary.MaxVarintLen64)
binary.PutVarint(data, int64(lastIndex))
return writer.Put(rsl.RaftLastIndexKey(), data)
}
func loadAppliedIndex(
ctx context.Context, reader engine.Reader, groupID multiraftbase.GroupID,
) (uint64, error) {
helper.Println(5, " loadAppliedIndex, groupID:", groupID)
rsl := makeReplicaStateLoader(groupID)
return rsl.loadAppliedIndex(ctx, reader)
}