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
|
package fuzzy
import (
"bufio"
"encoding/binary"
"fmt"
"hash/adler32"
"io"
"os"
"github.com/hashicorp/raft"
)
type logHash struct {
lastHash []byte
}
func (l *logHash) Add(d []byte) {
hasher := adler32.New()
hasher.Write(l.lastHash)
hasher.Write(d)
l.lastHash = hasher.Sum(nil)
}
type applyItem struct {
index uint64
term uint64
data []byte
}
func (a *applyItem) set(l *raft.Log) {
a.index = l.Index
a.term = l.Term
a.data = make([]byte, len(l.Data))
copy(a.data, l.Data)
}
type fuzzyFSM struct {
logHash
lastTerm uint64
lastIndex uint64
applied []applyItem
}
func (f *fuzzyFSM) Apply(l *raft.Log) interface{} {
if l.Index <= f.lastIndex {
panic(fmt.Errorf("fsm.Apply received log entry with invalid Index %v (lastIndex we saw was %d)", l, f.lastIndex))
}
if l.Term < f.lastTerm {
panic(fmt.Errorf("fsm.Apply received log entry with invalid Term %v (lastTerm we saw was %d)", l, f.lastTerm))
}
f.lastIndex = l.Index
f.lastTerm = l.Term
f.Add(l.Data)
f.applied = append(f.applied, applyItem{})
f.applied[len(f.applied)-1].set(l)
return nil
}
func (f *fuzzyFSM) WriteTo(fn string) error {
fw, err := os.Create(fn)
if err != nil {
return err
}
defer fw.Close()
w := bufio.NewWriter(fw)
defer w.Flush()
for _, i := range f.applied {
fmt.Fprintf(w, "%d.%8d: %X\n", i.term, i.index, i.data)
}
return nil
}
func (f *fuzzyFSM) Snapshot() (raft.FSMSnapshot, error) {
s := *f
return &s, nil
}
func (f *fuzzyFSM) Restore(r io.ReadCloser) error {
err := binary.Read(r, binary.LittleEndian, &f.lastTerm)
if err == nil {
err = binary.Read(r, binary.LittleEndian, &f.lastIndex)
}
if err == nil {
f.lastHash = make([]byte, adler32.Size)
_, err = r.Read(f.lastHash)
}
return err
}
func (f *fuzzyFSM) Persist(sink raft.SnapshotSink) error {
err := binary.Write(sink, binary.LittleEndian, f.lastTerm)
if err == nil {
err = binary.Write(sink, binary.LittleEndian, f.lastIndex)
}
if err == nil {
_, err = sink.Write(f.lastHash)
}
return err
}
func (f *fuzzyFSM) Release() {
}
|
