pin: use separate dagservice for storing pinsets
License: MIT Signed-off-by: Jeromy <why@ipfs.io>
Jeromy committed
Aug 19, 2016 at 17:45 UTC
db228e1054f1b26b8d45eb860ef4106303f89e4f
4 files changed
+23
-41
core/builder.go
+4
-2
@@ -171,13 +171,15 @@ func setupNode(ctx context.Context, n *IpfsNode, cfg *BuildCfg) error {
171
172
n.Blocks = bserv.New(n.Blockstore, n.Exchange)
173
n.DAG = dag.NewDAGService(n.Blocks)
174
- n.Pinning, err = pin.LoadPinner(n.Repo.Datastore(), n.DAG)
174
+
175
+ internalDag := dag.NewDAGService(bserv.New(n.Blockstore, offline.Exchange(n.Blockstore)))
176
+ n.Pinning, err = pin.LoadPinner(n.Repo.Datastore(), n.DAG, internalDag)
177
if err != nil {
178
// TODO: we should move towards only running 'NewPinner' explicity on
179
// node init instead of implicitly here as a result of the pinner keys
180
// not being found in the datastore.
181
// this is kinda sketchy and could cause data loss
180
- n.Pinning = pin.NewPinner(n.Repo.Datastore(), n.DAG)
182
+ n.Pinning = pin.NewPinner(n.Repo.Datastore(), n.DAG, internalDag)
183
}
184
n.Resolver = &path.Resolver{DAG: n.DAG}
185
merkledag/merkledag_test.go
+2
-23
@@ -10,7 +10,6 @@ import (
10
"sync"
11
"testing"
12
13
- bstore "github.com/ipfs/go-ipfs/blocks/blockstore"
13
key "github.com/ipfs/go-ipfs/blocks/key"
14
bserv "github.com/ipfs/go-ipfs/blockservice"
15
bstest "github.com/ipfs/go-ipfs/blockservice/test"
@@ -19,31 +18,11 @@ import (
18
chunk "github.com/ipfs/go-ipfs/importer/chunk"
19
. "github.com/ipfs/go-ipfs/merkledag"
20
dstest "github.com/ipfs/go-ipfs/merkledag/test"
22
- "github.com/ipfs/go-ipfs/pin"
21
uio "github.com/ipfs/go-ipfs/unixfs/io"
24
- ds "gx/ipfs/QmTxLSvdhwg68WJimdS6icLPhZi28aTp6b7uihC2Yb47Xk/go-datastore"
25
- dssync "gx/ipfs/QmTxLSvdhwg68WJimdS6icLPhZi28aTp6b7uihC2Yb47Xk/go-datastore/sync"
22
u "gx/ipfs/QmZNVWh8LLjAavuQ2JXuFmuYH3C11xo988vSgp7UQrTRj1/go-ipfs-util"
23
"gx/ipfs/QmZy2y8t9zQH2a1b8q2ZSLKp17ATuJoCNxxyMFG5qFExpt/go-net/context"
24
)
25
30
-type dagservAndPinner struct {
31
- ds DAGService
32
- mp pin.Pinner
33
-}
34
-
35
-func getDagservAndPinner(t *testing.T) dagservAndPinner {
36
- db := dssync.MutexWrap(ds.NewMapDatastore())
37
- bs := bstore.NewBlockstore(db)
38
- blockserv := bserv.New(bs, offline.Exchange(bs))
39
- dserv := NewDAGService(blockserv)
40
- mpin := pin.NewPinner(db, dserv)
41
- return dagservAndPinner{
42
- ds: dserv,
43
- mp: mpin,
44
- }
45
-}
46
-
26
func TestNode(t *testing.T) {
27
28
n1 := NodeWithData([]byte("beep"))
@@ -254,7 +233,7 @@ func TestEmptyKey(t *testing.T) {
233
}
234
235
func TestCantGet(t *testing.T) {
257
- dsp := getDagservAndPinner(t)
236
+ ds := dstest.Mock()
237
a := NodeWithData([]byte("A"))
238
239
k, err := a.Key()
@@ -262,7 +241,7 @@ func TestCantGet(t *testing.T) {
241
t.Fatal(err)
242
}
243
265
- _, err = dsp.ds.Get(context.Background(), k)
244
+ _, err = ds.Get(context.Background(), k)
245
if !strings.Contains(err.Error(), "not found") {
246
t.Fatal("expected err not found, got: ", err)
247
}
pin/pin.go
+12
-11
@@ -110,15 +110,14 @@ type pinner struct {
110
// not delete them.
111
internalPin map[key.Key]struct{}
112
dserv mdag.DAGService
113
+ internal mdag.DAGService // dagservice used to store internal objects
114
dstore ds.Datastore
115
}
116
117
// NewPinner creates a new pinner using the given datastore as a backend
117
-func NewPinner(dstore ds.Datastore, serv mdag.DAGService) Pinner {
118
+func NewPinner(dstore ds.Datastore, serv, internal mdag.DAGService) Pinner {
119
119
- // Load set from given datastore...
120
rcset := set.NewSimpleBlockSet()
121
-
121
dirset := set.NewSimpleBlockSet()
122
123
return &pinner{
@@ -126,6 +125,7 @@ func NewPinner(dstore ds.Datastore, serv mdag.DAGService) Pinner {
125
directPin: dirset,
126
dserv: serv,
127
dstore: dstore,
128
+ internal: internal,
129
}
130
}
131
@@ -344,7 +344,7 @@ func (p *pinner) RemovePinWithMode(key key.Key, mode PinMode) {
344
}
345
346
// LoadPinner loads a pinner and its keysets from the given datastore
347
-func LoadPinner(d ds.Datastore, dserv mdag.DAGService) (Pinner, error) {
347
+func LoadPinner(d ds.Datastore, dserv, internal mdag.DAGService) (Pinner, error) {
348
p := new(pinner)
349
350
rootKeyI, err := d.Get(pinDatastoreKey)
@@ -361,7 +361,7 @@ func LoadPinner(d ds.Datastore, dserv mdag.DAGService) (Pinner, error) {
361
ctx, cancel := context.WithTimeout(context.TODO(), time.Second*5)
362
defer cancel()
363
364
- root, err := dserv.Get(ctx, rootKey)
364
+ root, err := internal.Get(ctx, rootKey)
365
if err != nil {
366
return nil, fmt.Errorf("cannot find pinning root object: %v", err)
367
}
@@ -374,7 +374,7 @@ func LoadPinner(d ds.Datastore, dserv mdag.DAGService) (Pinner, error) {
374
}
375
376
{ // load recursive set
377
- recurseKeys, err := loadSet(ctx, dserv, root, linkRecursive, recordInternal)
377
+ recurseKeys, err := loadSet(ctx, internal, root, linkRecursive, recordInternal)
378
if err != nil {
379
return nil, fmt.Errorf("cannot load recursive pins: %v", err)
380
}
@@ -382,7 +382,7 @@ func LoadPinner(d ds.Datastore, dserv mdag.DAGService) (Pinner, error) {
382
}
383
384
{ // load direct set
385
- directKeys, err := loadSet(ctx, dserv, root, linkDirect, recordInternal)
385
+ directKeys, err := loadSet(ctx, internal, root, linkDirect, recordInternal)
386
if err != nil {
387
return nil, fmt.Errorf("cannot load direct pins: %v", err)
388
}
@@ -394,6 +394,7 @@ func LoadPinner(d ds.Datastore, dserv mdag.DAGService) (Pinner, error) {
394
// assign services
395
p.dserv = dserv
396
p.dstore = d
397
+ p.internal = internal
398
399
return p, nil
400
}
@@ -422,7 +423,7 @@ func (p *pinner) Flush() error {
423
424
root := &mdag.Node{}
425
{
425
- n, err := storeSet(ctx, p.dserv, p.directPin.GetKeys(), recordInternal)
426
+ n, err := storeSet(ctx, p.internal, p.directPin.GetKeys(), recordInternal)
427
if err != nil {
428
return err
429
}
@@ -432,7 +433,7 @@ func (p *pinner) Flush() error {
433
}
434
435
{
435
- n, err := storeSet(ctx, p.dserv, p.recursePin.GetKeys(), recordInternal)
436
+ n, err := storeSet(ctx, p.internal, p.recursePin.GetKeys(), recordInternal)
437
if err != nil {
438
return err
439
}
@@ -442,12 +443,12 @@ func (p *pinner) Flush() error {
443
}
444
445
// add the empty node, its referenced by the pin sets but never created
445
- _, err := p.dserv.Add(new(mdag.Node))
446
+ _, err := p.internal.Add(new(mdag.Node))
447
if err != nil {
448
return err
449
}
450
450
- k, err := p.dserv.Add(root)
451
+ k, err := p.internal.Add(root)
452
if err != nil {
453
return err
454
}
pin/pin_test.go
+5
-5
@@ -45,7 +45,7 @@ func TestPinnerBasic(t *testing.T) {
45
dserv := mdag.NewDAGService(bserv)
46
47
// TODO does pinner need to share datastore with blockservice?
48
- p := NewPinner(dstore, dserv)
48
+ p := NewPinner(dstore, dserv, dserv)
49
50
a, ak := randNode()
51
_, err := dserv.Add(a)
@@ -133,7 +133,7 @@ func TestPinnerBasic(t *testing.T) {
133
t.Fatal(err)
134
}
135
136
- np, err := LoadPinner(dstore, dserv)
136
+ np, err := LoadPinner(dstore, dserv, dserv)
137
if err != nil {
138
t.Fatal(err)
139
}
@@ -154,7 +154,7 @@ func TestDuplicateSemantics(t *testing.T) {
154
dserv := mdag.NewDAGService(bserv)
155
156
// TODO does pinner need to share datastore with blockservice?
157
- p := NewPinner(dstore, dserv)
157
+ p := NewPinner(dstore, dserv, dserv)
158
159
a, _ := randNode()
160
_, err := dserv.Add(a)
@@ -187,7 +187,7 @@ func TestFlush(t *testing.T) {
187
bserv := bs.New(bstore, offline.Exchange(bstore))
188
189
dserv := mdag.NewDAGService(bserv)
190
- p := NewPinner(dstore, dserv)
190
+ p := NewPinner(dstore, dserv, dserv)
191
_, k := randNode()
192
193
p.PinWithMode(k, Recursive)
@@ -204,7 +204,7 @@ func TestPinRecursiveFail(t *testing.T) {
204
bserv := bs.New(bstore, offline.Exchange(bstore))
205
dserv := mdag.NewDAGService(bserv)
206
207
- p := NewPinner(dstore, dserv)
207
+ p := NewPinner(dstore, dserv, dserv)
208
209
a, _ := randNode()
210
b, _ := randNode()