-
Notifications
You must be signed in to change notification settings - Fork 175
/
forest.go
374 lines (324 loc) · 12 KB
/
forest.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
package mtrie
import (
"bytes"
"encoding/hex"
"errors"
"fmt"
lru "github.com/hashicorp/golang-lru"
"github.com/onflow/flow-go/ledger"
"github.com/onflow/flow-go/ledger/complete/mtrie/trie"
"github.com/onflow/flow-go/module"
)
// Forest holds several in-memory tries. As Forest is a storage-abstraction layer,
// we assume that all registers are addressed via paths of pre-defined uniform length.
//
// Forest has a limit, the forestCapacity, on the number of tries it is able to store.
// If more tries are added than the capacity, the Least Recently Used trie is
// removed (evicted) from the Forest. THIS IS A ROUGH HEURISTIC as it might evict
// tries that are still needed. In fully matured Flow, we will have an
// explicit eviction policy.
//
// TODO: Storage Eviction Policy for Forest
// For the execution node: we only evict on sealing a result.
type Forest struct {
// tries stores all MTries in the forest. It is NOT a CACHE in the conventional sense:
// there is no mechanism to load a trie from disk in case of a cache miss. Missing a
// needed trie in the forest might cause a fatal application logic error.
tries *lru.Cache
forestCapacity int
onTreeEvicted func(tree *trie.MTrie) error
pathByteSize int // length [bytes] of register path
metrics module.LedgerMetrics
}
// NewForest returns a new instance of memory forest.
//
// CAUTION on forestCapacity: the specified capacity MUST be SUFFICIENT to store all needed MTries in the forest.
// If more tries are added than the capacity, the Least Recently Used trie is removed (evicted) from the Forest.
// THIS IS A ROUGH HEURISTIC as it might evict tries that are still needed.
// Make sure you chose a sufficiently large forestCapacity, such that, when reaching the capacity, the
// Least Recently Used trie will never be needed again.
func NewForest(pathByteSize int, forestCapacity int, metrics module.LedgerMetrics, onTreeEvicted func(tree *trie.MTrie) error) (*Forest, error) {
// init LRU cache as a SHORTCUT for a usage-related storage eviction policy
var cache *lru.Cache
var err error
if onTreeEvicted != nil {
cache, err = lru.NewWithEvict(forestCapacity, func(key interface{}, value interface{}) {
trie, ok := value.(*trie.MTrie)
if !ok {
panic(fmt.Sprintf("cache contains item of type %T", value))
}
//TODO Log error
_ = onTreeEvicted(trie)
})
} else {
cache, err = lru.New(forestCapacity)
}
if err != nil {
return nil, fmt.Errorf("cannot create forest cache: %w", err)
}
// init Forest and add an empty trie
if pathByteSize < 1 {
return nil, errors.New("trie's path size [in bytes] must be positive")
}
forest := &Forest{tries: cache,
forestCapacity: forestCapacity,
onTreeEvicted: onTreeEvicted,
pathByteSize: pathByteSize,
metrics: metrics,
}
// add empty roothash
emptyTrie, err := trie.NewEmptyMTrie(pathByteSize)
if err != nil {
return nil, fmt.Errorf("constructing empty trie for forest failed: %w", err)
}
err = forest.AddTrie(emptyTrie)
if err != nil {
return nil, fmt.Errorf("adding empty trie to forest failed: %w", err)
}
return forest, nil
}
// Read reads values for an slice of paths and returns values and error (if any)
// TODO: can be optimized further if we don't care about changing the order of the input r.Paths
func (f *Forest) Read(r *ledger.TrieRead) ([]*ledger.Payload, error) {
if len(r.Paths) == 0 {
return []*ledger.Payload{}, nil
}
// lookup the trie by rootHash
trie, err := f.GetTrie(r.RootHash)
if err != nil {
return nil, err
}
// deduplicate keys:
// Generally, we expect the VM to deduplicate reads and writes. Hence, the following is a pre-caution.
// TODO: We could take out the following de-duplication logic
// Which increases the cost for duplicates but reduces read complexity without duplicates.
deduplicatedPaths := make([]ledger.Path, 0, len(r.Paths))
pathOrgIndex := make(map[string][]int)
for i, path := range r.Paths {
// check key sizes
if len(path) != f.pathByteSize {
return nil, fmt.Errorf("path size doesn't match the trie height: %x", len(path))
}
// only collect duplicated keys once
indices, ok := pathOrgIndex[string(path)]
if !ok { // deduplication here is optional
deduplicatedPaths = append(deduplicatedPaths, path)
}
// append the index
pathOrgIndex[string(path)] = append(indices, i)
}
payloads := trie.UnsafeRead(deduplicatedPaths) // this sorts deduplicatedPaths IN-PLACE
// reconstruct the payloads in the same key order that called the method
orderedPayloads := make([]*ledger.Payload, len(r.Paths))
totalPayloadSize := 0
for i, p := range deduplicatedPaths {
payload := payloads[i]
indices := pathOrgIndex[string(p)]
for _, j := range indices {
orderedPayloads[j] = payload.DeepCopy()
}
totalPayloadSize += len(indices) * payload.Size()
}
// TODO rename the metrics
f.metrics.ReadValuesSize(uint64(totalPayloadSize))
return orderedPayloads, nil
}
// Update updates the Values for the registers and returns rootHash and error (if any).
// In case there are multiple updates to the same register, Update will persist the latest
// written value.
func (f *Forest) Update(u *ledger.TrieUpdate) (ledger.RootHash, error) {
parentTrie, err := f.GetTrie(u.RootHash)
if err != nil {
return nil, err
}
if len(u.Paths) == 0 { // no key no change
return u.RootHash, nil
}
// Deduplicate writes to the same register: we only retain the value of the last write
// Generally, we expect the VM to deduplicate reads and writes.
deduplicatedPaths := make([]ledger.Path, 0, len(u.Paths))
deduplicatedPayloads := make([]ledger.Payload, 0, len(u.Paths))
payloadMap := make(map[string]int) // index into deduplicatedPaths, deduplicatedPayloads with register update
totalPayloadSize := 0
for i, path := range u.Paths {
// check path sizes
if len(path) != f.pathByteSize {
return nil, fmt.Errorf("path size doesn't match the trie height: %x", len(path))
}
payload := u.Payloads[i]
// check if we already have encountered an update for the respective register
if idx, ok := payloadMap[string(path)]; ok {
oldPayload := deduplicatedPayloads[idx]
deduplicatedPayloads[idx] = *payload
totalPayloadSize += -oldPayload.Size() + payload.Size()
} else {
payloadMap[string(path)] = len(deduplicatedPaths)
deduplicatedPaths = append(deduplicatedPaths, path)
deduplicatedPayloads = append(deduplicatedPayloads, *u.Payloads[i])
totalPayloadSize += payload.Size()
}
}
// TODO rename metrics names
f.metrics.UpdateValuesSize(uint64(totalPayloadSize))
newTrie, err := trie.NewTrieWithUpdatedRegisters(parentTrie, deduplicatedPaths, deduplicatedPayloads)
if err != nil {
return nil, fmt.Errorf("constructing updated trie failed: %w", err)
}
f.metrics.LatestTrieRegCount(newTrie.AllocatedRegCount())
f.metrics.LatestTrieRegCountDiff(newTrie.AllocatedRegCount() - parentTrie.AllocatedRegCount())
f.metrics.LatestTrieMaxDepth(uint64(newTrie.MaxDepth()))
f.metrics.LatestTrieMaxDepthDiff(uint64(newTrie.MaxDepth() - parentTrie.MaxDepth()))
err = f.AddTrie(newTrie)
if err != nil {
return nil, fmt.Errorf("adding updated trie to forest failed: %w", err)
}
return ledger.RootHash(newTrie.RootHash()), nil
}
// Proofs returns a batch proof for the given paths
func (f *Forest) Proofs(r *ledger.TrieRead) (*ledger.TrieBatchProof, error) {
// no path, empty batchproof
if len(r.Paths) == 0 {
return ledger.NewTrieBatchProof(), nil
}
// look up for non existing paths
retPayloads, err := f.Read(r)
if err != nil {
return nil, err
}
deduplicatedPaths := make([]ledger.Path, 0)
notFoundPaths := make([]ledger.Path, 0)
notFoundPayloads := make([]ledger.Payload, 0)
pathOrgIndex := make(map[string][]int)
for i, path := range r.Paths {
// check key sizes
if len(path) != f.pathByteSize {
return nil, fmt.Errorf("path size doesn't match the trie height: %x", len(path))
}
// only collect duplicated keys once
if _, ok := pathOrgIndex[string(path)]; !ok {
deduplicatedPaths = append(deduplicatedPaths, path)
pathOrgIndex[string(path)] = []int{i}
// add it only once if is empty
if retPayloads[i].IsEmpty() {
notFoundPaths = append(notFoundPaths, path)
notFoundPayloads = append(notFoundPayloads, *ledger.EmptyPayload())
}
} else {
// handles duplicated keys
pathOrgIndex[string(path)] = append(pathOrgIndex[string(path)], i)
}
}
stateTrie, err := f.GetTrie(r.RootHash)
if err != nil {
return nil, err
}
// if we have to insert empty values
if len(notFoundPaths) > 0 {
newTrie, err := trie.NewTrieWithUpdatedRegisters(stateTrie, notFoundPaths, notFoundPayloads)
if err != nil {
return nil, err
}
// rootHash shouldn't change
if !bytes.Equal(newTrie.RootHash(), r.RootHash) {
return nil, errors.New("root hash has changed during the operation")
}
stateTrie = newTrie
}
bp := ledger.NewTrieBatchProofWithEmptyProofs(len(deduplicatedPaths))
for _, p := range bp.Proofs {
p.Flags = make([]byte, f.pathByteSize)
p.Inclusion = false
}
stateTrie.UnsafeProofs(deduplicatedPaths, bp.Proofs)
// reconstruct the proofs in the same key order that called the method
retbp := ledger.NewTrieBatchProofWithEmptyProofs(len(r.Paths))
for i, p := range deduplicatedPaths {
for _, j := range pathOrgIndex[string(p)] {
retbp.Proofs[j] = bp.Proofs[i]
}
}
return retbp, nil
}
// PathLength return the length [in bytes] the trie operates with.
// Concurrency safe (as Tries are immutable structures by convention)
func (f *Forest) PathLength() int { return f.pathByteSize }
// GetTrie returns trie at specific rootHash
// warning, use this function for read-only operation
func (f *Forest) GetTrie(rootHash ledger.RootHash) (*trie.MTrie, error) {
encRootHash := hex.EncodeToString(rootHash)
// if in memory
if ent, ok := f.tries.Get(encRootHash); ok {
return ent.(*trie.MTrie), nil
}
return nil, fmt.Errorf("trie with the given rootHash [%v] not found", encRootHash)
}
// GetTries returns list of currently cached tree root hashes
func (f *Forest) GetTries() ([]*trie.MTrie, error) {
// ToDo needs concurrency safety
keys := f.tries.Keys()
tries := make([]*trie.MTrie, 0, len(keys))
for _, key := range keys {
t, ok := f.tries.Get(key)
if !ok {
return nil, errors.New("concurrent Forest modification")
}
tries = append(tries, t.(*trie.MTrie))
}
return tries, nil
}
// AddTries adds a trie to the forest
func (f *Forest) AddTries(newTries []*trie.MTrie) error {
for _, t := range newTries {
err := f.AddTrie(t)
if err != nil {
return fmt.Errorf("adding tries to forest failed: %w", err)
}
}
return nil
}
// AddTrie adds a trie to the forest
func (f *Forest) AddTrie(newTrie *trie.MTrie) error {
if newTrie == nil {
return nil
}
if newTrie.PathLength() != f.pathByteSize {
return fmt.Errorf("forest has path length %d, but new trie has path length %d", f.pathByteSize, newTrie.PathLength())
}
// TODO: check Thread safety
// TODO what is this string root hash
hashString := newTrie.StringRootHash()
if storedTrie, found := f.tries.Get(hashString); found {
foo := storedTrie.(*trie.MTrie)
if foo.Equals(newTrie) {
return nil
}
return fmt.Errorf("forest already contains a tree with same root hash but other properties")
}
f.tries.Add(hashString, newTrie)
f.metrics.ForestNumberOfTrees(uint64(f.tries.Len()))
return nil
}
// RemoveTrie removes a trie to the forest
func (f *Forest) RemoveTrie(rootHash []byte) {
// TODO remove from the file as well
encRootHash := hex.EncodeToString(rootHash)
f.tries.Remove(encRootHash)
f.metrics.ForestNumberOfTrees(uint64(f.tries.Len()))
}
// GetEmptyRootHash returns the rootHash of empty Trie
func (f *Forest) GetEmptyRootHash() []byte {
return trie.EmptyTrieRootHash(f.pathByteSize)
}
// MostRecentTouchedRootHash returns the rootHash of the most recently touched trie
func (f *Forest) MostRecentTouchedRootHash() ([]byte, error) {
keys := f.tries.Keys()
if len(keys) > 0 {
encodedRootHash := keys[len(keys)-1].(string)
return hex.DecodeString(encodedRootHash)
}
return nil, fmt.Errorf("no trie is stored in the forest")
}
// Size returns the number of active tries in this store
func (f *Forest) Size() int {
return f.tries.Len()
}