arbo/addbatch.go
addbatch.goRaw
1package arbo
2
3import (
4 "bytes"
5 "fmt"
6 "math"
7 "sort"
8)
9
10/*
11
12AddBatch design
13===============
14
15
16CASE A: Empty Tree --> if tree is empty (root==0)
17=================================================
18- Build the full tree from bottom to top (from all the leaf to the root)
19
20
21CASE B: ALMOST CASE A, Almost empty Tree --> if Tree has numLeafs < minLeafsThreshold
22==============================================================================
23- Get the Leafs (key & value) (iterate the tree from the current root getting
24the leafs)
25- Create a new empty Tree
26- Do CASE A for the new Tree, giving the already existing key&values (leafs)
27from the original Tree + the new key&values to be added from the AddBatch call
28
29
30 R R
31 / \ / \
32 A * / \
33 / \ / \
34 B C * *
35 / | / \
36 / | / \
37 / | / \
38 L: A B G D
39 / \
40 / \
41 / \
42 C *
43 / \
44 / \
45 / \
46 ... ... (nLeafs < minLeafsThreshold)
47
48
49CASE C: ALMOST CASE B --> if Tree has few Leafs (but numLeafs>=minLeafsThreshold)
50==============================================================================
51- Use A, B, G, F as Roots of subtrees
52- Do CASE B for each subtree
53- Then go from L to the Root
54
55 R
56 / \
57 / \
58 / \
59 * *
60 / | / \
61 / | / \
62 / | / \
63L: A B G D
64 / \
65 / \
66 / \
67 C *
68 / \
69 / \
70 / \
71 ... ... (nLeafs >= minLeafsThreshold)
72
73
74
75CASE D: Already populated Tree
76==============================
77- Use A, B, C, D as subtree
78- Sort the Keys in Buckets that share the initial part of the path
79- For each subtree add there the new leafs
80
81 R
82 / \
83 / \
84 / \
85 * *
86 / | / \
87 / | / \
88 / | / \
89L: A B C D
90 /\ /\ / \ / \
91 ... ... ... ... ... ...
92
93
94CASE E: Already populated Tree Unbalanced
95=========================================
96- Need to fill M1 and M2, and then will be able to use CASE D
97 - Search for M1 & M2 in the inputed Keys
98 - Add M1 & M2 to the Tree
99 - From here can use CASE D
100
101 R
102 / \
103 / \
104 / \
105 * *
106 | \
107 | \
108 | \
109L: M1 * M2 * (where M1 and M2 are empty)
110 / | /
111 / | /
112 / | /
113 A * *
114 / \ | \
115 / \ | \
116 / \ | \
117 B * * C
118 / \ |\
119 ... ... | \
120 | \
121 D E
122
123
124
125Algorithm decision
126==================
127- if nLeafs==0 (root==0): CASE A
128- if nLeafs<minLeafsThreshold: CASE B
129- if nLeafs>=minLeafsThreshold && (nLeafs/nBuckets) < minLeafsThreshold: CASE C
130- else: CASE D & CASE E
131
132
133- Multiple tree.Add calls: O(n log n)
134 - Used in: cases A, B, C
135- Tree from bottom to top: O(log n)
136 - Used in: cases D, E
137
138*/
139
140const (
141 minLeafsThreshold = uint64(100) // nolint:gomnd // TMP WIP this will be autocalculated
142 nBuckets = uint64(4) // TMP WIP this will be autocalculated from
143)
144
145// AddBatchOpt is the WIP implementation of the AddBatch method in a more
146// optimized approach.
147func (t *Tree) AddBatchOpt(keys, values [][]byte) ([]int, error) {
148 t.updateAccessTime()
149 t.Lock()
150 defer t.Unlock()
151
152 // TODO if len(keys) is not a power of 2, add padding of empty
153 // keys&values. Maybe when len(keyvalues) is not a power of 2, cut at
154 // the biggest power of 2 under the len(keys), add those 2**n key-values
155 // using the AddBatch approach, and then add the remaining key-values
156 // using tree.Add.
157
158 kvs, err := t.keysValuesToKvs(keys, values)
159 if err != nil {
160 return nil, err
161 }
162
163 t.tx, err = t.db.NewTx() // TODO add t.tx.Commit()
164 if err != nil {
165 return nil, err
166 }
167
168 // CASE A: if nLeafs==0 (root==0)
169 if bytes.Equal(t.root, t.emptyHash) {
170 // sort keys & values by path
171 sortKvs(kvs)
172 return t.buildTreeBottomUp(kvs)
173 }
174
175 // CASE B: if nLeafs<nBuckets
176 nLeafs, err := t.GetNLeafs()
177 if err != nil {
178 return nil, err
179 }
180 if nLeafs < minLeafsThreshold { // CASE B
181 invalids, excedents, err := t.caseB(0, kvs)
182 if err != nil {
183 return nil, err
184 }
185 // add the excedents
186 for i := 0; i < len(excedents); i++ {
187 err = t.add(0, excedents[i].k, excedents[i].v)
188 if err != nil {
189 invalids = append(invalids, excedents[i].pos)
190 }
191 }
192 return invalids, nil
193 }
194
195 // CASE C: if nLeafs>=minLeafsThreshold && (nLeafs/nBuckets) < minLeafsThreshold
196 // available parallelization, will need to be a power of 2 (2**n)
197 var excedents []kv
198 l := int(math.Log2(float64(nBuckets)))
199 if nLeafs >= minLeafsThreshold && (nLeafs/nBuckets) < minLeafsThreshold {
200 // TODO move to own function
201 // 1. go down until level L (L=log2(nBuckets))
202 keysAtL, err := t.getKeysAtLevel(l + 1)
203 if err != nil {
204 return nil, err
205 }
206
207 buckets := splitInBuckets(kvs, nBuckets)
208
209 // 2. use keys at level L as roots of the subtrees under each one
210 var subRoots [][]byte
211 // TODO parallelize
212 for i := 0; i < len(keysAtL); i++ {
213 bucketTree := Tree{tx: t.tx, db: t.db, maxLevels: t.maxLevels,
214 hashFunction: t.hashFunction, root: keysAtL[i]}
215
216 // 3. and do CASE B for each
217 _, bucketExcedents, err := bucketTree.caseB(l, buckets[i])
218 if err != nil {
219 return nil, err
220 }
221 excedents = append(excedents, bucketExcedents...)
222 subRoots = append(subRoots, bucketTree.root)
223 }
224 // 4. go upFromKeys from the new roots of the subtrees
225 newRoot, err := t.upFromKeys(subRoots)
226 if err != nil {
227 return nil, err
228 }
229 t.root = newRoot
230
231 var invalids []int
232 for i := 0; i < len(excedents); i++ {
233 // Add until the level L
234 err = t.add(0, excedents[i].k, excedents[i].v)
235 if err != nil {
236 invalids = append(invalids, excedents[i].pos) // TODO WIP
237 }
238 }
239
240 return invalids, nil
241 }
242
243 // TODO store t.root into DB
244 // TODO update NLeafs from DB
245
246 return nil, fmt.Errorf("UNIMPLEMENTED")
247}
248
249func (t *Tree) caseB(l int, kvs []kv) ([]int, []kv, error) {
250 // get already existing keys
251 aKs, aVs, err := t.getLeafs(t.root)
252 if err != nil {
253 return nil, nil, err
254 }
255 aKvs, err := t.keysValuesToKvs(aKs, aVs)
256 if err != nil {
257 return nil, nil, err
258 }
259 // add already existing key-values to the inputted key-values
260 kvs = append(kvs, aKvs...)
261
262 // proceed with CASE A
263 sortKvs(kvs)
264
265 // cutPowerOfTwo, the excedent add it as normal Tree.Add
266 kvsP2, kvsNonP2 := cutPowerOfTwo(kvs)
267 invalids, err := t.buildTreeBottomUp(kvsP2)
268 if err != nil {
269 return nil, nil, err
270 }
271 // return the excedents which will be added at the full tree at the end
272 return invalids, kvsNonP2, nil
273}
274
275func splitInBuckets(kvs []kv, nBuckets uint64) [][]kv {
276 buckets := make([][]kv, nBuckets)
277 // 1. classify the keyvalues into buckets
278 for i := 0; i < len(kvs); i++ {
279 pair := kvs[i]
280
281 bucketnum := keyToBucket(pair.k, int(nBuckets))
282 buckets[bucketnum] = append(buckets[bucketnum], pair)
283 }
284 return buckets
285}
286
287// TODO rename in a more 'real' name (calculate bucket from/for key)
288func keyToBucket(k []byte, nBuckets int) int {
289 nLevels := int(math.Log2(float64(nBuckets)))
290 b := make([]int, nBuckets)
291 for i := 0; i < nBuckets; i++ {
292 b[i] = i
293 }
294 r := b
295 mid := len(r) / 2 //nolint:gomnd
296 for i := 0; i < nLevels; i++ {
297 if int(k[i/8]&(1<<(i%8))) != 0 {
298 r = r[mid:]
299 mid = len(r) / 2 //nolint:gomnd
300 } else {
301 r = r[:mid]
302 mid = len(r) / 2 //nolint:gomnd
303 }
304 }
305 return r[0]
306}
307
308type kv struct {
309 pos int // original position in the array
310 keyPath []byte
311 k []byte
312 v []byte
313}
314
315// compareBytes compares byte slices where the bytes are compared from left to
316// right and each byte is compared by bit from right to left
317func compareBytes(a, b []byte) bool {
318 // WIP
319 for i := 0; i < len(a); i++ {
320 for j := 0; j < 8; j++ {
321 aBit := a[i] & (1 << j)
322 bBit := b[i] & (1 << j)
323 if aBit > bBit {
324 return false
325 } else if aBit < bBit {
326 return true
327 }
328 }
329 }
330 return false
331}
332
333// sortKvs sorts the kv by path
334func sortKvs(kvs []kv) {
335 sort.Slice(kvs, func(i, j int) bool {
336 return compareBytes(kvs[i].keyPath, kvs[j].keyPath)
337 })
338}
339
340func (t *Tree) keysValuesToKvs(ks, vs [][]byte) ([]kv, error) {
341 if len(ks) != len(vs) {
342 return nil, fmt.Errorf("len(keys)!=len(values) (%d!=%d)",
343 len(ks), len(vs))
344 }
345 kvs := make([]kv, len(ks))
346 for i := 0; i < len(ks); i++ {
347 keyPath := make([]byte, t.hashFunction.Len())
348 copy(keyPath[:], ks[i])
349 kvs[i].pos = i
350 kvs[i].keyPath = ks[i]
351 kvs[i].k = ks[i]
352 kvs[i].v = vs[i]
353 }
354
355 return kvs, nil
356}
357
358/*
359func (t *Tree) kvsToKeysValues(kvs []kv) ([][]byte, [][]byte) {
360 ks := make([][]byte, len(kvs))
361 vs := make([][]byte, len(kvs))
362 for i := 0; i < len(kvs); i++ {
363 ks[i] = kvs[i].k
364 vs[i] = kvs[i].v
365 }
366 return ks, vs
367}
368*/
369
370// keys & values must be sorted by path, and the array ks must be length
371// multiple of 2
372// TODO return index of failed keyvaules
373func (t *Tree) buildTreeBottomUp(kvs []kv) ([]int, error) {
374 // build the leafs
375 leafKeys := make([][]byte, len(kvs))
376 for i := 0; i < len(kvs); i++ {
377 // TODO handle the case where Key&Value == 0
378 leafKey, leafValue, err := newLeafValue(t.hashFunction, kvs[i].k, kvs[i].v)
379 if err != nil {
380 return nil, err
381 }
382 // store leafKey & leafValue to db
383 if err := t.tx.Put(leafKey, leafValue); err != nil {
384 return nil, err
385 }
386 leafKeys[i] = leafKey
387 }
388 // TODO parallelize t.upFromKeys until level log2(nBuckets) is reached
389 r, err := t.upFromKeys(leafKeys)
390 if err != nil {
391 return nil, err
392 }
393 t.root = r
394 return nil, nil
395}
396
397// keys & values must be sorted by path, and the array ks must be length
398// multiple of 2
399func (t *Tree) upFromKeys(ks [][]byte) ([]byte, error) {
400 if len(ks) == 1 {
401 return ks[0], nil
402 }
403
404 var rKs [][]byte
405 for i := 0; i < len(ks); i += 2 {
406 // TODO handle the case where Key&Value == 0
407 k, v, err := newIntermediate(t.hashFunction, ks[i], ks[i+1])
408 if err != nil {
409 return nil, err
410 }
411 // store k-v to db
412 if err = t.tx.Put(k, v); err != nil {
413 return nil, err
414 }
415 rKs = append(rKs, k)
416 }
417 return t.upFromKeys(rKs)
418}
419
420func (t *Tree) getLeafs(root []byte) ([][]byte, [][]byte, error) {
421 var ks, vs [][]byte
422 err := t.iter(root, func(k, v []byte) {
423 if v[0] != PrefixValueLeaf {
424 return
425 }
426 leafK, leafV := readLeafValue(v)
427 ks = append(ks, leafK)
428 vs = append(vs, leafV)
429 })
430 return ks, vs, err
431}
432
433func (t *Tree) getKeysAtLevel(l int) ([][]byte, error) {
434 var keys [][]byte
435 err := t.iterWithStop(t.root, 0, func(currLvl int, k, v []byte) bool {
436 if currLvl == l {
437 keys = append(keys, k)
438 }
439 if currLvl >= l {
440 return true // to stop the iter from going down
441 }
442 return false
443 })
444
445 return keys, err
446}
447
448// cutPowerOfTwo returns []kv of length that is a power of 2, and a second []kv
449// with the extra elements that don't fit in a power of 2 length
450func cutPowerOfTwo(kvs []kv) ([]kv, []kv) {
451 x := len(kvs)
452 if (x & (x - 1)) != 0 {
453 p2 := highestPowerOfTwo(x)
454 return kvs[:p2], kvs[p2:]
455 }
456 return kvs, nil
457}
458
459func highestPowerOfTwo(n int) int {
460 res := 0
461 for i := n; i >= 1; i-- {
462 if (i & (i - 1)) == 0 {
463 res = i
464 break
465 }
466 }
467 return res
468}
469
470// func computeSimpleAddCost(nLeafs int) int {
471// // nLvls 2^nLvls
472// nLvls := int(math.Log2(float64(nLeafs)))
473// return nLvls * int(math.Pow(2, float64(nLvls)))
474// }
475//
476// func computeBottomUpAddCost(nLeafs int) int {
477// // 2^nLvls * 2 - 1
478// nLvls := int(math.Log2(float64(nLeafs)))
479// return (int(math.Pow(2, float64(nLvls))) * 2) - 1
480// }