Make pinset sharding deterministic
Making this deterministic keeps us from creating an exponential amount of objects as the number of pins in the set increases. License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com>
Jeromy committed
Jan 27, 2017 at 17:19 UTC
728ff6dd2166cb780cfb3a2cbc83de196ed8ff48
2 files changed
+55
-33
pin/set.go
+5
-18
@@ -3,7 +3,6 @@ package pin
3
import (
4
"bytes"
5
"context"
6
- "crypto/rand"
6
"encoding/binary"
7
"errors"
8
"fmt"
@@ -26,14 +25,6 @@ const (
25
maxItems = 8192
26
)
27
29
-func randomSeed() (uint32, error) {
30
- var buf [4]byte
31
- if _, err := rand.Read(buf[:]); err != nil {
32
- return 0, err
33
- }
34
- return binary.LittleEndian.Uint32(buf[:]), nil
35
-}
36
-
28
func hash(seed uint32, c *cid.Cid) uint32 {
29
var buf [4]byte
30
binary.LittleEndian.PutUint32(buf[:], seed)
@@ -63,11 +54,7 @@ func (s sortByHash) Swap(a, b int) {
54
s.links[a], s.links[b] = s.links[b], s.links[a]
55
}
56
66
-func storeItems(ctx context.Context, dag merkledag.DAGService, estimatedLen uint64, iter itemIterator, internalKeys keyObserver) (*merkledag.ProtoNode, error) {
67
- seed, err := randomSeed()
68
- if err != nil {
69
- return nil, err
70
- }
57
+func storeItems(ctx context.Context, dag merkledag.DAGService, estimatedLen uint64, depth uint32, iter itemIterator, internalKeys keyObserver) (*merkledag.ProtoNode, error) {
58
links := make([]*node.Link, 0, defaultFanout+maxItems)
59
for i := 0; i < defaultFanout; i++ {
60
links = append(links, &node.Link{Cid: emptyKey})
@@ -82,7 +69,7 @@ func storeItems(ctx context.Context, dag merkledag.DAGService, estimatedLen uint
69
hdr := &pb.Set{
70
Version: proto.Uint32(1),
71
Fanout: proto.Uint32(defaultFanout),
85
- Seed: proto.Uint32(seed),
72
+ Seed: proto.Uint32(depth),
73
}
74
if err := writeHdr(n, hdr); err != nil {
75
return nil, err
@@ -129,7 +116,7 @@ func storeItems(ctx context.Context, dag merkledag.DAGService, estimatedLen uint
116
if !ok {
117
break
118
}
132
- h := hash(seed, k) % defaultFanout
119
+ h := hash(depth, k) % defaultFanout
120
hashed[h] = append(hashed[h], k)
121
}
122
@@ -142,7 +129,7 @@ func storeItems(ctx context.Context, dag merkledag.DAGService, estimatedLen uint
129
childIter := getCidListIterator(items)
130
131
// recursively create a pinset from the items for this bucket index
145
- child, err := storeItems(ctx, dag, uint64(len(items)), childIter, internalKeys)
132
+ child, err := storeItems(ctx, dag, uint64(len(items)), depth+1, childIter, internalKeys)
133
if err != nil {
134
return nil, err
135
}
@@ -296,7 +283,7 @@ func getCidListIterator(cids []*cid.Cid) itemIterator {
283
func storeSet(ctx context.Context, dag merkledag.DAGService, cids []*cid.Cid, internalKeys keyObserver) (*merkledag.ProtoNode, error) {
284
iter := getCidListIterator(cids)
285
299
- n, err := storeItems(ctx, dag, uint64(len(cids)), iter, internalKeys)
286
+ n, err := storeItems(ctx, dag, uint64(len(cids)), 0, iter, internalKeys)
287
if err != nil {
288
return nil, err
289
}
pin/set_test.go
+50
-15
@@ -2,40 +2,75 @@ package pin
2
3
import (
4
"context"
5
- "fmt"
6
- "os"
5
+ "encoding/binary"
6
"testing"
7
8
+ blockstore "github.com/ipfs/go-ipfs/blocks/blockstore"
9
+ bserv "github.com/ipfs/go-ipfs/blockservice"
10
+ offline "github.com/ipfs/go-ipfs/exchange/offline"
11
dag "github.com/ipfs/go-ipfs/merkledag"
10
- mdtest "github.com/ipfs/go-ipfs/merkledag/test"
12
13
+ ds "gx/ipfs/QmRWDav6mzWseLWeYfVd5fvUKiVe9xNH29YfMF438fG364/go-datastore"
14
+ dsq "gx/ipfs/QmRWDav6mzWseLWeYfVd5fvUKiVe9xNH29YfMF438fG364/go-datastore/query"
15
cid "gx/ipfs/QmcTcsTvfaeEBRFo1TkFgT8sRmgi1n1LTZpecfVP8fzpGD/go-cid"
16
)
17
18
func ignoreCids(_ *cid.Cid) {}
19
17
-func TestSet(t *testing.T) {
18
- ds := mdtest.Mock()
19
- limit := 10000 // 10000 reproduces the pinloss issue fairly reliably
20
-
21
- if os.Getenv("STRESS_IT_OUT_YO") != "" {
22
- limit = 10000000
20
+func objCount(d ds.Datastore) int {
21
+ q := dsq.Query{KeysOnly: true}
22
+ res, err := d.Query(q)
23
+ if err != nil {
24
+ panic(err)
25
}
24
- var inputs []*cid.Cid
25
- for i := 0; i < limit; i++ {
26
- c, err := ds.Add(dag.NodeWithData([]byte(fmt.Sprint(i))))
27
- if err != nil {
28
- t.Fatal(err)
26
+
27
+ var count int
28
+ for {
29
+ _, ok := res.NextSync()
30
+ if !ok {
31
+ break
32
}
33
34
+ count++
35
+ }
36
+ return count
37
+}
38
+
39
+func TestSet(t *testing.T) {
40
+ dst := ds.NewMapDatastore()
41
+ bstore := blockstore.NewBlockstore(dst)
42
+ ds := dag.NewDAGService(bserv.New(bstore, offline.Exchange(bstore)))
43
+
44
+ // this value triggers the creation of a recursive shard.
45
+ // If the recursive sharding is done improperly, this will result in
46
+ // an infinite recursion and crash (OOM)
47
+ limit := uint32((defaultFanout * maxItems) + 1)
48
+
49
+ var inputs []*cid.Cid
50
+ buf := make([]byte, 4)
51
+ for i := uint32(0); i < limit; i++ {
52
+ binary.BigEndian.PutUint32(buf, i)
53
+ c := dag.NewRawNode(buf).Cid()
54
inputs = append(inputs, c)
55
}
56
57
+ _, err := storeSet(context.Background(), ds, inputs[:len(inputs)-1], ignoreCids)
58
+ if err != nil {
59
+ t.Fatal(err)
60
+ }
61
+
62
+ objs1 := objCount(dst)
63
+
64
out, err := storeSet(context.Background(), ds, inputs, ignoreCids)
65
if err != nil {
66
t.Fatal(err)
67
}
68
69
+ objs2 := objCount(dst)
70
+ if objs2-objs1 > 2 {
71
+ t.Fatal("set sharding does not appear to be deterministic")
72
+ }
73
+
74
// weird wrapper node because loadSet expects us to pass an
75
// object pointing to multiple named sets
76
setroot := &dag.ProtoNode{}
@@ -49,7 +84,7 @@ func TestSet(t *testing.T) {
84
t.Fatal(err)
85
}
86
52
- if len(outset) != limit {
87
+ if uint32(len(outset)) != limit {
88
t.Fatal("got wrong number", len(outset), limit)
89
}
90