merkledag: switch to new dag interface
Also: * Update the blockstore/blockservice methods to match. * Construct a new temporary offline dag instead of having a GetOfflineLinkService method. License: MIT Signed-off-by: Steven Allen <steven@stebalien.com>
Steven Allen committed
Jan 25, 2018 at 12:21 UTC
d154b4a990d14126227b68d21274191046a186b5
68 files changed
+493
-655
blockservice/blockservice.go
+39
-33
@@ -7,6 +7,7 @@ import (
7
"context"
8
"errors"
9
"fmt"
10
+ "io"
11
12
"github.com/ipfs/go-ipfs/blocks/blockstore"
13
exchange "github.com/ipfs/go-ipfs/exchange"
@@ -21,10 +22,27 @@ var log = logging.Logger("blockservice")
22
23
var ErrNotFound = errors.New("blockservice: key not found")
24
25
+type BlockGetter interface {
26
+ // GetBlock gets the requested block.
27
+ GetBlock(ctx context.Context, c *cid.Cid) (blocks.Block, error)
28
+
29
+ // GetBlocks does a batch request for the given cids, returning blocks as
30
+ // they are found, in no particular order.
31
+ //
32
+ // It may not be able to find all requested blocks (or the context may
33
+ // be canceled). In that case, it will close the channel early. It is up
34
+ // to the consumer to detect this situation and keep track which blocks
35
+ // it has received and which it hasn't.
36
+ GetBlocks(ctx context.Context, ks []*cid.Cid) <-chan blocks.Block
37
+}
38
+
39
// BlockService is a hybrid block datastore. It stores data in a local
40
// datastore and may retrieve data from a remote Exchange.
41
// It uses an internal `datastore.Datastore` instance to store values.
42
type BlockService interface {
43
+ io.Closer
44
+ BlockGetter
45
+
46
// Blockstore returns a reference to the underlying blockstore
47
Blockstore() blockstore.Blockstore
48
@@ -32,20 +50,14 @@ type BlockService interface {
50
Exchange() exchange.Interface
51
52
// AddBlock puts a given block to the underlying datastore
35
- AddBlock(o blocks.Block) (*cid.Cid, error)
53
+ AddBlock(o blocks.Block) error
54
55
// AddBlocks adds a slice of blocks at the same time using batching
56
// capabilities of the underlying datastore whenever possible.
39
- AddBlocks(bs []blocks.Block) ([]*cid.Cid, error)
40
-
41
- GetBlock(ctx context.Context, c *cid.Cid) (blocks.Block, error)
42
- DeleteBlock(o blocks.Block) error
57
+ AddBlocks(bs []blocks.Block) error
58
44
- // GetBlocks does a batch request for the given cids, returning blocks as
45
- // they are found, in no particular order.
46
- GetBlocks(ctx context.Context, ks []*cid.Cid) <-chan blocks.Block
47
-
48
- Close() error
59
+ // DeleteBlock deletes the given block from the blockservice.
60
+ DeleteBlock(o *cid.Cid) error
61
}
62
63
type blockService struct {
@@ -110,38 +122,34 @@ func NewSession(ctx context.Context, bs BlockService) *Session {
122
123
// AddBlock adds a particular block to the service, Putting it into the datastore.
124
// TODO pass a context into this if the remote.HasBlock is going to remain here.
113
-func (s *blockService) AddBlock(o blocks.Block) (*cid.Cid, error) {
125
+func (s *blockService) AddBlock(o blocks.Block) error {
126
c := o.Cid()
127
if s.checkFirst {
116
- has, err := s.blockstore.Has(c)
117
- if err != nil {
118
- return nil, err
119
- }
120
-
121
- if has {
122
- return c, nil
128
+ if has, err := s.blockstore.Has(c); has || err != nil {
129
+ return err
130
}
131
}
132
126
- err := s.blockstore.Put(o)
127
- if err != nil {
128
- return nil, err
133
+ if err := s.blockstore.Put(o); err != nil {
134
+ return err
135
}
136
137
if err := s.exchange.HasBlock(o); err != nil {
132
- return nil, errors.New("blockservice is closed")
138
+ // TODO(stebalien): really an error?
139
+ return errors.New("blockservice is closed")
140
}
141
135
- return c, nil
142
+ return nil
143
}
144
138
-func (s *blockService) AddBlocks(bs []blocks.Block) ([]*cid.Cid, error) {
145
+func (s *blockService) AddBlocks(bs []blocks.Block) error {
146
var toput []blocks.Block
147
if s.checkFirst {
148
+ toput = make([]blocks.Block, 0, len(bs))
149
for _, b := range bs {
150
has, err := s.blockstore.Has(b.Cid())
151
if err != nil {
144
- return nil, err
152
+ return err
153
}
154
if !has {
155
toput = append(toput, b)
@@ -153,18 +161,16 @@ func (s *blockService) AddBlocks(bs []blocks.Block) ([]*cid.Cid, error) {
161
162
err := s.blockstore.PutMany(toput)
163
if err != nil {
156
- return nil, err
164
+ return err
165
}
166
159
- var ks []*cid.Cid
167
for _, o := range toput {
168
if err := s.exchange.HasBlock(o); err != nil {
162
- return nil, fmt.Errorf("blockservice is closed (%s)", err)
169
+ // TODO(stebalien): Should this really *return*?
170
+ return fmt.Errorf("blockservice is closed (%s)", err)
171
}
164
-
165
- ks = append(ks, o.Cid())
172
}
167
- return ks, nil
173
+ return nil
174
}
175
176
// GetBlock retrieves a particular block from the service,
@@ -256,8 +262,8 @@ func getBlocks(ctx context.Context, ks []*cid.Cid, bs blockstore.Blockstore, f e
262
}
263
264
// DeleteBlock deletes a block in the blockservice from the datastore
259
-func (s *blockService) DeleteBlock(o blocks.Block) error {
260
- return s.blockstore.DeleteBlock(o.Cid())
265
+func (s *blockService) DeleteBlock(c *cid.Cid) error {
266
+ return s.blockstore.DeleteBlock(c)
267
}
268
269
func (s *blockService) Close() error {
blockservice/test/blocks_test.go
+1
-5
@@ -33,16 +33,12 @@ func TestBlocks(t *testing.T) {
33
t.Error("Block key and data multihash key not equal")
34
}
35
36
- k, err := bs.AddBlock(o)
36
+ err := bs.AddBlock(o)
37
if err != nil {
38
t.Error("failed to add block to BlockService", err)
39
return
40
}
41
42
- if !k.Equals(o.Cid()) {
43
- t.Error("returned key is not equal to block key", err)
44
- }
45
-
42
ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
43
defer cancel()
44
b2, err := bs.GetBlock(ctx, o.Cid())
core/commands/block.go
+2
-2
@@ -199,14 +199,14 @@ It reads from stdin, and <key> is a base58 encoded multihash.
199
return
200
}
201
202
- k, err := n.Blocks.AddBlock(b)
202
+ err = n.Blocks.AddBlock(b)
203
if err != nil {
204
res.SetError(err, cmdkit.ErrNormal)
205
return
206
}
207
208
err = cmds.EmitOnce(res, &BlockStat{
209
- Key: k.String(),
209
+ Key: b.Cid().String(),
210
Size: len(data),
211
})
212
if err != nil {
core/commands/dag/dag.go
+3
-2
@@ -18,6 +18,7 @@ import (
18
cid "gx/ipfs/QmcZfnkapfECQGcLZaf9B79NRg7cRa9EnZh4LSbkCzwNvY/go-cid"
19
cmdkit "gx/ipfs/QmceUdzxkimdYsgtX733uNgzf1DLHyBKN6ehGSp85ayppM/go-ipfs-cmdkit"
20
files "gx/ipfs/QmceUdzxkimdYsgtX733uNgzf1DLHyBKN6ehGSp85ayppM/go-ipfs-cmdkit/files"
21
+ node "gx/ipfs/Qme5bWv7wtjUNGsK2BNGVUFPKiuxWrsqrtvYwCLRw8YFES/go-ipld-format"
22
)
23
24
var log = logging.Logger("cmds/files")
@@ -102,7 +103,7 @@ into an object of the specified format.
103
104
addAllAndPin := func(f files.File) error {
105
cids := cid.NewSet()
105
- b := n.DAG.Batch()
106
+ b := node.NewBatch(req.Context(), n.DAG)
107
108
for {
109
file, err := f.NextFile()
@@ -122,7 +123,7 @@ into an object of the specified format.
123
}
124
125
for _, nd := range nds {
125
- _, err := b.Add(nd)
126
+ err := b.Add(nd)
127
if err != nil {
128
return err
129
}
core/commands/dht.go
+2
-1
@@ -19,6 +19,7 @@ import (
19
peer "gx/ipfs/Qma7H6RW8wRrfZpNSXwxYGcd1E149s42FpWNpDNieSVrnU/go-libp2p-peer"
20
cid "gx/ipfs/QmcZfnkapfECQGcLZaf9B79NRg7cRa9EnZh4LSbkCzwNvY/go-cid"
21
"gx/ipfs/QmceUdzxkimdYsgtX733uNgzf1DLHyBKN6ehGSp85ayppM/go-ipfs-cmdkit"
22
+ node "gx/ipfs/Qme5bWv7wtjUNGsK2BNGVUFPKiuxWrsqrtvYwCLRw8YFES/go-ipld-format"
23
pstore "gx/ipfs/QmeZVQzUrXqaszo24DAoHfGzcmCptN9JyngLkGAiEfk2x7/go-libp2p-peerstore"
24
ipdht "gx/ipfs/QmfChjky1VNaHUQR9F2xqR1QEyX45pqU78nhsoq5GDYoKL/go-libp2p-kad-dht"
25
)
@@ -377,7 +378,7 @@ func provideKeys(ctx context.Context, r routing.IpfsRouting, cids []*cid.Cid) er
378
return nil
379
}
380
380
-func provideKeysRec(ctx context.Context, r routing.IpfsRouting, dserv dag.DAGService, cids []*cid.Cid) error {
381
+func provideKeysRec(ctx context.Context, r routing.IpfsRouting, dserv node.DAGService, cids []*cid.Cid) error {
382
provided := cid.NewSet()
383
for _, c := range cids {
384
kset := cid.NewSet()
core/commands/files/files.go
+1
-1
@@ -170,7 +170,7 @@ func statGetFormatOptions(req cmds.Request) (string, error) {
170
}
171
}
172
173
-func statNode(ds dag.DAGService, fsn mfs.FSNode) (*Object, error) {
173
+func statNode(ds node.DAGService, fsn mfs.FSNode) (*Object, error) {
174
nd, err := fsn.GetNode()
175
if err != nil {
176
return nil, err
core/commands/ls.go
+1
-1
@@ -134,7 +134,7 @@ The JSON output contains type information.
134
t := unixfspb.Data_DataType(-1)
135
136
linkNode, err := link.GetNode(req.Context(), dserv)
137
- if err == merkledag.ErrNotFound && !resolve {
137
+ if err == node.ErrNotFound && !resolve {
138
// not an error
139
linkNode = nil
140
} else if err != nil {
core/commands/object/object.go
+6
-5
@@ -2,6 +2,7 @@ package objectcmd
2
3
import (
4
"bytes"
5
+ "context"
6
"encoding/base64"
7
"encoding/json"
8
"encoding/xml"
@@ -422,7 +423,7 @@ And then run:
423
defer n.Blockstore.PinLock().Unlock()
424
}
425
425
- objectCid, err := objectPut(n, input, inputenc, datafieldenc)
426
+ objectCid, err := objectPut(req.Context(), n, input, inputenc, datafieldenc)
427
if err != nil {
428
errType := cmdkit.ErrNormal
429
if err == ErrUnknownObjectEnc {
@@ -504,12 +505,12 @@ Available templates:
505
}
506
}
507
507
- k, err := n.DAG.Add(node)
508
+ err = n.DAG.Add(req.Context(), node)
509
if err != nil {
510
res.SetError(err, cmdkit.ErrNormal)
511
return
512
}
512
- res.SetOutput(&Object{Hash: k.String()})
513
+ res.SetOutput(&Object{Hash: node.Cid().String()})
514
},
515
Marshalers: cmds.MarshalerMap{
516
cmds.Text: func(res cmds.Response) (io.Reader, error) {
@@ -542,7 +543,7 @@ func nodeFromTemplate(template string) (*dag.ProtoNode, error) {
543
var ErrEmptyNode = errors.New("no data or links in this node")
544
545
// objectPut takes a format option, serializes bytes from stdin and updates the dag with that data
545
-func objectPut(n *core.IpfsNode, input io.Reader, encoding string, dataFieldEncoding string) (*cid.Cid, error) {
546
+func objectPut(ctx context.Context, n *core.IpfsNode, input io.Reader, encoding string, dataFieldEncoding string) (*cid.Cid, error) {
547
548
data, err := ioutil.ReadAll(io.LimitReader(input, inputLimit+10))
549
if err != nil {
@@ -602,7 +603,7 @@ func objectPut(n *core.IpfsNode, input io.Reader, encoding string, dataFieldEnco
603
return nil, err
604
}
605
605
- _, err = n.DAG.Add(dagnode)
606
+ err = n.DAG.Add(ctx, dagnode)
607
if err != nil {
608
return nil, err
609
}
core/commands/object/patch.go
+6
-6
@@ -109,13 +109,13 @@ the limit will not be respected by the network.
109
110
rtpb.SetData(append(rtpb.Data(), data...))
111
112
- newkey, err := nd.DAG.Add(rtpb)
112
+ err = nd.DAG.Add(req.Context(), rtpb)
113
if err != nil {
114
res.SetError(err, cmdkit.ErrNormal)
115
return
116
}
117
118
- res.SetOutput(&Object{Hash: newkey.String()})
118
+ res.SetOutput(&Object{Hash: rtpb.Cid().String()})
119
},
120
Type: Object{},
121
Marshalers: cmds.MarshalerMap{
@@ -177,13 +177,13 @@ Example:
177
178
rtpb.SetData(data)
179
180
- newkey, err := nd.DAG.Add(rtpb)
180
+ err = nd.DAG.Add(req.Context(), rtpb)
181
if err != nil {
182
res.SetError(err, cmdkit.ErrNormal)
183
return
184
}
185
186
- res.SetOutput(&Object{Hash: newkey.String()})
186
+ res.SetOutput(&Object{Hash: rtpb.Cid().String()})
187
},
188
Type: Object{},
189
Marshalers: cmds.MarshalerMap{
@@ -237,7 +237,7 @@ Removes a link by the given name from root.
237
return
238
}
239
240
- nnode, err := e.Finalize(nd.DAG)
240
+ nnode, err := e.Finalize(req.Context(), nd.DAG)
241
if err != nil {
242
res.SetError(err, cmdkit.ErrNormal)
243
return
@@ -334,7 +334,7 @@ to a file containing 'bar', and returns the hash of the new object.
334
return
335
}
336
337
- nnode, err := e.Finalize(nd.DAG)
337
+ nnode, err := e.Finalize(req.Context(), nd.DAG)
338
if err != nil {
339
res.SetError(err, cmdkit.ErrNormal)
340
return
core/commands/pin.go
+7
-2
@@ -7,10 +7,12 @@ import (
7
"io"
8
"time"
9
10
+ bserv "github.com/ipfs/go-ipfs/blockservice"
11
cmds "github.com/ipfs/go-ipfs/commands"
12
core "github.com/ipfs/go-ipfs/core"
13
e "github.com/ipfs/go-ipfs/core/commands/e"
14
corerepo "github.com/ipfs/go-ipfs/core/corerepo"
15
+ offline "github.com/ipfs/go-ipfs/exchange/offline"
16
dag "github.com/ipfs/go-ipfs/merkledag"
17
path "github.com/ipfs/go-ipfs/path"
18
pin "github.com/ipfs/go-ipfs/pin"
@@ -555,7 +557,7 @@ func pinLsAll(typeStr string, ctx context.Context, n *core.IpfsNode) (map[string
557
if typeStr == "indirect" || typeStr == "all" {
558
set := cid.NewSet()
559
for _, k := range n.Pinning.RecursiveKeys() {
558
- err := dag.EnumerateChildren(n.Context(), n.DAG.GetLinks, k, set.Visit)
560
+ err := dag.EnumerateChildren(n.Context(), dag.GetLinksWithDAG(n.DAG), k, set.Visit)
561
if err != nil {
562
return nil, err
563
}
@@ -594,7 +596,10 @@ type pinVerifyOpts struct {
596
597
func pinVerify(ctx context.Context, n *core.IpfsNode, opts pinVerifyOpts) <-chan interface{} {
598
visited := make(map[string]PinStatus)
597
- getLinks := n.DAG.GetOfflineLinkService().GetLinks
599
+
600
+ bs := n.Blocks.Blockstore()
601
+ DAG := dag.NewDAGService(bserv.New(bs, offline.Exchange(bs)))
602
+ getLinks := dag.GetLinksWithDAG(DAG)
603
recPins := n.Pinning.RecursiveKeys()
604
605
var checkPin func(root *cid.Cid) PinStatus
core/commands/pubsub.go
+2
-2
@@ -100,13 +100,13 @@ This command outputs data in the following encodings:
100
if discover {
101
go func() {
102
blk := blocks.NewBlock([]byte("floodsub:" + topic))
103
- cid, err := n.Blocks.AddBlock(blk)
103
+ err := n.Blocks.AddBlock(blk)
104
if err != nil {
105
log.Error("pubsub discovery: ", err)
106
return
107
}
108
109
- connectToPubSubPeers(req.Context, n, cid)
109
+ connectToPubSubPeers(req.Context, n, blk.Cid())
110
}()
111
}
112
core/commands/refs.go
+2
-3
@@ -10,7 +10,6 @@ import (
10
cmds "github.com/ipfs/go-ipfs/commands"
11
"github.com/ipfs/go-ipfs/core"
12
e "github.com/ipfs/go-ipfs/core/commands/e"
13
- dag "github.com/ipfs/go-ipfs/merkledag"
13
path "github.com/ipfs/go-ipfs/path"
14
15
cid "gx/ipfs/QmcZfnkapfECQGcLZaf9B79NRg7cRa9EnZh4LSbkCzwNvY/go-cid"
@@ -222,7 +221,7 @@ type RefWrapper struct {
221
222
type RefWriter struct {
223
out chan interface{}
225
- DAG dag.DAGService
224
+ DAG node.DAGService
225
Ctx context.Context
226
227
Unique bool
@@ -244,7 +243,7 @@ func (rw *RefWriter) writeRefsRecursive(n node.Node) (int, error) {
243
nc := n.Cid()
244
245
var count int
247
- for i, ng := range dag.GetDAG(rw.Ctx, rw.DAG, n) {
246
+ for i, ng := range node.GetDAG(rw.Ctx, rw.DAG, n) {
247
lc := n.Links()[i].Cid
248
if rw.skip(lc) {
249
continue
core/commands/tar.go
+1
-1
@@ -51,7 +51,7 @@ represent it.
51
return
52
}
53
54
- node, err := tar.ImportTar(fi, nd.DAG)
54
+ node, err := tar.ImportTar(req.Context(), fi, nd.DAG)
55
if err != nil {
56
res.SetError(err, cmdkit.ErrNormal)
57
return
core/core.go
+3
-2
@@ -70,6 +70,7 @@ import (
70
ic "gx/ipfs/QmaPbCnUMBohSGo3KnxEa2bHqyJVVeEEcwtqJAYxerieBo/go-libp2p-crypto"
71
metrics "gx/ipfs/Qmb1QrSXKwGFWgiGEcyac4s5wakJG4yPvCPk49xZHxr5ux/go-libp2p-metrics"
72
cid "gx/ipfs/QmcZfnkapfECQGcLZaf9B79NRg7cRa9EnZh4LSbkCzwNvY/go-cid"
73
+ node "gx/ipfs/Qme5bWv7wtjUNGsK2BNGVUFPKiuxWrsqrtvYwCLRw8YFES/go-ipld-format"
74
pstore "gx/ipfs/QmeZVQzUrXqaszo24DAoHfGzcmCptN9JyngLkGAiEfk2x7/go-libp2p-peerstore"
75
mafilter "gx/ipfs/Qmf2UAmRwDG4TvnkQpHZWPAzw7rpCYVhxmRXmYxXr5LD1g/go-maddr-filter"
76
dht "gx/ipfs/QmfChjky1VNaHUQR9F2xqR1QEyX45pqU78nhsoq5GDYoKL/go-libp2p-kad-dht"
@@ -117,7 +118,7 @@ type IpfsNode struct {
118
BaseBlocks bstore.Blockstore // the raw blockstore, no filestore wrapping
119
GCLocker bstore.GCLocker // the locker used to protect the blockstore during gc
120
Blocks bserv.BlockService // the block service, get/add blocks.
120
- DAG merkledag.DAGService // the merkle dag service, get/add objects.
121
+ DAG node.DAGService // the merkle dag service, get/add objects.
122
Resolver *path.Resolver // the path resolution system
123
Reporter metrics.Reporter
124
Discovery discovery.Service
@@ -739,7 +740,7 @@ func (n *IpfsNode) loadFilesRoot() error {
740
switch {
741
case err == ds.ErrNotFound || val == nil:
742
nd = ft.EmptyDirNode()
742
- _, err := n.DAG.Add(nd)
743
+ err := n.DAG.Add(n.Context(), nd)
744
if err != nil {
745
return fmt.Errorf("failure writing to dagstore: %s", err)
746
}
core/coreapi/dag.go
+1
-1
@@ -41,7 +41,7 @@ func (api *DagAPI) Put(ctx context.Context, src io.Reader, opts ...caopts.DagPut
41
return nil, fmt.Errorf("no node returned from ParseInputs")
42
}
43
44
- _, err = api.node.DAG.Add(nds[0])
44
+ err = api.node.DAG.Add(ctx, nds[0])
45
if err != nil {
46
return nil, err
47
}
core/coreapi/unixfs_test.go
+9
-8
@@ -200,12 +200,12 @@ func TestCatDir(t *testing.T) {
200
if err != nil {
201
t.Error(err)
202
}
203
-
204
- c, err := node.DAG.Add(unixfs.EmptyDirNode())
203
+ edir := unixfs.EmptyDirNode()
204
+ err = node.DAG.Add(ctx, edir)
205
if err != nil {
206
t.Error(err)
207
}
208
- p := coreapi.ParseCid(c)
208
+ p := coreapi.ParseCid(edir.Cid())
209
210
if p.String() != emptyDir.String() {
211
t.Fatalf("expected path %s, got: %s", emptyDir, p)
@@ -224,12 +224,13 @@ func TestCatNonUnixfs(t *testing.T) {
224
t.Error(err)
225
}
226
227
- c, err := node.DAG.Add(new(mdag.ProtoNode))
227
+ nd := new(mdag.ProtoNode)
228
+ err = node.DAG.Add(ctx, nd)
229
if err != nil {
230
t.Error(err)
231
}
232
232
- _, err = api.Unixfs().Cat(ctx, coreapi.ParseCid(c))
233
+ _, err = api.Unixfs().Cat(ctx, coreapi.ParseCid(nd.Cid()))
234
if !strings.Contains(err.Error(), "proto: required field") {
235
t.Fatalf("expected protobuf error, got: %s", err)
236
}
@@ -292,7 +293,7 @@ func TestLsEmptyDir(t *testing.T) {
293
t.Error(err)
294
}
295
295
- _, err = node.DAG.Add(unixfs.EmptyDirNode())
296
+ err = node.DAG.Add(ctx, unixfs.EmptyDirNode())
297
if err != nil {
298
t.Error(err)
299
}
@@ -320,12 +321,12 @@ func TestLsNonUnixfs(t *testing.T) {
321
t.Fatal(err)
322
}
323
323
- c, err := node.DAG.Add(nd)
324
+ err = node.DAG.Add(ctx, nd)
325
if err != nil {
326
t.Error(err)
327
}
328
328
- links, err := api.Unixfs().Ls(ctx, coreapi.ParseCid(c))
329
+ links, err := api.Unixfs().Ls(ctx, coreapi.ParseCid(nd.Cid()))
330
if err != nil {
331
t.Error(err)
332
}
core/corehttp/gateway_handler.go
+6
-6
@@ -474,7 +474,7 @@ func (i *gatewayHandler) putHandler(w http.ResponseWriter, r *http.Request) {
474
return
475
}
476
477
- nnode, err := e.Finalize(i.node.DAG)
477
+ nnode, err := e.Finalize(ctx, i.node.DAG)
478
if err != nil {
479
webError(w, "putHandler: could not get node", err, http.StatusInternalServerError)
480
return
@@ -498,11 +498,11 @@ func (i *gatewayHandler) putHandler(w http.ResponseWriter, r *http.Request) {
498
// object set-data case
499
pbnd.SetData(pbnewnode.Data())
500
501
- newcid, err = i.node.DAG.Add(pbnd)
501
+ newcid = pbnd.Cid()
502
+ err = i.node.DAG.Add(ctx, pbnd)
503
if err != nil {
504
nnk := newnode.Cid()
504
- rk := pbnd.Cid()
505
- webError(w, fmt.Sprintf("putHandler: Could not add newnode(%q) to root(%q)", nnk.String(), rk.String()), err, http.StatusInternalServerError)
505
+ webError(w, fmt.Sprintf("putHandler: Could not add newnode(%q) to root(%q)", nnk.String(), newcid.String()), err, http.StatusInternalServerError)
506
return
507
}
508
default:
@@ -561,7 +561,7 @@ func (i *gatewayHandler) deleteHandler(w http.ResponseWriter, r *http.Request) {
561
562
var newnode *dag.ProtoNode = pbnd
563
for j := len(pathNodes) - 2; j >= 0; j-- {
564
- if _, err := i.node.DAG.Add(newnode); err != nil {
564
+ if err := i.node.DAG.Add(ctx, newnode); err != nil {
565
webError(w, "Could not add node", err, http.StatusInternalServerError)
566
return
567
}
@@ -579,7 +579,7 @@ func (i *gatewayHandler) deleteHandler(w http.ResponseWriter, r *http.Request) {
579
}
580
}
581
582
- if _, err := i.node.DAG.Add(newnode); err != nil {
582
+ if err := i.node.DAG.Add(ctx, newnode); err != nil {
583
webError(w, "Could not add root node", err, http.StatusInternalServerError)
584
return
585
}
core/corehttp/gateway_test.go
+11
-5
@@ -178,6 +178,9 @@ func TestGatewayGet(t *testing.T) {
178
}
179
180
func TestIPNSHostnameRedirect(t *testing.T) {
181
+ ctx, cancel := context.WithCancel(context.Background())
182
+ defer cancel()
183
+
184
ns := mockNamesys{}
185
ts, n := newTestServerAndNode(t, ns)
186
t.Logf("test server url: %s", ts.URL)
@@ -199,12 +202,12 @@ func TestIPNSHostnameRedirect(t *testing.T) {
202
t.Fatal(err)
203
}
204
202
- _, err = n.DAG.Add(dagn2)
205
+ err = n.DAG.Add(ctx, dagn2)
206
if err != nil {
207
t.Fatal(err)
208
}
209
207
- _, err = n.DAG.Add(dagn1)
210
+ err = n.DAG.Add(ctx, dagn1)
211
if err != nil {
212
t.Fatal(err)
213
}
@@ -262,6 +265,9 @@ func TestIPNSHostnameRedirect(t *testing.T) {
265
}
266
267
func TestIPNSHostnameBacklinks(t *testing.T) {
268
+ ctx, cancel := context.WithCancel(context.Background())
269
+ defer cancel()
270
+
271
ns := mockNamesys{}
272
ts, n := newTestServerAndNode(t, ns)
273
t.Logf("test server url: %s", ts.URL)
@@ -286,15 +292,15 @@ func TestIPNSHostnameBacklinks(t *testing.T) {
292
t.Fatal(err)
293
}
294
289
- _, err = n.DAG.Add(dagn3)
295
+ err = n.DAG.Add(ctx, dagn3)
296
if err != nil {
297
t.Fatal(err)
298
}
293
- _, err = n.DAG.Add(dagn2)
299
+ err = n.DAG.Add(ctx, dagn2)
300
if err != nil {
301
t.Fatal(err)
302
}
297
- _, err = n.DAG.Add(dagn1)
303
+ err = n.DAG.Add(ctx, dagn1)
304
if err != nil {
305
t.Fatal(err)
306
}
core/corerepo/gc.go
+2
-2
@@ -86,7 +86,7 @@ func GarbageCollect(n *core.IpfsNode, ctx context.Context) error {
86
if err != nil {
87
return err
88
}
89
- rmed := gc.GC(ctx, n.Blockstore, n.DAG, n.Pinning, roots)
89
+ rmed := gc.GC(ctx, n.Blockstore, n.Pinning, roots)
90
91
return CollectResult(ctx, rmed, nil)
92
}
@@ -154,7 +154,7 @@ func GarbageCollectAsync(n *core.IpfsNode, ctx context.Context) <-chan gc.Result
154
return out
155
}
156
157
- return gc.GC(ctx, n.Blockstore, n.DAG, n.Pinning, roots)
157
+ return gc.GC(ctx, n.Blockstore, n.Pinning, roots)
158
}
159
160
func PeriodicGC(ctx context.Context, node *core.IpfsNode) error {
core/coreunix/add.go
+7
-5
@@ -74,7 +74,7 @@ type AddedObject struct {
74
}
75
76
// NewAdder Returns a new Adder used for a file add operation.
77
-func NewAdder(ctx context.Context, p pin.Pinner, bs bstore.GCBlockstore, ds dag.DAGService) (*Adder, error) {
77
+func NewAdder(ctx context.Context, p pin.Pinner, bs bstore.GCBlockstore, ds node.DAGService) (*Adder, error) {
78
return &Adder{
79
ctx: ctx,
80
pinning: p,
@@ -94,7 +94,7 @@ type Adder struct {
94
ctx context.Context
95
pinning pin.Pinner
96
blockstore bstore.GCBlockstore
97
- dagService dag.DAGService
97
+ dagService node.DAGService
98
Out chan interface{}
99
Progress bool
100
Hidden bool
@@ -195,7 +195,9 @@ func (adder *Adder) PinRoot() error {
195
return nil
196
}
197
198
- rnk, err := adder.dagService.Add(root)
198
+ rnk := root.Cid()
199
+
200
+ err = adder.dagService.Add(adder.ctx, root)
201
if err != nil {
202
return err
203
}
@@ -470,7 +472,7 @@ func (adder *Adder) addFile(file files.File) error {
472
473
dagnode := dag.NodeWithData(sdata)
474
dagnode.SetPrefix(adder.Prefix)
473
- _, err = adder.dagService.Add(dagnode)
475
+ err = adder.dagService.Add(adder.ctx, dagnode)
476
if err != nil {
477
return err
478
}
@@ -573,7 +575,7 @@ func outputDagnode(out chan interface{}, name string, dn node.Node) error {
575
}
576
577
// NewMemoryDagService builds and returns a new mem-datastore.
576
-func NewMemoryDagService() dag.DAGService {
578
+func NewMemoryDagService() node.DAGService {
579
// build mem-datastore for editor's intermediary nodes
580
bs := bstore.NewBlockstore(syncds.MutexWrap(ds.NewMapDatastore()))
581
bsrv := bserv.New(bs, offline.Exchange(bs))
core/coreunix/add_test.go
+2
-2
@@ -104,7 +104,7 @@ func TestAddGCLive(t *testing.T) {
104
gcstarted := make(chan struct{})
105
go func() {
106
defer close(gcstarted)
107
- gcout = gc.GC(context.Background(), node.Blockstore, node.DAG, node.Pinning, nil)
107
+ gcout = gc.GC(context.Background(), node.Blockstore, node.Pinning, nil)
108
}()
109
110
// gc shouldnt start until we let the add finish its current file.
@@ -150,7 +150,7 @@ func TestAddGCLive(t *testing.T) {
150
defer cancel()
151
152
set := cid.NewSet()
153
- err = dag.EnumerateChildren(ctx, node.DAG.GetLinks, last, set.Visit)
153
+ err = dag.EnumerateChildren(ctx, dag.GetLinksWithDAG(node.DAG), last, set.Visit)
154
if err != nil {
155
t.Fatal(err)
156
}
core/coreunix/metadata.go
+2
-2
@@ -29,12 +29,12 @@ func AddMetadataTo(n *core.IpfsNode, skey string, m *ft.Metadata) (string, error
29
return "", err
30
}
31
32
- nk, err := n.DAG.Add(mdnode)
32
+ err = n.DAG.Add(n.Context(), mdnode)
33
if err != nil {
34
return "", err
35
}
36
37
- return nk.String(), nil
37
+ return mdnode.Cid().String(), nil
38
}
39
40
func Metadata(n *core.IpfsNode, skey string) (*ft.Metadata, error) {
core/coreunix/metadata_test.go
+2
-1
@@ -20,9 +20,10 @@ import (
20
ds "gx/ipfs/QmPpegoMqhAEqjncrzArm7KVWAkCm78rqL2DPuNjhPrshg/go-datastore"
21
dssync "gx/ipfs/QmPpegoMqhAEqjncrzArm7KVWAkCm78rqL2DPuNjhPrshg/go-datastore/sync"
22
cid "gx/ipfs/QmcZfnkapfECQGcLZaf9B79NRg7cRa9EnZh4LSbkCzwNvY/go-cid"
23
+ node "gx/ipfs/Qme5bWv7wtjUNGsK2BNGVUFPKiuxWrsqrtvYwCLRw8YFES/go-ipld-format"
24
)
25
25
-func getDagserv(t *testing.T) merkledag.DAGService {
26
+func getDagserv(t *testing.T) node.DAGService {
27
db := dssync.MutexWrap(ds.NewMapDatastore())
28
bs := bstore.NewBlockstore(db)
29
blockserv := bserv.New(bs, offline.Exchange(bs))
exchange/reprovide/providers.go
+4
-3
@@ -6,6 +6,7 @@ import (
6
blocks "github.com/ipfs/go-ipfs/blocks/blockstore"
7
merkledag "github.com/ipfs/go-ipfs/merkledag"
8
pin "github.com/ipfs/go-ipfs/pin"
9
+ node "gx/ipfs/Qme5bWv7wtjUNGsK2BNGVUFPKiuxWrsqrtvYwCLRw8YFES/go-ipld-format"
10
11
cid "gx/ipfs/QmcZfnkapfECQGcLZaf9B79NRg7cRa9EnZh4LSbkCzwNvY/go-cid"
12
)
@@ -18,7 +19,7 @@ func NewBlockstoreProvider(bstore blocks.Blockstore) KeyChanFunc {
19
}
20
21
// NewPinnedProvider returns provider supplying pinned keys
21
-func NewPinnedProvider(pinning pin.Pinner, dag merkledag.DAGService, onlyRoots bool) KeyChanFunc {
22
+func NewPinnedProvider(pinning pin.Pinner, dag node.DAGService, onlyRoots bool) KeyChanFunc {
23
return func(ctx context.Context) (<-chan *cid.Cid, error) {
24
set, err := pinSet(ctx, pinning, dag, onlyRoots)
25
if err != nil {
@@ -42,7 +43,7 @@ func NewPinnedProvider(pinning pin.Pinner, dag merkledag.DAGService, onlyRoots b
43
}
44
}
45
45
-func pinSet(ctx context.Context, pinning pin.Pinner, dag merkledag.DAGService, onlyRoots bool) (*streamingSet, error) {
46
+func pinSet(ctx context.Context, pinning pin.Pinner, dag node.DAGService, onlyRoots bool) (*streamingSet, error) {
47
set := newStreamingSet()
48
49
go func() {
@@ -56,7 +57,7 @@ func pinSet(ctx context.Context, pinning pin.Pinner, dag merkledag.DAGService, o
57
set.add(key)
58
59
if !onlyRoots {
59
- err := merkledag.EnumerateChildren(ctx, dag.GetLinks, key, set.add)
60
+ err := merkledag.EnumerateChildren(ctx, merkledag.GetLinksWithDAG(dag), key, set.add)
61
if err != nil {
62
log.Errorf("reprovide indirect pins: %s", err)
63
return
fuse/readonly/ipfs_test.go
+3
-3
@@ -145,7 +145,7 @@ func TestIpfsStressRead(t *testing.T) {
145
t.Fatal(err)
146
}
147
148
- _, err = nd.DAG.Add(newdir)
148
+ err = nd.DAG.Add(nd.Context(), newdir)
149
if err != nil {
150
t.Fatal(err)
151
}
@@ -224,12 +224,12 @@ func TestIpfsBasicDirRead(t *testing.T) {
224
t.Fatal(err)
225
}
226
227
- d1ndk, err := nd.DAG.Add(d1nd)
227
+ err = nd.DAG.Add(nd.Context(), d1nd)
228
if err != nil {
229
t.Fatal(err)
230
}
231
232
- dirname := path.Join(mnt.Dir, d1ndk.String())
232
+ dirname := path.Join(mnt.Dir, d1nd.Cid().String())
233
fname := path.Join(dirname, "actual")
234
rbuf, err := ioutil.ReadFile(fname)
235
if err != nil {
fuse/readonly/readonly_unix.go
+4
-4
@@ -21,7 +21,7 @@ import (
21
lgbl "gx/ipfs/QmaDoQyTYCS3DrPLBLXMixXfuCstBVVR81J3UY1vMxghpT/go-libp2p-loggables"
22
fuse "gx/ipfs/QmaFNtBAXX4nVMQWbUqNysXyhevUj1k4B1y5uS45LC7Vw9/fuse"
23
fs "gx/ipfs/QmaFNtBAXX4nVMQWbUqNysXyhevUj1k4B1y5uS45LC7Vw9/fuse/fs"
24
- format "gx/ipfs/Qme5bWv7wtjUNGsK2BNGVUFPKiuxWrsqrtvYwCLRw8YFES/go-ipld-format"
24
+ node "gx/ipfs/Qme5bWv7wtjUNGsK2BNGVUFPKiuxWrsqrtvYwCLRw8YFES/go-ipld-format"
25
)
26
27
var log = logging.Logger("fuse/ipfs")
@@ -92,7 +92,7 @@ func (*Root) ReadDirAll(ctx context.Context) ([]fuse.Dirent, error) {
92
// Node is the core object representing a filesystem tree node.
93
type Node struct {
94
Ipfs *core.IpfsNode
95
- Nd format.Node
95
+ Nd node.Node
96
cached *ftpb.Data
97
}
98
@@ -157,7 +157,7 @@ func (s *Node) Lookup(ctx context.Context, name string) (fs.Node, error) {
157
158
nd, err := s.Ipfs.DAG.Get(ctx, link.Cid)
159
switch err {
160
- case mdag.ErrNotFound:
160
+ case node.ErrNotFound:
161
default:
162
log.Errorf("fuse lookup %q: %s", name, err)
163
return nil, err
@@ -177,7 +177,7 @@ func (s *Node) ReadDirAll(ctx context.Context) ([]fuse.Dirent, error) {
177
}
178
179
var entries []fuse.Dirent
180
- err = dir.ForEachLink(ctx, func(lnk *format.Link) error {
180
+ err = dir.ForEachLink(ctx, func(lnk *node.Link) error {
181
n := lnk.Name
182
if len(n) == 0 {
183
n = lnk.Cid.String()
importer/balanced/balanced_test.go
+3
-2
@@ -16,11 +16,12 @@ import (
16
uio "github.com/ipfs/go-ipfs/unixfs/io"
17
18
u "gx/ipfs/QmNiJuT8Ja3hMVpBHXv3Q6dwmperaQ6JjLtpMQgMCD7xvx/go-ipfs-util"
19
+ node "gx/ipfs/Qme5bWv7wtjUNGsK2BNGVUFPKiuxWrsqrtvYwCLRw8YFES/go-ipld-format"
20
)
21
22
// TODO: extract these tests and more as a generic layout test suite
23
23
-func buildTestDag(ds dag.DAGService, spl chunk.Splitter) (*dag.ProtoNode, error) {
24
+func buildTestDag(ds node.DAGService, spl chunk.Splitter) (*dag.ProtoNode, error) {
25
dbp := h.DagBuilderParams{
26
Dagserv: ds,
27
Maxlinks: h.DefaultLinksPerBlock,
@@ -34,7 +35,7 @@ func buildTestDag(ds dag.DAGService, spl chunk.Splitter) (*dag.ProtoNode, error)
35
return nd.(*dag.ProtoNode), nil
36
}
37
37
-func getTestDag(t *testing.T, ds dag.DAGService, size int64, blksize int64) (*dag.ProtoNode, []byte) {
38
+func getTestDag(t *testing.T, ds node.DAGService, size int64, blksize int64) (*dag.ProtoNode, []byte) {
39
data := make([]byte, size)
40
u.NewTimeSeededRand().Read(data)
41
r := bytes.NewReader(data)
importer/helpers/dagbuilder.go
+7
-6
@@ -1,6 +1,7 @@
1
package helpers
2
3
import (
4
+ "context"
5
"io"
6
"os"
7
@@ -16,13 +17,13 @@ import (
17
// DagBuilderHelper wraps together a bunch of objects needed to
18
// efficiently create unixfs dag trees
19
type DagBuilderHelper struct {
19
- dserv dag.DAGService
20
+ dserv node.DAGService
21
spl chunk.Splitter
22
recvdErr error
23
rawLeaves bool
24
nextData []byte // the next item to return.
25
maxlinks int
25
- batch *dag.Batch
26
+ batch *node.Batch
27
fullPath string
28
stat os.FileInfo
29
prefix *cid.Prefix
@@ -40,7 +41,7 @@ type DagBuilderParams struct {
41
Prefix *cid.Prefix
42
43
// DAGService to write blocks to (required)
43
- Dagserv dag.DAGService
44
+ Dagserv node.DAGService
45
46
// NoCopy signals to the chunker that it should track fileinfo for
47
// filestore adds
@@ -56,7 +57,7 @@ func (dbp *DagBuilderParams) New(spl chunk.Splitter) *DagBuilderHelper {
57
rawLeaves: dbp.RawLeaves,
58
prefix: dbp.Prefix,
59
maxlinks: dbp.Maxlinks,
59
- batch: dbp.Dagserv.Batch(),
60
+ batch: node.NewBatch(context.TODO(), dbp.Dagserv),
61
}
62
if fi, ok := spl.Reader().(files.FileInfo); dbp.NoCopy && ok {
63
db.fullPath = fi.AbsPath()
@@ -106,7 +107,7 @@ func (db *DagBuilderHelper) Next() ([]byte, error) {
107
}
108
109
// GetDagServ returns the dagservice object this Helper is using
109
-func (db *DagBuilderHelper) GetDagServ() dag.DAGService {
110
+func (db *DagBuilderHelper) GetDagServ() node.DAGService {
111
return db.dserv
112
}
113
@@ -199,7 +200,7 @@ func (db *DagBuilderHelper) Add(node *UnixfsNode) (node.Node, error) {
200
return nil, err
201
}
202
202
- _, err = db.dserv.Add(dn)
203
+ err = db.dserv.Add(context.TODO(), dn)
204
if err != nil {
205
return nil, err
206
}
importer/helpers/helpers.go
+7
-3
@@ -78,7 +78,7 @@ func (n *UnixfsNode) Set(other *UnixfsNode) {
78
}
79
}
80
81
-func (n *UnixfsNode) GetChild(ctx context.Context, i int, ds dag.DAGService) (*UnixfsNode, error) {
81
+func (n *UnixfsNode) GetChild(ctx context.Context, i int, ds node.DAGService) (*UnixfsNode, error) {
82
nd, err := n.node.Links()[i].GetNode(ctx, ds)
83
if err != nil {
84
return nil, err
@@ -110,7 +110,7 @@ func (n *UnixfsNode) AddChild(child *UnixfsNode, db *DagBuilderHelper) error {
110
return err
111
}
112
113
- _, err = db.batch.Add(childnode)
113
+ err = db.batch.Add(childnode)
114
115
return err
116
}
@@ -133,7 +133,11 @@ func (n *UnixfsNode) FileSize() uint64 {
133
}
134
135
func (n *UnixfsNode) SetPosInfo(offset uint64, fullPath string, stat os.FileInfo) {
136
- n.posInfo = &pi.PosInfo{offset, fullPath, stat}
136
+ n.posInfo = &pi.PosInfo{
137
+ Offset: offset,
138
+ FullPath: fullPath,
139
+ Stat: stat,
140
+ }
141
}
142
143
// getDagNode fills out the proper formatting for the unixfs node
importer/importer.go
+3
-4
@@ -10,7 +10,6 @@ import (
10
"github.com/ipfs/go-ipfs/importer/chunk"
11
h "github.com/ipfs/go-ipfs/importer/helpers"
12
trickle "github.com/ipfs/go-ipfs/importer/trickle"
13
- dag "github.com/ipfs/go-ipfs/merkledag"
13
"gx/ipfs/QmceUdzxkimdYsgtX733uNgzf1DLHyBKN6ehGSp85ayppM/go-ipfs-cmdkit/files"
14
15
node "gx/ipfs/Qme5bWv7wtjUNGsK2BNGVUFPKiuxWrsqrtvYwCLRw8YFES/go-ipld-format"
@@ -18,7 +17,7 @@ import (
17
18
// Builds a DAG from the given file, writing created blocks to disk as they are
19
// created
21
-func BuildDagFromFile(fpath string, ds dag.DAGService) (node.Node, error) {
20
+func BuildDagFromFile(fpath string, ds node.DAGService) (node.Node, error) {
21
stat, err := os.Lstat(fpath)
22
if err != nil {
23
return nil, err
@@ -37,7 +36,7 @@ func BuildDagFromFile(fpath string, ds dag.DAGService) (node.Node, error) {
36
return BuildDagFromReader(ds, chunk.DefaultSplitter(f))
37
}
38
40
-func BuildDagFromReader(ds dag.DAGService, spl chunk.Splitter) (node.Node, error) {
39
+func BuildDagFromReader(ds node.DAGService, spl chunk.Splitter) (node.Node, error) {
40
dbp := h.DagBuilderParams{
41
Dagserv: ds,
42
Maxlinks: h.DefaultLinksPerBlock,
@@ -46,7 +45,7 @@ func BuildDagFromReader(ds dag.DAGService, spl chunk.Splitter) (node.Node, error
45
return bal.BalancedLayout(dbp.New(spl))
46
}
47
49
-func BuildTrickleDagFromReader(ds dag.DAGService, spl chunk.Splitter) (node.Node, error) {
48
+func BuildTrickleDagFromReader(ds node.DAGService, spl chunk.Splitter) (node.Node, error) {
49
dbp := h.DagBuilderParams{
50
Dagserv: ds,
51
Maxlinks: h.DefaultLinksPerBlock,
importer/importer_test.go
+3
-4
@@ -8,7 +8,6 @@ import (
8
"testing"
9
10
chunk "github.com/ipfs/go-ipfs/importer/chunk"
11
- dag "github.com/ipfs/go-ipfs/merkledag"
11
mdtest "github.com/ipfs/go-ipfs/merkledag/test"
12
uio "github.com/ipfs/go-ipfs/unixfs/io"
13
@@ -16,7 +15,7 @@ import (
15
node "gx/ipfs/Qme5bWv7wtjUNGsK2BNGVUFPKiuxWrsqrtvYwCLRw8YFES/go-ipld-format"
16
)
17
19
-func getBalancedDag(t testing.TB, size int64, blksize int64) (node.Node, dag.DAGService) {
18
+func getBalancedDag(t testing.TB, size int64, blksize int64) (node.Node, node.DAGService) {
19
ds := mdtest.Mock()
20
r := io.LimitReader(u.NewTimeSeededRand(), size)
21
nd, err := BuildDagFromReader(ds, chunk.NewSizeSplitter(r, blksize))
@@ -26,7 +25,7 @@ func getBalancedDag(t testing.TB, size int64, blksize int64) (node.Node, dag.DAG
25
return nd, ds
26
}
27
29
-func getTrickleDag(t testing.TB, size int64, blksize int64) (node.Node, dag.DAGService) {
28
+func getTrickleDag(t testing.TB, size int64, blksize int64) (node.Node, node.DAGService) {
29
ds := mdtest.Mock()
30
r := io.LimitReader(u.NewTimeSeededRand(), size)
31
nd, err := BuildTrickleDagFromReader(ds, chunk.NewSizeSplitter(r, blksize))
@@ -102,7 +101,7 @@ func BenchmarkTrickleReadFull(b *testing.B) {
101
runReadBench(b, nd, ds)
102
}
103
105
-func runReadBench(b *testing.B, nd node.Node, ds dag.DAGService) {
104
+func runReadBench(b *testing.B, nd node.Node, ds node.DAGService) {
105
for i := 0; i < b.N; i++ {
106
ctx, cancel := context.WithCancel(context.Background())
107
read, err := uio.NewDagReader(ctx, nd, ds)
importer/trickle/trickle_test.go
+2
-1
@@ -17,6 +17,7 @@ import (
17
uio "github.com/ipfs/go-ipfs/unixfs/io"
18
19
u "gx/ipfs/QmNiJuT8Ja3hMVpBHXv3Q6dwmperaQ6JjLtpMQgMCD7xvx/go-ipfs-util"
20
+ node "gx/ipfs/Qme5bWv7wtjUNGsK2BNGVUFPKiuxWrsqrtvYwCLRw8YFES/go-ipld-format"
21
)
22
23
type UseRawLeaves bool
@@ -31,7 +32,7 @@ func runBothSubtests(t *testing.T, tfunc func(*testing.T, UseRawLeaves)) {
32
t.Run("leaves=Raw", func(t *testing.T) { tfunc(t, RawLeaves) })
33
}
34
34
-func buildTestDag(ds merkledag.DAGService, spl chunk.Splitter, rawLeaves UseRawLeaves) (*merkledag.ProtoNode, error) {
35
+func buildTestDag(ds node.DAGService, spl chunk.Splitter, rawLeaves UseRawLeaves) (*merkledag.ProtoNode, error) {
36
dbp := h.DagBuilderParams{
37
Dagserv: ds,
38
Maxlinks: h.DefaultLinksPerBlock,
merkledag/batch.go
deleted
-99
@@ -1,99 +0,0 @@
1
-package merkledag
2
-
3
-import (
4
- "runtime"
5
-
6
- cid "gx/ipfs/QmcZfnkapfECQGcLZaf9B79NRg7cRa9EnZh4LSbkCzwNvY/go-cid"
7
- node "gx/ipfs/Qme5bWv7wtjUNGsK2BNGVUFPKiuxWrsqrtvYwCLRw8YFES/go-ipld-format"
8
- blocks "gx/ipfs/Qmej7nf81hi2x2tvjRBF3mcp74sQyuDH4VMYDGd1YtXjb2/go-block-format"
9
-)
10
-
11
-// ParallelBatchCommits is the number of batch commits that can be in-flight before blocking.
12
-// TODO(#4299): Experiment with multiple datastores, storage devices, and CPUs to find
13
-// the right value/formula.
14
-var ParallelBatchCommits = runtime.NumCPU() * 2
15
-
16
-// Batch is a buffer for batching adds to a dag.
17
-type Batch struct {
18
- ds *dagService
19
-
20
- activeCommits int
21
- commitError error
22
- commitResults chan error
23
-
24
- blocks []blocks.Block
25
- size int
26
-
27
- MaxSize int
28
- MaxBlocks int
29
-}
30
-
31
-func (t *Batch) processResults() {
32
- for t.activeCommits > 0 && t.commitError == nil {
33
- select {
34
- case err := <-t.commitResults:
35
- t.activeCommits--
36
- if err != nil {
37
- t.commitError = err
38
- }
39
- default:
40
- return
41
- }
42
- }
43
-}
44
-
45
-func (t *Batch) asyncCommit() {
46
- numBlocks := len(t.blocks)
47
- if numBlocks == 0 || t.commitError != nil {
48
- return
49
- }
50
- if t.activeCommits >= ParallelBatchCommits {
51
- err := <-t.commitResults
52
- t.activeCommits--
53
-
54
- if err != nil {
55
- t.commitError = err
56
- return
57
- }
58
- }
59
- go func(b []blocks.Block) {
60
- _, err := t.ds.Blocks.AddBlocks(b)
61
- t.commitResults <- err
62
- }(t.blocks)
63
-
64
- t.activeCommits++
65
- t.blocks = make([]blocks.Block, 0, numBlocks)
66
- t.size = 0
67
-
68
- return
69
-}
70
-
71
-// Add adds a node to the batch and commits the batch if necessary.
72
-func (t *Batch) Add(nd node.Node) (*cid.Cid, error) {
73
- // Not strictly necessary but allows us to catch errors early.
74
- t.processResults()
75
- if t.commitError != nil {
76
- return nil, t.commitError
77
- }
78
-
79
- t.blocks = append(t.blocks, nd)
80
- t.size += len(nd.RawData())
81
- if t.size > t.MaxSize || len(t.blocks) > t.MaxBlocks {
82
- t.asyncCommit()
83
- }
84
- return nd.Cid(), t.commitError
85
-}
86
-
87
-// Commit commits batched nodes.
88
-func (t *Batch) Commit() error {
89
- t.asyncCommit()
90
- for t.activeCommits > 0 && t.commitError == nil {
91
- err := <-t.commitResults
92
- t.activeCommits--
93
- if err != nil {
94
- t.commitError = err
95
- }
96
- }
97
-
98
- return t.commitError
99
-}
merkledag/merkledag.go
+69
-218
@@ -7,11 +7,11 @@ import (
7
"sync"
8
9
bserv "github.com/ipfs/go-ipfs/blockservice"
10
- offline "github.com/ipfs/go-ipfs/exchange/offline"
10
11
ipldcbor "gx/ipfs/QmNRz7BDWfdFNVLt7AVvmRefkrURD25EeoipcXqo6yoXU1/go-ipld-cbor"
12
cid "gx/ipfs/QmcZfnkapfECQGcLZaf9B79NRg7cRa9EnZh4LSbkCzwNvY/go-cid"
13
node "gx/ipfs/Qme5bWv7wtjUNGsK2BNGVUFPKiuxWrsqrtvYwCLRw8YFES/go-ipld-format"
14
+ blocks "gx/ipfs/Qmej7nf81hi2x2tvjRBF3mcp74sQyuDH4VMYDGd1YtXjb2/go-block-format"
15
)
16
17
// TODO: We should move these registrations elsewhere. Really, most of the IPLD
@@ -23,37 +23,7 @@ func init() {
23
node.Register(cid.DagCBOR, ipldcbor.DecodeBlock)
24
}
25
26
-var ErrNotFound = fmt.Errorf("merkledag: not found")
27
-
28
-// DAGService is an IPFS Merkle DAG service.
29
-type DAGService interface {
30
- // Add adds the node to the DAGService
31
- Add(node.Node) (*cid.Cid, error)
32
- // Get gets the node the from the DAGService
33
- Get(context.Context, *cid.Cid) (node.Node, error)
34
- // Remove removes the node from the DAGService
35
- Remove(node.Node) error
36
-
37
- // GetMany returns a channel of NodeOption given
38
- // a set of CIDs.
39
- GetMany(context.Context, []*cid.Cid) <-chan *NodeOption
40
-
41
- // Batch is a buffer for batching adds to a dag.
42
- Batch() *Batch
43
-
44
- LinkService
45
-}
46
-
47
-type LinkService interface {
48
- // GetLinks return all links for a node. The complete node does not
49
- // necessarily have to exist locally, or at all. For example, raw
50
- // leaves cannot possibly have links so there is no need to look
51
- // at the node.
52
- GetLinks(context.Context, *cid.Cid) ([]*node.Link, error)
53
-
54
- GetOfflineLinkService() LinkService
55
-}
56
-
26
+// NewDAGService constructs a new DAGService (using the default implementation).
27
func NewDAGService(bs bserv.BlockService) *dagService {
28
return &dagService{Blocks: bs}
29
}
@@ -68,25 +38,20 @@ type dagService struct {
38
}
39
40
// Add adds a node to the dagService, storing the block in the BlockService
71
-func (n *dagService) Add(nd node.Node) (*cid.Cid, error) {
41
+func (n *dagService) Add(ctx context.Context, nd node.Node) error {
42
if n == nil { // FIXME remove this assertion. protect with constructor invariant
73
- return nil, fmt.Errorf("dagService is nil")
43
+ return fmt.Errorf("dagService is nil")
44
}
45
46
return n.Blocks.AddBlock(nd)
47
}
48
79
-func (n *dagService) Batch() *Batch {
80
- return &Batch{
81
- ds: n,
82
- commitResults: make(chan error, ParallelBatchCommits),
83
- MaxSize: 8 << 20,
84
-
85
- // By default, only batch up to 128 nodes at a time.
86
- // The current implementation of flatfs opens this many file
87
- // descriptors at the same time for the optimized batch write.
88
- MaxBlocks: 128,
49
+func (n *dagService) AddMany(ctx context.Context, nds []node.Node) error {
50
+ blks := make([]blocks.Block, len(nds))
51
+ for i, nd := range nds {
52
+ blks[i] = nd
53
}
54
+ return n.Blocks.AddBlocks(blks)
55
}
56
57
// Get retrieves a node from the dagService, fetching the block in the BlockService
@@ -101,7 +66,7 @@ func (n *dagService) Get(ctx context.Context, c *cid.Cid) (node.Node, error) {
66
b, err := n.Blocks.GetBlock(ctx, c)
67
if err != nil {
68
if err == bserv.ErrNotFound {
104
- return nil, ErrNotFound
69
+ return nil, node.ErrNotFound
70
}
71
return nil, fmt.Errorf("Failed to get block for %s: %v", c, err)
72
}
@@ -122,17 +87,23 @@ func (n *dagService) GetLinks(ctx context.Context, c *cid.Cid) ([]*node.Link, er
87
return node.Links(), nil
88
}
89
125
-func (n *dagService) GetOfflineLinkService() LinkService {
126
- if n.Blocks.Exchange().IsOnline() {
127
- bsrv := bserv.New(n.Blocks.Blockstore(), offline.Exchange(n.Blocks.Blockstore()))
128
- return NewDAGService(bsrv)
129
- } else {
130
- return n
131
- }
90
+func (n *dagService) Remove(ctx context.Context, c *cid.Cid) error {
91
+ return n.Blocks.DeleteBlock(c)
92
}
93
134
-func (n *dagService) Remove(nd node.Node) error {
135
- return n.Blocks.DeleteBlock(nd)
94
+// RemoveMany removes multiple nodes from the DAG. It will likely be faster than
95
+// removing them individually.
96
+//
97
+// This operation is not atomic. If it returns an error, some nodes may or may
98
+// not have been removed.
99
+func (n *dagService) RemoveMany(ctx context.Context, cids []*cid.Cid) error {
100
+ // TODO(#4608): make this batch all the way down.
101
+ for _, c := range cids {
102
+ if err := n.Blocks.DeleteBlock(c); err != nil {
103
+ return err
104
+ }
105
+ }
106
+ return nil
107
}
108
109
// GetLinksDirect creates a function to get the links for a node, from
@@ -140,14 +111,14 @@ func (n *dagService) Remove(nd node.Node) error {
111
// locally (and can not be retrieved) an error will be returned.
112
func GetLinksDirect(serv node.NodeGetter) GetLinks {
113
return func(ctx context.Context, c *cid.Cid) ([]*node.Link, error) {
143
- node, err := serv.Get(ctx, c)
114
+ nd, err := serv.Get(ctx, c)
115
if err != nil {
116
if err == bserv.ErrNotFound {
146
- err = ErrNotFound
117
+ err = node.ErrNotFound
118
}
119
return nil, err
120
}
150
- return node.Links(), nil
121
+ return nd.Links(), nil
122
}
123
}
124
@@ -155,11 +126,12 @@ type sesGetter struct {
126
bs *bserv.Session
127
}
128
129
+// Get gets a single node from the DAG.
130
func (sg *sesGetter) Get(ctx context.Context, c *cid.Cid) (node.Node, error) {
131
blk, err := sg.bs.GetBlock(ctx, c)
132
switch err {
133
case bserv.ErrNotFound:
162
- return nil, ErrNotFound
134
+ return nil, node.ErrNotFound
135
default:
136
return nil, err
137
case nil:
@@ -169,8 +141,13 @@ func (sg *sesGetter) Get(ctx context.Context, c *cid.Cid) (node.Node, error) {
141
return node.Decode(blk)
142
}
143
144
+// GetMany gets many nodes at once, batching the request if possible.
145
+func (sg *sesGetter) GetMany(ctx context.Context, keys []*cid.Cid) <-chan *node.NodeOption {
146
+ return getNodesFromBG(ctx, sg.bs, keys)
147
+}
148
+
149
// FetchGraph fetches all nodes that are children of the given node
173
-func FetchGraph(ctx context.Context, root *cid.Cid, serv DAGService) error {
150
+func FetchGraph(ctx context.Context, root *cid.Cid, serv node.DAGService) error {
151
var ng node.NodeGetter = serv
152
ds, ok := serv.(*dagService)
153
if ok {
@@ -205,14 +182,18 @@ func FindLinks(links []*cid.Cid, c *cid.Cid, start int) []int {
182
return out
183
}
184
208
-type NodeOption struct {
209
- Node node.Node
210
- Err error
185
+// GetMany gets many nodes from the DAG at once.
186
+//
187
+// This method may not return all requested nodes (and may or may not return an
188
+// error indicating that it failed to do so. It is up to the caller to verify
189
+// that it received all nodes.
190
+func (n *dagService) GetMany(ctx context.Context, keys []*cid.Cid) <-chan *node.NodeOption {
191
+ return getNodesFromBG(ctx, n.Blocks, keys)
192
}
193
213
-func (ds *dagService) GetMany(ctx context.Context, keys []*cid.Cid) <-chan *NodeOption {
214
- out := make(chan *NodeOption, len(keys))
215
- blocks := ds.Blocks.GetBlocks(ctx, keys)
194
+func getNodesFromBG(ctx context.Context, bs bserv.BlockGetter, keys []*cid.Cid) <-chan *node.NodeOption {
195
+ out := make(chan *node.NodeOption, len(keys))
196
+ blocks := bs.GetBlocks(ctx, keys)
197
var count int
198
199
go func() {
@@ -222,182 +203,43 @@ func (ds *dagService) GetMany(ctx context.Context, keys []*cid.Cid) <-chan *Node
203
case b, ok := <-blocks:
204
if !ok {
205
if count != len(keys) {
225
- out <- &NodeOption{Err: fmt.Errorf("failed to fetch all nodes")}
206
+ out <- &node.NodeOption{Err: fmt.Errorf("failed to fetch all nodes")}
207
}
208
return
209
}
210
211
nd, err := node.Decode(b)
212
if err != nil {
232
- out <- &NodeOption{Err: err}
213
+ out <- &node.NodeOption{Err: err}
214
return
215
}
216
236
- out <- &NodeOption{Node: nd}
217
+ out <- &node.NodeOption{Node: nd}
218
count++
219
220
case <-ctx.Done():
240
- out <- &NodeOption{Err: ctx.Err()}
241
- return
242
- }
243
- }
244
- }()
245
- return out
246
-}
247
-
248
-// GetDAG will fill out all of the links of the given Node.
249
-// It returns a channel of nodes, which the caller can receive
250
-// all the child nodes of 'root' on, in proper order.
251
-func GetDAG(ctx context.Context, ds DAGService, root node.Node) []NodeGetter {
252
- var cids []*cid.Cid
253
- for _, lnk := range root.Links() {
254
- cids = append(cids, lnk.Cid)
255
- }
256
-
257
- return GetNodes(ctx, ds, cids)
258
-}
259
-
260
-// GetNodes returns an array of 'NodeGetter' promises, with each corresponding
261
-// to the key with the same index as the passed in keys
262
-func GetNodes(ctx context.Context, ds DAGService, keys []*cid.Cid) []NodeGetter {
263
-
264
- // Early out if no work to do
265
- if len(keys) == 0 {
266
- return nil
267
- }
268
-
269
- promises := make([]NodeGetter, len(keys))
270
- for i := range keys {
271
- promises[i] = newNodePromise(ctx)
272
- }
273
-
274
- dedupedKeys := dedupeKeys(keys)
275
- go func() {
276
- ctx, cancel := context.WithCancel(ctx)
277
- defer cancel()
278
-
279
- nodechan := ds.GetMany(ctx, dedupedKeys)
280
-
281
- for count := 0; count < len(keys); {
282
- select {
283
- case opt, ok := <-nodechan:
284
- if !ok {
285
- for _, p := range promises {
286
- p.Fail(ErrNotFound)
287
- }
288
- return
289
- }
290
-
291
- if opt.Err != nil {
292
- for _, p := range promises {
293
- p.Fail(opt.Err)
294
- }
295
- return
296
- }
297
-
298
- nd := opt.Node
299
- is := FindLinks(keys, nd.Cid(), 0)
300
- for _, i := range is {
301
- count++
302
- promises[i].Send(nd)
303
- }
304
- case <-ctx.Done():
221
+ out <- &node.NodeOption{Err: ctx.Err()}
222
return
223
}
224
}
225
}()
309
- return promises
310
-}
311
-
312
-// Remove duplicates from a list of keys
313
-func dedupeKeys(cids []*cid.Cid) []*cid.Cid {
314
- out := make([]*cid.Cid, 0, len(cids))
315
- set := cid.NewSet()
316
- for _, c := range cids {
317
- if set.Visit(c) {
318
- out = append(out, c)
319
- }
320
- }
226
return out
227
}
228
324
-func newNodePromise(ctx context.Context) NodeGetter {
325
- return &nodePromise{
326
- recv: make(chan node.Node, 1),
327
- ctx: ctx,
328
- err: make(chan error, 1),
329
- }
330
-}
331
-
332
-type nodePromise struct {
333
- cache node.Node
334
- clk sync.Mutex
335
- recv chan node.Node
336
- ctx context.Context
337
- err chan error
338
-}
339
-
340
-// NodeGetter provides a promise like interface for a dag Node
341
-// the first call to Get will block until the Node is received
342
-// from its internal channels, subsequent calls will return the
343
-// cached node.
344
-type NodeGetter interface {
345
- Get(context.Context) (node.Node, error)
346
- Fail(err error)
347
- Send(node.Node)
348
-}
349
-
350
-func (np *nodePromise) Fail(err error) {
351
- np.clk.Lock()
352
- v := np.cache
353
- np.clk.Unlock()
354
-
355
- // if promise has a value, don't fail it
356
- if v != nil {
357
- return
358
- }
359
-
360
- np.err <- err
361
-}
362
-
363
-func (np *nodePromise) Send(nd node.Node) {
364
- var already bool
365
- np.clk.Lock()
366
- if np.cache != nil {
367
- already = true
368
- }
369
- np.cache = nd
370
- np.clk.Unlock()
371
-
372
- if already {
373
- panic("sending twice to the same promise is an error!")
374
- }
375
-
376
- np.recv <- nd
377
-}
378
-
379
-func (np *nodePromise) Get(ctx context.Context) (node.Node, error) {
380
- np.clk.Lock()
381
- c := np.cache
382
- np.clk.Unlock()
383
- if c != nil {
384
- return c, nil
385
- }
229
+// GetLinks is the type of function passed to the EnumerateChildren function(s)
230
+// for getting the children of an IPLD node.
231
+type GetLinks func(context.Context, *cid.Cid) ([]*node.Link, error)
232
387
- select {
388
- case nd := <-np.recv:
389
- return nd, nil
390
- case <-np.ctx.Done():
391
- return nil, np.ctx.Err()
392
- case <-ctx.Done():
393
- return nil, ctx.Err()
394
- case err := <-np.err:
395
- return nil, err
233
+// GetLinksWithDAG returns a GetLinks function that tries to use the given
234
+// NodeGetter as a LinkGetter to get the children of a given IPLD node. This may
235
+// allow us to traverse the DAG without actually loading and parsing the node in
236
+// question (if we already have the links cached).
237
+func GetLinksWithDAG(ng node.NodeGetter) GetLinks {
238
+ return func(ctx context.Context, c *cid.Cid) ([]*node.Link, error) {
239
+ return node.GetLinks(ctx, ng, c)
240
}
241
}
242
399
-type GetLinks func(context.Context, *cid.Cid) ([]*node.Link, error)
400
-
243
// EnumerateChildren will walk the dag below the given root node and add all
244
// unseen children to the passed in set.
245
// TODO: parallelize to avoid disk latency perf hits?
@@ -443,6 +285,10 @@ func (p *ProgressTracker) Value() int {
285
// 'fetchNodes' will start at a time
286
var FetchGraphConcurrency = 8
287
288
+// EnumerateChildrenAsync is equivalent to EnumerateChildren *except* that it
289
+// fetches children in parallel.
290
+//
291
+// NOTE: It *does not* make multiple concurrent calls to the passed `visit` function.
292
func EnumerateChildrenAsync(ctx context.Context, getLinks GetLinks, c *cid.Cid, visit func(*cid.Cid) bool) error {
293
feed := make(chan *cid.Cid)
294
out := make(chan []*node.Link)
@@ -523,3 +369,8 @@ func EnumerateChildrenAsync(ctx context.Context, getLinks GetLinks, c *cid.Cid,
369
}
370
371
}
372
+
373
+var _ node.LinkGetter = &dagService{}
374
+var _ node.NodeGetter = &dagService{}
375
+var _ node.NodeGetter = &sesGetter{}
376
+var _ node.DAGService = &dagService{}
merkledag/merkledag_test.go
+34
-22
@@ -131,7 +131,7 @@ func TestBatchFetchDupBlock(t *testing.T) {
131
132
func runBatchFetchTest(t *testing.T, read io.Reader) {
133
ctx := context.Background()
134
- var dagservs []DAGService
134
+ var dagservs []node.DAGService
135
for _, bsi := range bstest.Mocks(5) {
136
dagservs = append(dagservs, NewDAGService(bsi))
137
}
@@ -155,7 +155,7 @@ func runBatchFetchTest(t *testing.T, read io.Reader) {
155
t.Fatal(err)
156
}
157
158
- _, err = dagservs[0].Add(root)
158
+ err = dagservs[0].Add(ctx, root)
159
if err != nil {
160
t.Fatal(err)
161
}
@@ -221,7 +221,7 @@ func TestCantGet(t *testing.T) {
221
}
222
223
func TestFetchGraph(t *testing.T) {
224
- var dservs []DAGService
224
+ var dservs []node.DAGService
225
bsis := bstest.Mocks(2)
226
for _, bsi := range bsis {
227
dservs = append(dservs, NewDAGService(bsi))
@@ -285,13 +285,15 @@ func TestEnumerateChildren(t *testing.T) {
285
}
286
287
func TestFetchFailure(t *testing.T) {
288
+ ctx := context.Background()
289
+
290
ds := dstest.Mock()
291
ds_bad := dstest.Mock()
292
293
top := new(ProtoNode)
294
for i := 0; i < 10; i++ {
295
nd := NodeWithData([]byte{byte('a' + i)})
294
- _, err := ds.Add(nd)
296
+ err := ds.Add(ctx, nd)
297
if err != nil {
298
t.Fatal(err)
299
}
@@ -304,7 +306,7 @@ func TestFetchFailure(t *testing.T) {
306
307
for i := 0; i < 10; i++ {
308
nd := NodeWithData([]byte{'f', 'a' + byte(i)})
307
- _, err := ds_bad.Add(nd)
309
+ err := ds_bad.Add(ctx, nd)
310
if err != nil {
311
t.Fatal(err)
312
}
@@ -315,9 +317,9 @@ func TestFetchFailure(t *testing.T) {
317
}
318
}
319
318
- getters := GetDAG(context.Background(), ds, top)
320
+ getters := node.GetDAG(ctx, ds, top)
321
for i, getter := range getters {
320
- _, err := getter.Get(context.Background())
322
+ _, err := getter.Get(ctx)
323
if err != nil && i < 10 {
324
t.Fatal(err)
325
}
@@ -352,15 +354,17 @@ func TestUnmarshalFailure(t *testing.T) {
354
}
355
356
func TestBasicAddGet(t *testing.T) {
357
+ ctx := context.Background()
358
+
359
ds := dstest.Mock()
360
nd := new(ProtoNode)
361
358
- c, err := ds.Add(nd)
362
+ err := ds.Add(ctx, nd)
363
if err != nil {
364
t.Fatal(err)
365
}
366
363
- out, err := ds.Get(context.Background(), c)
367
+ out, err := ds.Get(ctx, nd.Cid())
368
if err != nil {
369
t.Fatal(err)
370
}
@@ -371,20 +375,22 @@ func TestBasicAddGet(t *testing.T) {
375
}
376
377
func TestGetRawNodes(t *testing.T) {
378
+ ctx := context.Background()
379
+
380
rn := NewRawNode([]byte("test"))
381
382
ds := dstest.Mock()
383
378
- c, err := ds.Add(rn)
384
+ err := ds.Add(ctx, rn)
385
if err != nil {
386
t.Fatal(err)
387
}
388
383
- if !c.Equals(rn.Cid()) {
389
+ if !rn.Cid().Equals(rn.Cid()) {
390
t.Fatal("output cids didnt match")
391
}
392
387
- out, err := ds.Get(context.TODO(), c)
393
+ out, err := ds.Get(ctx, rn.Cid())
394
if err != nil {
395
t.Fatal(err)
396
}
@@ -449,6 +455,8 @@ func TestProtoNodeResolve(t *testing.T) {
455
}
456
457
func TestCidRetention(t *testing.T) {
458
+ ctx := context.Background()
459
+
460
nd := new(ProtoNode)
461
nd.SetData([]byte("fooooo"))
462
@@ -466,13 +474,13 @@ func TestCidRetention(t *testing.T) {
474
}
475
476
bs := dstest.Bserv()
469
- _, err = bs.AddBlock(blk)
477
+ err = bs.AddBlock(blk)
478
if err != nil {
479
t.Fatal(err)
480
}
481
482
ds := NewDAGService(bs)
475
- out, err := ds.Get(context.Background(), c2)
483
+ out, err := ds.Get(ctx, c2)
484
if err != nil {
485
t.Fatal(err)
486
}
@@ -501,6 +509,8 @@ func TestCidRawDoesnNeedData(t *testing.T) {
509
}
510
511
func TestEnumerateAsyncFailsNotFound(t *testing.T) {
512
+ ctx := context.Background()
513
+
514
a := NodeWithData([]byte("foo1"))
515
b := NodeWithData([]byte("foo2"))
516
c := NodeWithData([]byte("foo3"))
@@ -508,7 +518,7 @@ func TestEnumerateAsyncFailsNotFound(t *testing.T) {
518
519
ds := dstest.Mock()
520
for _, n := range []node.Node{a, b, c} {
511
- _, err := ds.Add(n)
521
+ err := ds.Add(ctx, n)
522
if err != nil {
523
t.Fatal(err)
524
}
@@ -531,13 +541,13 @@ func TestEnumerateAsyncFailsNotFound(t *testing.T) {
541
t.Fatal(err)
542
}
543
534
- pcid, err := ds.Add(parent)
544
+ err := ds.Add(ctx, parent)
545
if err != nil {
546
t.Fatal(err)
547
}
548
549
cset := cid.NewSet()
540
- err = EnumerateChildrenAsync(context.Background(), GetLinksDirect(ds), pcid, cset.Visit)
550
+ err = EnumerateChildrenAsync(ctx, GetLinksDirect(ds), parent.Cid(), cset.Visit)
551
if err == nil {
552
t.Fatal("this should have failed")
553
}
@@ -570,7 +580,9 @@ func testProgressIndicator(t *testing.T, depth int) {
580
}
581
}
582
573
-func mkDag(ds DAGService, depth int) (*cid.Cid, int) {
583
+func mkDag(ds node.DAGService, depth int) (*cid.Cid, int) {
584
+ ctx := context.Background()
585
+
586
totalChildren := 0
587
f := func() *ProtoNode {
588
p := new(ProtoNode)
@@ -578,7 +590,7 @@ func mkDag(ds DAGService, depth int) (*cid.Cid, int) {
590
rand.Read(buf)
591
592
p.SetData(buf)
581
- _, err := ds.Add(p)
593
+ err := ds.Add(ctx, p)
594
if err != nil {
595
panic(err)
596
}
@@ -589,7 +601,7 @@ func mkDag(ds DAGService, depth int) (*cid.Cid, int) {
601
thisf := f
602
f = func() *ProtoNode {
603
pn := mkNodeWithChildren(thisf, 10)
592
- _, err := ds.Add(pn)
604
+ err := ds.Add(ctx, pn)
605
if err != nil {
606
panic(err)
607
}
@@ -599,12 +611,12 @@ func mkDag(ds DAGService, depth int) (*cid.Cid, int) {
611
}
612
613
nd := f()
602
- c, err := ds.Add(nd)
614
+ err := ds.Add(ctx, nd)
615
if err != nil {
616
panic(err)
617
}
618
607
- return c, totalChildren
619
+ return nd.Cid(), totalChildren
620
}
621
622
func mkNodeWithChildren(getChild func() *ProtoNode, width int) *ProtoNode {
merkledag/node.go
+3
-3
@@ -140,7 +140,7 @@ func (n *ProtoNode) RemoveNodeLink(name string) error {
140
n.links = good
141
142
if !found {
143
- return ErrNotFound
143
+ return node.ErrNotFound
144
}
145
146
return nil
@@ -160,7 +160,7 @@ func (n *ProtoNode) GetNodeLink(name string) (*node.Link, error) {
160
return nil, ErrLinkNotFound
161
}
162
163
-func (n *ProtoNode) GetLinkedProtoNode(ctx context.Context, ds DAGService, name string) (*ProtoNode, error) {
163
+func (n *ProtoNode) GetLinkedProtoNode(ctx context.Context, ds node.DAGService, name string) (*ProtoNode, error) {
164
nd, err := n.GetLinkedNode(ctx, ds, name)
165
if err != nil {
166
return nil, err
@@ -174,7 +174,7 @@ func (n *ProtoNode) GetLinkedProtoNode(ctx context.Context, ds DAGService, name
174
return pbnd, nil
175
}
176
177
-func (n *ProtoNode) GetLinkedNode(ctx context.Context, ds DAGService, name string) (node.Node, error) {
177
+func (n *ProtoNode) GetLinkedNode(ctx context.Context, ds node.DAGService, name string) (node.Node, error) {
178
lnk, err := n.GetNodeLink(name)
179
if err != nil {
180
return nil, err
merkledag/node_test.go
+12
-7
@@ -41,7 +41,7 @@ func TestRemoveLink(t *testing.T) {
41
42
// should fail
43
err = nd.RemoveNodeLink("a")
44
- if err != ErrNotFound {
44
+ if err != node.ErrNotFound {
45
t.Fatal("should have failed to remove link")
46
}
47
@@ -60,20 +60,25 @@ func TestRemoveLink(t *testing.T) {
60
}
61
62
func TestFindLink(t *testing.T) {
63
+ ctx := context.Background()
64
+
65
ds := mdtest.Mock()
64
- k, err := ds.Add(new(ProtoNode))
66
+ ndEmpty := new(ProtoNode)
67
+ err := ds.Add(ctx, ndEmpty)
68
if err != nil {
69
t.Fatal(err)
70
}
71
72
+ kEmpty := ndEmpty.Cid()
73
+
74
nd := &ProtoNode{}
75
nd.SetLinks([]*node.Link{
71
- {Name: "a", Cid: k},
72
- {Name: "c", Cid: k},
73
- {Name: "b", Cid: k},
76
+ {Name: "a", Cid: kEmpty},
77
+ {Name: "c", Cid: kEmpty},
78
+ {Name: "b", Cid: kEmpty},
79
})
80
76
- _, err = ds.Add(nd)
81
+ err = ds.Add(ctx, nd)
82
if err != nil {
83
t.Fatal(err)
84
}
@@ -107,7 +112,7 @@ func TestFindLink(t *testing.T) {
112
t.Fatal(err)
113
}
114
110
- if olnk.Cid.String() == k.String() {
115
+ if olnk.Cid.String() == kEmpty.String() {
116
t.Fatal("new link should have different hash")
117
}
118
}
merkledag/test/utils.go
+3
-1
@@ -5,11 +5,13 @@ import (
5
bsrv "github.com/ipfs/go-ipfs/blockservice"
6
"github.com/ipfs/go-ipfs/exchange/offline"
7
dag "github.com/ipfs/go-ipfs/merkledag"
8
+
9
ds "gx/ipfs/QmPpegoMqhAEqjncrzArm7KVWAkCm78rqL2DPuNjhPrshg/go-datastore"
10
dssync "gx/ipfs/QmPpegoMqhAEqjncrzArm7KVWAkCm78rqL2DPuNjhPrshg/go-datastore/sync"
11
+ node "gx/ipfs/Qme5bWv7wtjUNGsK2BNGVUFPKiuxWrsqrtvYwCLRw8YFES/go-ipld-format"
12
)
13
12
-func Mock() dag.DAGService {
14
+func Mock() node.DAGService {
15
return dag.NewDAGService(Bserv())
16
}
17
merkledag/traverse/traverse_test.go
+8
-7
@@ -2,6 +2,7 @@ package traverse
2
3
import (
4
"bytes"
5
+ "context"
6
"fmt"
7
"testing"
8
@@ -350,7 +351,7 @@ func testWalkOutputs(t *testing.T, root node.Node, opts Options, expect []byte)
351
}
352
}
353
353
-func newFan(t *testing.T, ds mdag.DAGService) node.Node {
354
+func newFan(t *testing.T, ds node.DAGService) node.Node {
355
a := mdag.NodeWithData([]byte("/a"))
356
addLink(t, ds, a, child(t, ds, a, "aa"))
357
addLink(t, ds, a, child(t, ds, a, "ab"))
@@ -359,7 +360,7 @@ func newFan(t *testing.T, ds mdag.DAGService) node.Node {
360
return a
361
}
362
362
-func newLinkedList(t *testing.T, ds mdag.DAGService) node.Node {
363
+func newLinkedList(t *testing.T, ds node.DAGService) node.Node {
364
a := mdag.NodeWithData([]byte("/a"))
365
aa := child(t, ds, a, "aa")
366
aaa := child(t, ds, aa, "aaa")
@@ -372,7 +373,7 @@ func newLinkedList(t *testing.T, ds mdag.DAGService) node.Node {
373
return a
374
}
375
375
-func newBinaryTree(t *testing.T, ds mdag.DAGService) node.Node {
376
+func newBinaryTree(t *testing.T, ds node.DAGService) node.Node {
377
a := mdag.NodeWithData([]byte("/a"))
378
aa := child(t, ds, a, "aa")
379
ab := child(t, ds, a, "ab")
@@ -385,7 +386,7 @@ func newBinaryTree(t *testing.T, ds mdag.DAGService) node.Node {
386
return a
387
}
388
388
-func newBinaryDAG(t *testing.T, ds mdag.DAGService) node.Node {
389
+func newBinaryDAG(t *testing.T, ds node.DAGService) node.Node {
390
a := mdag.NodeWithData([]byte("/a"))
391
aa := child(t, ds, a, "aa")
392
aaa := child(t, ds, aa, "aaa")
@@ -402,9 +403,9 @@ func newBinaryDAG(t *testing.T, ds mdag.DAGService) node.Node {
403
return a
404
}
405
405
-func addLink(t *testing.T, ds mdag.DAGService, a, b node.Node) {
406
+func addLink(t *testing.T, ds node.DAGService, a, b node.Node) {
407
to := string(a.(*mdag.ProtoNode).Data()) + "2" + string(b.(*mdag.ProtoNode).Data())
407
- if _, err := ds.Add(b); err != nil {
408
+ if err := ds.Add(context.Background(), b); err != nil {
409
t.Error(err)
410
}
411
if err := a.(*mdag.ProtoNode).AddNodeLink(to, b.(*mdag.ProtoNode)); err != nil {
@@ -412,6 +413,6 @@ func addLink(t *testing.T, ds mdag.DAGService, a, b node.Node) {
413
}
414
}
415
415
-func child(t *testing.T, ds mdag.DAGService, a node.Node, name string) node.Node {
416
+func child(t *testing.T, ds node.DAGService, a node.Node, name string) node.Node {
417
return mdag.NodeWithData([]byte(string(a.(*mdag.ProtoNode).Data()) + "/" + name))
418
}
merkledag/utils/diff.go
+3
-3
@@ -37,7 +37,7 @@ func (c *Change) String() string {
37
}
38
}
39
40
-func ApplyChange(ctx context.Context, ds dag.DAGService, nd *dag.ProtoNode, cs []*Change) (*dag.ProtoNode, error) {
40
+func ApplyChange(ctx context.Context, ds node.DAGService, nd *dag.ProtoNode, cs []*Change) (*dag.ProtoNode, error) {
41
e := NewDagEditor(nd, ds)
42
for _, c := range cs {
43
switch c.Type {
@@ -85,11 +85,11 @@ func ApplyChange(ctx context.Context, ds dag.DAGService, nd *dag.ProtoNode, cs [
85
}
86
}
87
88
- return e.Finalize(ds)
88
+ return e.Finalize(ctx, ds)
89
}
90
91
// Diff returns a set of changes that transform node 'a' into node 'b'
92
-func Diff(ctx context.Context, ds dag.DAGService, a, b node.Node) ([]*Change, error) {
92
+func Diff(ctx context.Context, ds node.DAGService, a, b node.Node) ([]*Change, error) {
93
if len(a.Links()) == 0 && len(b.Links()) == 0 {
94
return []*Change{
95
&Change{
merkledag/utils/diffenum_test.go
+20
-4
@@ -136,7 +136,7 @@ func TestDiffEnumBasic(t *testing.T) {
136
lgds := &getLogger{ds: ds}
137
138
for _, nd := range nds {
139
- _, err := ds.Add(nd)
139
+ err := ds.Add(ctx, nd)
140
if err != nil {
141
t.Fatal(err)
142
}
@@ -167,6 +167,22 @@ func (gl *getLogger) Get(ctx context.Context, c *cid.Cid) (node.Node, error) {
167
return nd, nil
168
}
169
170
+func (gl *getLogger) GetMany(ctx context.Context, cids []*cid.Cid) <-chan *node.NodeOption {
171
+ outCh := make(chan *node.NodeOption, len(cids))
172
+ nds := gl.ds.GetMany(ctx, cids)
173
+ for no := range nds {
174
+ if no.Err == nil {
175
+ gl.log = append(gl.log, no.Node.Cid())
176
+ }
177
+ select {
178
+ case outCh <- no:
179
+ default:
180
+ panic("too many responses")
181
+ }
182
+ }
183
+ return nds
184
+}
185
+
186
func assertCidList(a, b []*cid.Cid) error {
187
if len(a) != len(b) {
188
return fmt.Errorf("got different number of cids than expected")
@@ -188,14 +204,14 @@ func TestDiffEnumFail(t *testing.T) {
204
lgds := &getLogger{ds: ds}
205
206
for _, s := range []string{"a1", "a2", "b", "c"} {
191
- _, err := ds.Add(nds[s])
207
+ err := ds.Add(ctx, nds[s])
208
if err != nil {
209
t.Fatal(err)
210
}
211
}
212
213
err := DiffEnumerate(ctx, lgds, nds["a1"].Cid(), nds["a2"].Cid())
198
- if err != dag.ErrNotFound {
214
+ if err != node.ErrNotFound {
215
t.Fatal("expected err not found")
216
}
217
@@ -215,7 +231,7 @@ func TestDiffEnumRecurse(t *testing.T) {
231
lgds := &getLogger{ds: ds}
232
233
for _, s := range []string{"a1", "a2", "b", "c", "d"} {
218
- _, err := ds.Add(nds[s])
234
+ err := ds.Add(ctx, nds[s])
235
if err != nil {
236
t.Fatal(err)
237
}
merkledag/utils/utils.go
+24
-23
@@ -20,14 +20,14 @@ type Editor struct {
20
21
// tmp is a temporary in memory (for now) dagstore for all of the
22
// intermediary nodes to be stored in
23
- tmp dag.DAGService
23
+ tmp node.DAGService
24
25
// src is the dagstore with *all* of the data on it, it is used to pull
26
// nodes from for modification (nil is a valid value)
27
- src dag.DAGService
27
+ src node.DAGService
28
}
29
30
-func NewMemoryDagService() dag.DAGService {
30
+func NewMemoryDagService() node.DAGService {
31
// build mem-datastore for editor's intermediary nodes
32
bs := bstore.NewBlockstore(syncds.MutexWrap(ds.NewMapDatastore()))
33
bsrv := bserv.New(bs, offline.Exchange(bs))
@@ -35,7 +35,7 @@ func NewMemoryDagService() dag.DAGService {
35
}
36
37
// root is the node to be modified, source is the dagstore to pull nodes from (optional)
38
-func NewDagEditor(root *dag.ProtoNode, source dag.DAGService) *Editor {
38
+func NewDagEditor(root *dag.ProtoNode, source node.DAGService) *Editor {
39
return &Editor{
40
root: root,
41
tmp: NewMemoryDagService(),
@@ -47,22 +47,22 @@ func (e *Editor) GetNode() *dag.ProtoNode {
47
return e.root.Copy().(*dag.ProtoNode)
48
}
49
50
-func (e *Editor) GetDagService() dag.DAGService {
50
+func (e *Editor) GetDagService() node.DAGService {
51
return e.tmp
52
}
53
54
-func addLink(ctx context.Context, ds dag.DAGService, root *dag.ProtoNode, childname string, childnd node.Node) (*dag.ProtoNode, error) {
54
+func addLink(ctx context.Context, ds node.DAGService, root *dag.ProtoNode, childname string, childnd node.Node) (*dag.ProtoNode, error) {
55
if childname == "" {
56
return nil, errors.New("cannot create link with no name!")
57
}
58
59
// ensure that the node we are adding is in the dagservice
60
- _, err := ds.Add(childnd)
60
+ err := ds.Add(ctx, childnd)
61
if err != nil {
62
return nil, err
63
}
64
65
- _ = ds.Remove(root)
65
+ _ = ds.Remove(ctx, root.Cid())
66
67
// ensure no link with that name already exists
68
_ = root.RemoveNodeLink(childname) // ignore error, only option is ErrNotFound
@@ -71,7 +71,7 @@ func addLink(ctx context.Context, ds dag.DAGService, root *dag.ProtoNode, childn
71
return nil, err
72
}
73
74
- if _, err := ds.Add(root); err != nil {
74
+ if err := ds.Add(ctx, root); err != nil {
75
return nil, err
76
}
77
return root, nil
@@ -98,7 +98,7 @@ func (e *Editor) insertNodeAtPath(ctx context.Context, root *dag.ProtoNode, path
98
if err == dag.ErrLinkNotFound && create != nil {
99
nd = create()
100
err = nil // no longer an error case
101
- } else if err == dag.ErrNotFound {
101
+ } else if err == node.ErrNotFound {
102
// try finding it in our source dagstore
103
nd, err = root.GetLinkedProtoNode(ctx, e.src, path[0])
104
}
@@ -115,7 +115,7 @@ func (e *Editor) insertNodeAtPath(ctx context.Context, root *dag.ProtoNode, path
115
return nil, err
116
}
117
118
- _ = e.tmp.Remove(root)
118
+ _ = e.tmp.Remove(ctx, root.Cid())
119
120
_ = root.RemoveNodeLink(path[0])
121
err = root.AddNodeLinkClean(path[0], ndprime)
@@ -123,7 +123,7 @@ func (e *Editor) insertNodeAtPath(ctx context.Context, root *dag.ProtoNode, path
123
return nil, err
124
}
125
126
- _, err = e.tmp.Add(root)
126
+ err = e.tmp.Add(ctx, root)
127
if err != nil {
128
return nil, err
129
}
@@ -149,7 +149,7 @@ func (e *Editor) rmLink(ctx context.Context, root *dag.ProtoNode, path []string)
149
return nil, err
150
}
151
152
- _, err = e.tmp.Add(root)
152
+ err = e.tmp.Add(ctx, root)
153
if err != nil {
154
return nil, err
155
}
@@ -159,7 +159,7 @@ func (e *Editor) rmLink(ctx context.Context, root *dag.ProtoNode, path []string)
159
160
// search for node in both tmp dagstore and source dagstore
161
nd, err := root.GetLinkedProtoNode(ctx, e.tmp, path[0])
162
- if err == dag.ErrNotFound {
162
+ if err == node.ErrNotFound {
163
nd, err = root.GetLinkedProtoNode(ctx, e.src, path[0])
164
}
165
@@ -172,7 +172,7 @@ func (e *Editor) rmLink(ctx context.Context, root *dag.ProtoNode, path []string)
172
return nil, err
173
}
174
175
- _ = e.tmp.Remove(root)
175
+ e.tmp.Remove(ctx, root.Cid())
176
177
_ = root.RemoveNodeLink(path[0])
178
err = root.AddNodeLinkClean(path[0], nnode)
@@ -180,7 +180,7 @@ func (e *Editor) rmLink(ctx context.Context, root *dag.ProtoNode, path []string)
180
return nil, err
181
}
182
183
- _, err = e.tmp.Add(root)
183
+ err = e.tmp.Add(ctx, root)
184
if err != nil {
185
return nil, err
186
}
@@ -188,22 +188,23 @@ func (e *Editor) rmLink(ctx context.Context, root *dag.ProtoNode, path []string)
188
return root, nil
189
}
190
191
-func (e *Editor) Finalize(ds dag.DAGService) (*dag.ProtoNode, error) {
191
+func (e *Editor) Finalize(ctx context.Context, ds node.DAGService) (*dag.ProtoNode, error) {
192
nd := e.GetNode()
193
- err := copyDag(nd, e.tmp, ds)
193
+ err := copyDag(ctx, nd, e.tmp, ds)
194
return nd, err
195
}
196
197
-func copyDag(nd node.Node, from, to dag.DAGService) error {
198
- _, err := to.Add(nd)
197
+func copyDag(ctx context.Context, nd node.Node, from, to node.DAGService) error {
198
+ // TODO(#4609): make this batch.
199
+ err := to.Add(ctx, nd)
200
if err != nil {
201
return err
202
}
203
204
for _, lnk := range nd.Links() {
204
- child, err := lnk.GetNode(context.Background(), from)
205
+ child, err := lnk.GetNode(ctx, from)
206
if err != nil {
206
- if err == dag.ErrNotFound {
207
+ if err == node.ErrNotFound {
208
// not found means we didnt modify it, and it should
209
// already be in the target datastore
210
continue
@@ -211,7 +212,7 @@ func copyDag(nd node.Node, from, to dag.DAGService) error {
212
return err
213
}
214
214
- err = copyDag(child, from, to)
215
+ err = copyDag(ctx, child, from, to)
216
if err != nil {
217
return err
218
}
merkledag/utils/utils_test.go
+11
-7
@@ -9,35 +9,39 @@ import (
9
path "github.com/ipfs/go-ipfs/path"
10
11
cid "gx/ipfs/QmcZfnkapfECQGcLZaf9B79NRg7cRa9EnZh4LSbkCzwNvY/go-cid"
12
+ node "gx/ipfs/Qme5bWv7wtjUNGsK2BNGVUFPKiuxWrsqrtvYwCLRw8YFES/go-ipld-format"
13
)
14
15
func TestAddLink(t *testing.T) {
16
+ ctx, context := context.WithCancel(context.Background())
17
+ defer context()
18
+
19
ds := mdtest.Mock()
20
fishnode := dag.NodeWithData([]byte("fishcakes!"))
21
18
- fk, err := ds.Add(fishnode)
22
+ err := ds.Add(ctx, fishnode)
23
if err != nil {
24
t.Fatal(err)
25
}
26
27
nd := new(dag.ProtoNode)
24
- nnode, err := addLink(context.Background(), ds, nd, "fish", fishnode)
28
+ nnode, err := addLink(ctx, ds, nd, "fish", fishnode)
29
if err != nil {
30
t.Fatal(err)
31
}
32
29
- fnprime, err := nnode.GetLinkedNode(context.Background(), ds, "fish")
33
+ fnprime, err := nnode.GetLinkedNode(ctx, ds, "fish")
34
if err != nil {
35
t.Fatal(err)
36
}
37
38
fnpkey := fnprime.Cid()
35
- if !fnpkey.Equals(fk) {
39
+ if !fnpkey.Equals(fishnode.Cid()) {
40
t.Fatal("wrong child node found!")
41
}
42
}
43
40
-func assertNodeAtPath(t *testing.T, ds dag.DAGService, root *dag.ProtoNode, pth string, exp *cid.Cid) {
44
+func assertNodeAtPath(t *testing.T, ds node.DAGService, root *dag.ProtoNode, pth string, exp *cid.Cid) {
45
parts := path.SplitList(pth)
46
cur := root
47
for _, e := range parts {
@@ -78,7 +82,7 @@ func TestInsertNode(t *testing.T) {
82
83
func testInsert(t *testing.T, e *Editor, path, data string, create bool, experr string) {
84
child := dag.NodeWithData([]byte(data))
81
- ck, err := e.tmp.Add(child)
85
+ err := e.tmp.Add(context.Background(), child)
86
if err != nil {
87
t.Fatal(err)
88
}
@@ -106,5 +110,5 @@ func testInsert(t *testing.T, e *Editor, path, data string, create bool, experr
110
t.Fatal(err, path, data, create, experr)
111
}
112
109
- assertNodeAtPath(t, e.tmp, e.root, path, ck)
113
+ assertNodeAtPath(t, e.tmp, e.root, path, child.Cid())
114
}
mfs/dir.go
+6
-6
@@ -24,7 +24,7 @@ var ErrInvalidChild = errors.New("invalid child node")
24
var ErrDirExists = errors.New("directory already has entry by that name")
25
26
type Directory struct {
27
- dserv dag.DAGService
27
+ dserv node.DAGService
28
parent childCloser
29
30
childDirs map[string]*Directory
@@ -40,7 +40,7 @@ type Directory struct {
40
name string
41
}
42
43
-func NewDirectory(ctx context.Context, name string, node node.Node, parent childCloser, dserv dag.DAGService) (*Directory, error) {
43
+func NewDirectory(ctx context.Context, name string, node node.Node, parent childCloser, dserv node.DAGService) (*Directory, error) {
44
db, err := uio.NewDirectoryFromNode(dserv, node)
45
if err != nil {
46
return nil, err
@@ -104,7 +104,7 @@ func (d *Directory) flushCurrentNode() (*dag.ProtoNode, error) {
104
return nil, err
105
}
106
107
- _, err = d.dserv.Add(nd)
107
+ err = d.dserv.Add(d.ctx, nd)
108
if err != nil {
109
return nil, err
110
}
@@ -306,7 +306,7 @@ func (d *Directory) Mkdir(name string) (*Directory, error) {
306
ndir := ft.EmptyDirNode()
307
ndir.SetPrefix(d.GetPrefix())
308
309
- _, err = d.dserv.Add(ndir)
309
+ err = d.dserv.Add(d.ctx, ndir)
310
if err != nil {
311
return nil, err
312
}
@@ -354,7 +354,7 @@ func (d *Directory) AddChild(name string, nd node.Node) error {
354
return ErrDirExists
355
}
356
357
- _, err = d.dserv.Add(nd)
357
+ err = d.dserv.Add(d.ctx, nd)
358
if err != nil {
359
return err
360
}
@@ -420,7 +420,7 @@ func (d *Directory) GetNode() (node.Node, error) {
420
return nil, err
421
}
422
423
- _, err = d.dserv.Add(nd)
423
+ err = d.dserv.Add(d.ctx, nd)
424
if err != nil {
425
return nil, err
426
}
mfs/fd.go
+1
-1
@@ -122,7 +122,7 @@ func (fi *fileDescriptor) flushUp(fullsync bool) error {
122
return err
123
}
124
125
- _, err = fi.inode.dserv.Add(nd)
125
+ err = fi.inode.dserv.Add(context.TODO(), nd)
126
if err != nil {
127
return err
128
}
mfs/file.go
+2
-2
@@ -20,7 +20,7 @@ type File struct {
20
21
desclock sync.RWMutex
22
23
- dserv dag.DAGService
23
+ dserv node.DAGService
24
node node.Node
25
nodelk sync.Mutex
26
@@ -29,7 +29,7 @@ type File struct {
29
30
// NewFile returns a NewFile object with the given parameters. If the
31
// Cid version is non-zero RawLeaves will be enabled.
32
-func NewFile(name string, node node.Node, parent childCloser, dserv dag.DAGService) (*File, error) {
32
+func NewFile(name string, node node.Node, parent childCloser, dserv node.DAGService) (*File, error) {
33
fi := &File{
34
dserv: dserv,
35
parent: parent,
mfs/mfs_test.go
+8
-8
@@ -35,19 +35,19 @@ func emptyDirNode() *dag.ProtoNode {
35
return dag.NodeWithData(ft.FolderPBData())
36
}
37
38
-func getDagserv(t *testing.T) dag.DAGService {
38
+func getDagserv(t *testing.T) node.DAGService {
39
db := dssync.MutexWrap(ds.NewMapDatastore())
40
bs := bstore.NewBlockstore(db)
41
blockserv := bserv.New(bs, offline.Exchange(bs))
42
return dag.NewDAGService(blockserv)
43
}
44
45
-func getRandFile(t *testing.T, ds dag.DAGService, size int64) node.Node {
45
+func getRandFile(t *testing.T, ds node.DAGService, size int64) node.Node {
46
r := io.LimitReader(u.NewTimeSeededRand(), size)
47
return fileNodeFromReader(t, ds, r)
48
}
49
50
-func fileNodeFromReader(t *testing.T, ds dag.DAGService, r io.Reader) node.Node {
50
+func fileNodeFromReader(t *testing.T, ds node.DAGService, r io.Reader) node.Node {
51
nd, err := importer.BuildDagFromReader(ds, chunk.DefaultSplitter(r))
52
if err != nil {
53
t.Fatal(err)
@@ -128,7 +128,7 @@ func compStrArrs(a, b []string) bool {
128
return true
129
}
130
131
-func assertFileAtPath(ds dag.DAGService, root *Directory, expn node.Node, pth string) error {
131
+func assertFileAtPath(ds node.DAGService, root *Directory, expn node.Node, pth string) error {
132
exp, ok := expn.(*dag.ProtoNode)
133
if !ok {
134
return dag.ErrNotProtobuf
@@ -182,7 +182,7 @@ func assertFileAtPath(ds dag.DAGService, root *Directory, expn node.Node, pth st
182
return nil
183
}
184
185
-func catNode(ds dag.DAGService, nd *dag.ProtoNode) ([]byte, error) {
185
+func catNode(ds node.DAGService, nd *dag.ProtoNode) ([]byte, error) {
186
r, err := uio.NewDagReader(context.TODO(), nd, ds)
187
if err != nil {
188
return nil, err
@@ -192,7 +192,7 @@ func catNode(ds dag.DAGService, nd *dag.ProtoNode) ([]byte, error) {
192
return ioutil.ReadAll(r)
193
}
194
195
-func setupRoot(ctx context.Context, t *testing.T) (dag.DAGService, *Root) {
195
+func setupRoot(ctx context.Context, t *testing.T) (node.DAGService, *Root) {
196
ds := getDagserv(t)
197
198
root := emptyDirNode()
@@ -284,7 +284,7 @@ func TestDirectoryLoadFromDag(t *testing.T) {
284
rootdir := rt.GetValue().(*Directory)
285
286
nd := getRandFile(t, ds, 1000)
287
- _, err := ds.Add(nd)
287
+ err := ds.Add(ctx, nd)
288
if err != nil {
289
t.Fatal(err)
290
}
@@ -292,7 +292,7 @@ func TestDirectoryLoadFromDag(t *testing.T) {
292
fihash := nd.Cid()
293
294
dir := emptyDirNode()
295
- _, err = ds.Add(dir)
295
+ err = ds.Add(ctx, dir)
296
if err != nil {
297
t.Fatal(err)
298
}
mfs/system.go
+4
-4
@@ -58,7 +58,7 @@ type Root struct {
58
59
repub *Republisher
60
61
- dserv dag.DAGService
61
+ dserv node.DAGService
62
63
Type string
64
}
@@ -67,7 +67,7 @@ type Root struct {
67
type PubFunc func(context.Context, *cid.Cid) error
68
69
// NewRoot creates a new Root and starts up a republisher routine for it.
70
-func NewRoot(parent context.Context, ds dag.DAGService, node *dag.ProtoNode, pf PubFunc) (*Root, error) {
70
+func NewRoot(parent context.Context, ds node.DAGService, node *dag.ProtoNode, pf PubFunc) (*Root, error) {
71
72
var repub *Republisher
73
if pf != nil {
@@ -160,13 +160,13 @@ func (kr *Root) FlushMemFree(ctx context.Context) error {
160
// closeChild implements the childCloser interface, and signals to the publisher that
161
// there are changes ready to be published.
162
func (kr *Root) closeChild(name string, nd node.Node, sync bool) error {
163
- c, err := kr.dserv.Add(nd)
163
+ err := kr.dserv.Add(context.TODO(), nd)
164
if err != nil {
165
return err
166
}
167
168
if kr.repub != nil {
169
- kr.repub.Update(c)
169
+ kr.repub.Update(nd.Cid())
170
}
171
return nil
172
}
path/resolver.go
+4
-4
@@ -35,12 +35,12 @@ func (e ErrNoLink) Error() string {
35
// TODO: now that this is more modular, try to unify this code with the
36
// the resolvers in namesys
37
type Resolver struct {
38
- DAG dag.DAGService
38
+ DAG node.DAGService
39
40
- ResolveOnce func(ctx context.Context, ds dag.DAGService, nd node.Node, names []string) (*node.Link, []string, error)
40
+ ResolveOnce func(ctx context.Context, ds node.DAGService, nd node.Node, names []string) (*node.Link, []string, error)
41
}
42
43
-func NewBasicResolver(ds dag.DAGService) *Resolver {
43
+func NewBasicResolver(ds node.DAGService) *Resolver {
44
return &Resolver{
45
DAG: ds,
46
ResolveOnce: ResolveSingle,
@@ -123,7 +123,7 @@ func (s *Resolver) ResolvePath(ctx context.Context, fpath Path) (node.Node, erro
123
124
// ResolveSingle simply resolves one hop of a path through a graph with no
125
// extra context (does not opaquely resolve through sharded nodes)
126
-func ResolveSingle(ctx context.Context, ds dag.DAGService, nd node.Node, names []string) (*node.Link, []string, error) {
126
+func ResolveSingle(ctx context.Context, ds node.DAGService, nd node.Node, names []string) (*node.Link, []string, error) {
127
return nd.ResolveLink(names)
128
}
129
path/resolver_test.go
+1
-1
@@ -39,7 +39,7 @@ func TestRecurivePathResolution(t *testing.T) {
39
}
40
41
for _, n := range []node.Node{a, b, c} {
42
- _, err = dagService.Add(n)
42
+ err = dagService.Add(ctx, n)
43
if err != nil {
44
t.Fatal(err)
45
}
pin/gc/gc.go
+10
-7
@@ -6,6 +6,8 @@ import (
6
"fmt"
7
8
bstore "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"
12
pin "github.com/ipfs/go-ipfs/pin"
13
@@ -33,7 +35,7 @@ type Result struct {
35
// The routine then iterates over every block in the blockstore and
36
// deletes any block that is not found in the marked set.
37
//
36
-func GC(ctx context.Context, bs bstore.GCBlockstore, ls dag.LinkService, pn pin.Pinner, bestEffortRoots []*cid.Cid) <-chan Result {
38
+func GC(ctx context.Context, bs bstore.GCBlockstore, pn pin.Pinner, bestEffortRoots []*cid.Cid) <-chan Result {
39
40
elock := log.EventBegin(ctx, "GC.lockWait")
41
unlocker := bs.GCLock()
@@ -41,7 +43,8 @@ func GC(ctx context.Context, bs bstore.GCBlockstore, ls dag.LinkService, pn pin.
43
elock = log.EventBegin(ctx, "GC.locked")
44
emark := log.EventBegin(ctx, "GC.mark")
45
44
- ls = ls.GetOfflineLinkService()
46
+ bsrv := bserv.New(bs, offline.Exchange(bs))
47
+ ds := dag.NewDAGService(bsrv)
48
49
output := make(chan Result, 128)
50
@@ -50,7 +53,7 @@ func GC(ctx context.Context, bs bstore.GCBlockstore, ls dag.LinkService, pn pin.
53
defer unlocker.Unlock()
54
defer elock.Done()
55
53
- gcs, err := ColoredSet(ctx, pn, ls, bestEffortRoots, output)
56
+ gcs, err := ColoredSet(ctx, pn, ds, bestEffortRoots, output)
57
if err != nil {
58
output <- Result{Error: err}
59
return
@@ -125,13 +128,13 @@ func Descendants(ctx context.Context, getLinks dag.GetLinks, set *cid.Set, roots
128
129
// ColoredSet computes the set of nodes in the graph that are pinned by the
130
// pins in the given pinner.
128
-func ColoredSet(ctx context.Context, pn pin.Pinner, ls dag.LinkService, bestEffortRoots []*cid.Cid, output chan<- Result) (*cid.Set, error) {
131
+func ColoredSet(ctx context.Context, pn pin.Pinner, ng node.NodeGetter, bestEffortRoots []*cid.Cid, output chan<- Result) (*cid.Set, error) {
132
// KeySet currently implemented in memory, in the future, may be bloom filter or
133
// disk backed to conserve memory.
134
errors := false
135
gcs := cid.NewSet()
136
getLinks := func(ctx context.Context, cid *cid.Cid) ([]*node.Link, error) {
134
- links, err := ls.GetLinks(ctx, cid)
137
+ links, err := node.GetLinks(ctx, ng, cid)
138
if err != nil {
139
errors = true
140
output <- Result{Error: &CannotFetchLinksError{cid, err}}
@@ -145,8 +148,8 @@ func ColoredSet(ctx context.Context, pn pin.Pinner, ls dag.LinkService, bestEffo
148
}
149
150
bestEffortGetLinks := func(ctx context.Context, cid *cid.Cid) ([]*node.Link, error) {
148
- links, err := ls.GetLinks(ctx, cid)
149
- if err != nil && err != dag.ErrNotFound {
151
+ links, err := node.GetLinks(ctx, ng, cid)
152
+ if err != nil && err != node.ErrNotFound {
153
errors = true
154
output <- Result{Error: &CannotFetchLinksError{cid, err}}
155
}
pin/pin.go
+15
-11
@@ -178,13 +178,13 @@ type pinner struct {
178
// Track the keys used for storing the pinning state, so gc does
179
// not delete them.
180
internalPin *cid.Set
181
- dserv mdag.DAGService
182
- internal mdag.DAGService // dagservice used to store internal objects
181
+ dserv node.DAGService
182
+ internal node.DAGService // dagservice used to store internal objects
183
dstore ds.Datastore
184
}
185
186
// NewPinner creates a new pinner using the given datastore as a backend
187
-func NewPinner(dstore ds.Datastore, serv, internal mdag.DAGService) Pinner {
187
+func NewPinner(dstore ds.Datastore, serv, internal node.DAGService) Pinner {
188
189
rcset := cid.NewSet()
190
dirset := cid.NewSet()
@@ -203,11 +203,13 @@ func NewPinner(dstore ds.Datastore, serv, internal mdag.DAGService) Pinner {
203
func (p *pinner) Pin(ctx context.Context, node node.Node, recurse bool) error {
204
p.lock.Lock()
205
defer p.lock.Unlock()
206
- c, err := p.dserv.Add(node)
206
+ err := p.dserv.Add(ctx, node)
207
if err != nil {
208
return err
209
}
210
211
+ c := node.Cid()
212
+
213
if recurse {
214
if p.recursePin.Has(c) {
215
return nil
@@ -356,7 +358,7 @@ func (p *pinner) CheckIfPinned(cids ...*cid.Cid) ([]Pinned, error) {
358
// Now walk all recursive pins to check for indirect pins
359
var checkChildren func(*cid.Cid, *cid.Cid) error
360
checkChildren = func(rk, parentKey *cid.Cid) error {
359
- links, err := p.dserv.GetLinks(context.Background(), parentKey)
361
+ links, err := node.GetLinks(context.TODO(), p.dserv, parentKey)
362
if err != nil {
363
return err
364
}
@@ -425,7 +427,7 @@ func cidSetWithValues(cids []*cid.Cid) *cid.Set {
427
}
428
429
// LoadPinner loads a pinner and its keysets from the given datastore
428
-func LoadPinner(d ds.Datastore, dserv, internal mdag.DAGService) (Pinner, error) {
430
+func LoadPinner(d ds.Datastore, dserv, internal node.DAGService) (Pinner, error) {
431
p := new(pinner)
432
433
rootKeyI, err := d.Get(pinDatastoreKey)
@@ -550,16 +552,18 @@ func (p *pinner) Flush() error {
552
}
553
554
// add the empty node, its referenced by the pin sets but never created
553
- _, err := p.internal.Add(new(mdag.ProtoNode))
555
+ err := p.internal.Add(ctx, new(mdag.ProtoNode))
556
if err != nil {
557
return err
558
}
559
558
- k, err := p.internal.Add(root)
560
+ err = p.internal.Add(ctx, root)
561
if err != nil {
562
return err
563
}
564
565
+ k := root.Cid()
566
+
567
internalset.Add(k)
568
if err := p.dstore.Put(pinDatastoreKey, k.Bytes()); err != nil {
569
return fmt.Errorf("cannot store pin state: %v", err)
@@ -593,8 +597,8 @@ func (p *pinner) PinWithMode(c *cid.Cid, mode PinMode) {
597
598
// hasChild recursively looks for a Cid among the children of a root Cid.
599
// The visit function can be used to shortcut already-visited branches.
596
-func hasChild(ds mdag.LinkService, root *cid.Cid, child *cid.Cid, visit func(*cid.Cid) bool) (bool, error) {
597
- links, err := ds.GetLinks(context.Background(), root)
600
+func hasChild(ng node.NodeGetter, root *cid.Cid, child *cid.Cid, visit func(*cid.Cid) bool) (bool, error) {
601
+ links, err := node.GetLinks(context.TODO(), ng, root)
602
if err != nil {
603
return false, err
604
}
@@ -604,7 +608,7 @@ func hasChild(ds mdag.LinkService, root *cid.Cid, child *cid.Cid, visit func(*ci
608
return true, nil
609
}
610
if visit(c) {
607
- has, err := hasChild(ds, c, child, visit)
611
+ has, err := hasChild(ng, c, child, visit)
612
if err != nil {
613
return false, err
614
}
pin/pin_test.go
+20
-16
@@ -59,7 +59,7 @@ func TestPinnerBasic(t *testing.T) {
59
p := NewPinner(dstore, dserv, dserv)
60
61
a, ak := randNode()
62
- _, err := dserv.Add(a)
62
+ err := dserv.Add(ctx, a)
63
if err != nil {
64
t.Fatal(err)
65
}
@@ -74,10 +74,11 @@ func TestPinnerBasic(t *testing.T) {
74
75
// create new node c, to be indirectly pinned through b
76
c, _ := randNode()
77
- ck, err := dserv.Add(c)
77
+ err = dserv.Add(ctx, c)
78
if err != nil {
79
t.Fatal(err)
80
}
81
+ ck := c.Cid()
82
83
// Create new node b, to be parent to a and c
84
b, _ := randNode()
@@ -91,10 +92,11 @@ func TestPinnerBasic(t *testing.T) {
92
t.Fatal(err)
93
}
94
94
- _, err = dserv.Add(b)
95
+ err = dserv.Add(ctx, b)
96
if err != nil {
97
t.Fatal(err)
98
}
99
+ bk := b.Cid()
100
101
// recursively pin B{A,C}
102
err = p.Pin(ctx, b, true)
@@ -104,7 +106,6 @@ func TestPinnerBasic(t *testing.T) {
106
107
assertPinned(t, p, ck, "child of recursively pinned node not found")
108
107
- bk := b.Cid()
109
assertPinned(t, p, bk, "Recursively pinned node not found..")
110
111
d, _ := randNode()
@@ -115,11 +116,11 @@ func TestPinnerBasic(t *testing.T) {
116
d.AddNodeLink("e", e)
117
118
// Must be in dagserv for unpin to work
118
- _, err = dserv.Add(e)
119
+ err = dserv.Add(ctx, e)
120
if err != nil {
121
t.Fatal(err)
122
}
122
- _, err = dserv.Add(d)
123
+ err = dserv.Add(ctx, d)
124
if err != nil {
125
t.Fatal(err)
126
}
@@ -194,13 +195,13 @@ func TestIsPinnedLookup(t *testing.T) {
195
}
196
}
197
197
- ak, err := dserv.Add(a)
198
+ err := dserv.Add(ctx, a)
199
if err != nil {
200
t.Fatal(err)
201
}
202
//t.Logf("a[%d] is %s", i, ak)
203
aNodes[i] = a
203
- aKeys[i] = ak
204
+ aKeys[i] = a.Cid()
205
}
206
207
// Pin A5 recursively
@@ -222,20 +223,22 @@ func TestIsPinnedLookup(t *testing.T) {
223
}
224
225
// Add C
225
- ck, err := dserv.Add(c)
226
+ err := dserv.Add(ctx, c)
227
if err != nil {
228
t.Fatal(err)
229
}
230
+ ck := c.Cid()
231
//t.Logf("C is %s", ck)
232
233
// Add C to B and Add B
234
if err := b.AddNodeLink("myotherchild", c); err != nil {
235
t.Fatal(err)
236
}
235
- bk, err := dserv.Add(b)
237
+ err = dserv.Add(ctx, b)
238
if err != nil {
239
t.Fatal(err)
240
}
241
+ bk := b.Cid()
242
//t.Logf("B is %s", bk)
243
244
// Pin C recursively
@@ -284,7 +287,7 @@ func TestDuplicateSemantics(t *testing.T) {
287
p := NewPinner(dstore, dserv, dserv)
288
289
a, _ := randNode()
287
- _, err := dserv.Add(a)
290
+ err := dserv.Add(ctx, a)
291
if err != nil {
292
t.Fatal(err)
293
}
@@ -349,12 +352,12 @@ func TestPinRecursiveFail(t *testing.T) {
352
t.Fatal("should have failed to pin here")
353
}
354
352
- _, err = dserv.Add(b)
355
+ err = dserv.Add(ctx, b)
356
if err != nil {
357
t.Fatal(err)
358
}
359
357
- _, err = dserv.Add(a)
360
+ err = dserv.Add(ctx, a)
361
if err != nil {
362
t.Fatal(err)
363
}
@@ -369,6 +372,8 @@ func TestPinRecursiveFail(t *testing.T) {
372
}
373
374
func TestPinUpdate(t *testing.T) {
375
+ ctx := context.Background()
376
+
377
dstore := dssync.MutexWrap(ds.NewMapDatastore())
378
bstore := blockstore.NewBlockstore(dstore)
379
bserv := bs.New(bstore, offline.Exchange(bstore))
@@ -378,10 +383,9 @@ func TestPinUpdate(t *testing.T) {
383
n1, c1 := randNode()
384
n2, c2 := randNode()
385
381
- dserv.Add(n1)
382
- dserv.Add(n2)
386
+ dserv.Add(ctx, n1)
387
+ dserv.Add(ctx, n2)
388
384
- ctx := context.Background()
389
if err := p.Pin(ctx, n1, true); err != nil {
390
t.Fatal(err)
391
}
pin/set.go
+8
-7
@@ -54,7 +54,7 @@ func (s sortByHash) Swap(a, b int) {
54
s.links[a], s.links[b] = s.links[b], s.links[a]
55
}
56
57
-func storeItems(ctx context.Context, dag merkledag.DAGService, estimatedLen uint64, depth uint32, iter itemIterator, internalKeys keyObserver) (*merkledag.ProtoNode, error) {
57
+func storeItems(ctx context.Context, dag node.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})
@@ -139,10 +139,11 @@ func storeItems(ctx context.Context, dag merkledag.DAGService, estimatedLen uint
139
return nil, err
140
}
141
142
- childKey, err := dag.Add(child)
142
+ err = dag.Add(ctx, child)
143
if err != nil {
144
return nil, err
145
}
146
+ childKey := child.Cid()
147
148
internalKeys(childKey)
149
@@ -202,7 +203,7 @@ func writeHdr(n *merkledag.ProtoNode, hdr *pb.Set) error {
203
204
type walkerFunc func(idx int, link *node.Link) error
205
205
-func walkItems(ctx context.Context, dag merkledag.DAGService, n *merkledag.ProtoNode, fn walkerFunc, children keyObserver) error {
206
+func walkItems(ctx context.Context, dag node.DAGService, n *merkledag.ProtoNode, fn walkerFunc, children keyObserver) error {
207
hdr, err := readHdr(n)
208
if err != nil {
209
return err
@@ -237,7 +238,7 @@ func walkItems(ctx context.Context, dag merkledag.DAGService, n *merkledag.Proto
238
return nil
239
}
240
240
-func loadSet(ctx context.Context, dag merkledag.DAGService, root *merkledag.ProtoNode, name string, internalKeys keyObserver) ([]*cid.Cid, error) {
241
+func loadSet(ctx context.Context, dag node.DAGService, root *merkledag.ProtoNode, name string, internalKeys keyObserver) ([]*cid.Cid, error) {
242
l, err := root.GetNodeLink(name)
243
if err != nil {
244
return nil, err
@@ -280,17 +281,17 @@ func getCidListIterator(cids []*cid.Cid) itemIterator {
281
}
282
}
283
283
-func storeSet(ctx context.Context, dag merkledag.DAGService, cids []*cid.Cid, internalKeys keyObserver) (*merkledag.ProtoNode, error) {
284
+func storeSet(ctx context.Context, dag node.DAGService, cids []*cid.Cid, internalKeys keyObserver) (*merkledag.ProtoNode, error) {
285
iter := getCidListIterator(cids)
286
287
n, err := storeItems(ctx, dag, uint64(len(cids)), 0, iter, internalKeys)
288
if err != nil {
289
return nil, err
290
}
290
- c, err := dag.Add(n)
291
+ err = dag.Add(ctx, n)
292
if err != nil {
293
return nil, err
294
}
294
- internalKeys(c)
295
+ internalKeys(n.Cid())
296
return n, nil
297
}
tar/format.go
+5
-5
@@ -34,7 +34,7 @@ func marshalHeader(h *tar.Header) ([]byte, error) {
34
return buf.Bytes(), nil
35
}
36
37
-func ImportTar(r io.Reader, ds dag.DAGService) (*dag.ProtoNode, error) {
37
+func ImportTar(ctx context.Context, r io.Reader, ds node.DAGService) (*dag.ProtoNode, error) {
38
tr := tar.NewReader(r)
39
40
root := new(dag.ProtoNode)
@@ -73,7 +73,7 @@ func ImportTar(r io.Reader, ds dag.DAGService) (*dag.ProtoNode, error) {
73
}
74
}
75
76
- _, err = ds.Add(header)
76
+ err = ds.Add(ctx, header)
77
if err != nil {
78
return nil, err
79
}
@@ -85,7 +85,7 @@ func ImportTar(r io.Reader, ds dag.DAGService) (*dag.ProtoNode, error) {
85
}
86
}
87
88
- return e.Finalize(ds)
88
+ return e.Finalize(ctx, ds)
89
}
90
91
// adds a '-' to the beginning of each path element so we can use 'data' as a
@@ -100,7 +100,7 @@ func escapePath(pth string) string {
100
101
type tarReader struct {
102
links []*node.Link
103
- ds dag.DAGService
103
+ ds node.DAGService
104
105
childRead *tarReader
106
hdrBuf *bytes.Reader
@@ -194,7 +194,7 @@ func (tr *tarReader) Read(b []byte) (int, error) {
194
return tr.Read(b)
195
}
196
197
-func ExportTar(ctx context.Context, root *dag.ProtoNode, ds dag.DAGService) (io.Reader, error) {
197
+func ExportTar(ctx context.Context, root *dag.ProtoNode, ds node.DAGService) (io.Reader, error) {
198
if string(root.Data()) != "ipfs/tar" {
199
return nil, errors.New("not an IPFS tarchive")
200
}
unixfs/archive/archive.go
+1
-2
@@ -7,7 +7,6 @@ import (
7
"io"
8
"path"
9
10
- mdag "github.com/ipfs/go-ipfs/merkledag"
10
tar "github.com/ipfs/go-ipfs/unixfs/archive/tar"
11
uio "github.com/ipfs/go-ipfs/unixfs/io"
12
@@ -31,7 +30,7 @@ func (i *identityWriteCloser) Close() error {
30
}
31
32
// DagArchive is equivalent to `ipfs getdag $hash | maybe_tar | maybe_gzip`
34
-func DagArchive(ctx context.Context, nd node.Node, name string, dag mdag.DAGService, archive bool, compression int) (io.Reader, error) {
33
+func DagArchive(ctx context.Context, nd node.Node, name string, dag node.DAGService, archive bool, compression int) (io.Reader, error) {
34
35
_, filename := path.Split(name)
36
unixfs/archive/tar/writer.go
+3
-3
@@ -21,14 +21,14 @@ import (
21
// unixfs merkledag nodes as a tar archive format.
22
// It wraps any io.Writer.
23
type Writer struct {
24
- Dag mdag.DAGService
24
+ Dag node.DAGService
25
TarW *tar.Writer
26
27
ctx context.Context
28
}
29
30
// NewWriter wraps given io.Writer.
31
-func NewWriter(ctx context.Context, dag mdag.DAGService, archive bool, compression int, w io.Writer) (*Writer, error) {
31
+func NewWriter(ctx context.Context, dag node.DAGService, archive bool, compression int, w io.Writer) (*Writer, error) {
32
return &Writer{
33
Dag: dag,
34
TarW: tar.NewWriter(w),
@@ -41,7 +41,7 @@ func (w *Writer) writeDir(nd *mdag.ProtoNode, fpath string) error {
41
return err
42
}
43
44
- for i, ng := range mdag.GetDAG(w.ctx, w.Dag, nd) {
44
+ for i, ng := range node.GetDAG(w.ctx, w.Dag, nd) {
45
child, err := ng.Get(w.ctx)
46
if err != nil {
47
return err
unixfs/hamt/hamt.go
+7
-7
@@ -57,7 +57,7 @@ type HamtShard struct {
57
prefixPadStr string
58
maxpadlen int
59
60
- dserv dag.DAGService
60
+ dserv node.DAGService
61
}
62
63
// child can either be another shard, or a leaf node value
@@ -66,7 +66,7 @@ type child interface {
66
Label() string
67
}
68
69
-func NewHamtShard(dserv dag.DAGService, size int) (*HamtShard, error) {
69
+func NewHamtShard(dserv node.DAGService, size int) (*HamtShard, error) {
70
ds, err := makeHamtShard(dserv, size)
71
if err != nil {
72
return nil, err
@@ -78,7 +78,7 @@ func NewHamtShard(dserv dag.DAGService, size int) (*HamtShard, error) {
78
return ds, nil
79
}
80
81
-func makeHamtShard(ds dag.DAGService, size int) (*HamtShard, error) {
81
+func makeHamtShard(ds node.DAGService, size int) (*HamtShard, error) {
82
lg2s := int(math.Log2(float64(size)))
83
if 1<<uint(lg2s) != size {
84
return nil, fmt.Errorf("hamt size should be a power of two")
@@ -93,7 +93,7 @@ func makeHamtShard(ds dag.DAGService, size int) (*HamtShard, error) {
93
}, nil
94
}
95
96
-func NewHamtFromDag(dserv dag.DAGService, nd node.Node) (*HamtShard, error) {
96
+func NewHamtFromDag(dserv node.DAGService, nd node.Node) (*HamtShard, error) {
97
pbnd, ok := nd.(*dag.ProtoNode)
98
if !ok {
99
return nil, dag.ErrLinkNotFound
@@ -184,7 +184,7 @@ func (ds *HamtShard) Node() (node.Node, error) {
184
185
out.SetData(data)
186
187
- _, err = ds.dserv.Add(out)
187
+ err = ds.dserv.Add(context.TODO(), out)
188
if err != nil {
189
return nil, err
190
}
@@ -221,7 +221,7 @@ func (ds *HamtShard) Label() string {
221
// Set sets 'name' = nd in the HAMT
222
func (ds *HamtShard) Set(ctx context.Context, name string, nd node.Node) error {
223
hv := &hashBits{b: hash([]byte(name))}
224
- _, err := ds.dserv.Add(nd)
224
+ err := ds.dserv.Add(ctx, nd)
225
if err != nil {
226
return err
227
}
@@ -335,7 +335,7 @@ func (ds *HamtShard) Link() (*node.Link, error) {
335
return nil, err
336
}
337
338
- _, err = ds.dserv.Add(nd)
338
+ err = ds.dserv.Add(context.TODO(), nd)
339
if err != nil {
340
return nil, err
341
}
unixfs/hamt/hamt_stress_test.go
+5
-4
@@ -8,9 +8,10 @@ import (
8
"testing"
9
"time"
10
11
- dag "github.com/ipfs/go-ipfs/merkledag"
11
mdtest "github.com/ipfs/go-ipfs/merkledag/test"
12
ft "github.com/ipfs/go-ipfs/unixfs"
13
+
14
+ node "gx/ipfs/Qme5bWv7wtjUNGsK2BNGVUFPKiuxWrsqrtvYwCLRw8YFES/go-ipld-format"
15
)
16
17
func getNames(prefix string, count int) []string {
@@ -112,7 +113,7 @@ func validateOpSetCompletion(t *testing.T, s *HamtShard, keep, temp []string) er
113
return nil
114
}
115
115
-func executeOpSet(t *testing.T, ds dag.DAGService, width int, ops []testOp) (*HamtShard, error) {
116
+func executeOpSet(t *testing.T, ds node.DAGService, width int, ops []testOp) (*HamtShard, error) {
117
ctx := context.TODO()
118
s, err := NewHamtShard(ds, width)
119
if err != nil {
@@ -120,7 +121,7 @@ func executeOpSet(t *testing.T, ds dag.DAGService, width int, ops []testOp) (*Ha
121
}
122
123
e := ft.EmptyDirNode()
123
- ds.Add(e)
124
+ ds.Add(ctx, e)
125
126
for _, o := range ops {
127
switch o.Op {
@@ -188,7 +189,7 @@ func genOpSet(seed int64, keep, temp []string) []testOp {
189
}
190
191
// executes the given op set with a repl to allow easier debugging
191
-/*func debugExecuteOpSet(ds dag.DAGService, width int, ops []testOp) (*HamtShard, error) {
192
+/*func debugExecuteOpSet(ds node.DAGService, width int, ops []testOp) (*HamtShard, error) {
193
194
s, err := NewHamtShard(ds, width)
195
if err != nil {
unixfs/hamt/hamt_test.go
+17
-10
@@ -13,6 +13,8 @@ import (
13
mdtest "github.com/ipfs/go-ipfs/merkledag/test"
14
dagutils "github.com/ipfs/go-ipfs/merkledag/utils"
15
ft "github.com/ipfs/go-ipfs/unixfs"
16
+
17
+ node "gx/ipfs/Qme5bWv7wtjUNGsK2BNGVUFPKiuxWrsqrtvYwCLRw8YFES/go-ipld-format"
18
)
19
20
func shuffle(seed int64, arr []string) {
@@ -24,11 +26,13 @@ func shuffle(seed int64, arr []string) {
26
}
27
}
28
27
-func makeDir(ds dag.DAGService, size int) ([]string, *HamtShard, error) {
29
+func makeDir(ds node.DAGService, size int) ([]string, *HamtShard, error) {
30
return makeDirWidth(ds, size, 256)
31
}
32
31
-func makeDirWidth(ds dag.DAGService, size, width int) ([]string, *HamtShard, error) {
33
+func makeDirWidth(ds node.DAGService, size, width int) ([]string, *HamtShard, error) {
34
+ ctx := context.Background()
35
+
36
s, _ := NewHamtShard(ds, width)
37
38
var dirs []string
@@ -40,8 +44,8 @@ func makeDirWidth(ds dag.DAGService, size, width int) ([]string, *HamtShard, err
44
45
for i := 0; i < len(dirs); i++ {
46
nd := ft.EmptyDirNode()
43
- ds.Add(nd)
44
- err := s.Set(context.Background(), dirs[i], nd)
47
+ ds.Add(ctx, nd)
48
+ err := s.Set(ctx, dirs[i], nd)
49
if err != nil {
50
return nil, nil, err
51
}
@@ -70,7 +74,7 @@ func assertLink(s *HamtShard, name string, found bool) error {
74
}
75
}
76
73
-func assertSerializationWorks(ds dag.DAGService, s *HamtShard) error {
77
+func assertSerializationWorks(ds node.DAGService, s *HamtShard) error {
78
ctx, cancel := context.WithCancel(context.Background())
79
defer cancel()
80
nd, err := s.Node()
@@ -459,12 +463,13 @@ func TestBitfieldIndexing(t *testing.T) {
463
// if improperly implemented, the parent hamt may assume the child is a part of
464
// itself.
465
func TestSetHamtChild(t *testing.T) {
466
+ ctx := context.Background()
467
+
468
ds := mdtest.Mock()
469
s, _ := NewHamtShard(ds, 256)
464
- ctx := context.Background()
470
471
e := ft.EmptyDirNode()
467
- ds.Add(e)
472
+ ds.Add(ctx, e)
473
474
err := s.Set(ctx, "bar", e)
475
if err != nil {
@@ -507,7 +512,7 @@ func TestSetHamtChild(t *testing.T) {
512
}
513
}
514
510
-func printDiff(ds dag.DAGService, a, b *dag.ProtoNode) {
515
+func printDiff(ds node.DAGService, a, b *dag.ProtoNode) {
516
diff, err := dagutils.Diff(context.TODO(), ds, a, b)
517
if err != nil {
518
panic(err)
@@ -519,6 +524,8 @@ func printDiff(ds dag.DAGService, a, b *dag.ProtoNode) {
524
}
525
526
func BenchmarkHAMTSet(b *testing.B) {
527
+ ctx := context.Background()
528
+
529
ds := mdtest.Mock()
530
sh, _ := NewHamtShard(ds, 256)
531
nd, err := sh.Node()
@@ -526,11 +533,11 @@ func BenchmarkHAMTSet(b *testing.B) {
533
b.Fatal(err)
534
}
535
529
- _, err = ds.Add(nd)
536
+ err = ds.Add(ctx, nd)
537
if err != nil {
538
b.Fatal(err)
539
}
533
- ds.Add(ft.EmptyDirNode())
540
+ ds.Add(ctx, ft.EmptyDirNode())
541
542
for i := 0; i < b.N; i++ {
543
s, err := NewHamtFromDag(ds, nd)
unixfs/io/dagreader.go
+1
-1
@@ -34,7 +34,7 @@ type ReadSeekCloser interface {
34
35
// NewDagReader creates a new reader object that reads the data represented by
36
// the given node, using the passed in DAGService for data retreival
37
-func NewDagReader(ctx context.Context, n node.Node, serv mdag.DAGService) (DagReader, error) {
37
+func NewDagReader(ctx context.Context, n node.Node, serv node.DAGService) (DagReader, error) {
38
switch n := n.(type) {
39
case *mdag.RawNode:
40
return NewBufDagReader(n.RawData()), nil
unixfs/io/dagreader_test.go
+4
-4
@@ -209,16 +209,16 @@ func TestBadPBData(t *testing.T) {
209
}
210
211
func TestMetadataNode(t *testing.T) {
212
+ ctx, closer := context.WithCancel(context.Background())
213
+ defer closer()
214
+
215
dserv := testu.GetDAGServ()
216
rdata, rnode := testu.GetRandomNode(t, dserv, 512, testu.UseProtoBufLeaves)
214
- _, err := dserv.Add(rnode)
217
+ err := dserv.Add(ctx, rnode)
218
if err != nil {
219
t.Fatal(err)
220
}
221
219
- ctx, closer := context.WithCancel(context.Background())
220
- defer closer()
221
-
222
data, err := unixfs.BytesForMetadata(&unixfs.Metadata{
223
MimeType: "text",
224
Size: 125,
unixfs/io/dirbuilder.go
+3
-3
@@ -26,14 +26,14 @@ var UseHAMTSharding = false
26
var DefaultShardWidth = 256
27
28
type Directory struct {
29
- dserv mdag.DAGService
29
+ dserv node.DAGService
30
dirnode *mdag.ProtoNode
31
32
shard *hamt.HamtShard
33
}
34
35
// NewDirectory returns a Directory. It needs a DAGService to add the Children
36
-func NewDirectory(dserv mdag.DAGService) *Directory {
36
+func NewDirectory(dserv node.DAGService) *Directory {
37
db := new(Directory)
38
db.dserv = dserv
39
if UseHAMTSharding {
@@ -51,7 +51,7 @@ func NewDirectory(dserv mdag.DAGService) *Directory {
51
// ErrNotADir implies that the given node was not a unixfs directory
52
var ErrNotADir = fmt.Errorf("merkledag node was not a directory or shard")
53
54
-func NewDirectoryFromNode(dserv mdag.DAGService, nd node.Node) (*Directory, error) {
54
+func NewDirectoryFromNode(dserv node.DAGService, nd node.Node) (*Directory, error) {
55
pbnd, ok := nd.(*mdag.ProtoNode)
56
if !ok {
57
return nil, ErrNotADir
unixfs/io/dirbuilder_test.go
+2
-2
@@ -22,7 +22,7 @@ func TestDirectoryGrowth(t *testing.T) {
22
ctx := context.Background()
23
24
d := ft.EmptyDirNode()
25
- ds.Add(d)
25
+ ds.Add(ctx, d)
26
27
nelems := 10000
28
@@ -102,7 +102,7 @@ func TestDirBuilder(t *testing.T) {
102
ctx := context.Background()
103
104
child := ft.EmptyDirNode()
105
- _, err := ds.Add(child)
105
+ err := ds.Add(ctx, child)
106
if err != nil {
107
t.Fatal(err)
108
}
unixfs/io/pbdagreader.go
+6
-6
@@ -17,7 +17,7 @@ import (
17
18
// DagReader provides a way to easily read the data contained in a dag.
19
type pbDagReader struct {
20
- serv mdag.DAGService
20
+ serv node.DAGService
21
22
// the node being read
23
node *mdag.ProtoNode
@@ -29,8 +29,8 @@ type pbDagReader struct {
29
// will either be a bytes.Reader or a child DagReader
30
buf ReadSeekCloser
31
32
- // NodeGetters for each of 'nodes' child links
33
- promises []mdag.NodeGetter
32
+ // NodePromises for each of 'nodes' child links
33
+ promises []*node.NodePromise
34
35
// the cid of each child of the current node
36
links []*cid.Cid
@@ -50,14 +50,14 @@ type pbDagReader struct {
50
51
var _ DagReader = (*pbDagReader)(nil)
52
53
-func NewPBFileReader(ctx context.Context, n *mdag.ProtoNode, pb *ftpb.Data, serv mdag.DAGService) *pbDagReader {
53
+func NewPBFileReader(ctx context.Context, n *mdag.ProtoNode, pb *ftpb.Data, serv node.DAGService) *pbDagReader {
54
fctx, cancel := context.WithCancel(ctx)
55
curLinks := getLinkCids(n)
56
return &pbDagReader{
57
node: n,
58
serv: serv,
59
buf: NewBufDagReader(pb.GetData()),
60
- promises: make([]mdag.NodeGetter, len(curLinks)),
60
+ promises: make([]*node.NodePromise, len(curLinks)),
61
links: curLinks,
62
ctx: fctx,
63
cancel: cancel,
@@ -74,7 +74,7 @@ func (dr *pbDagReader) preloadNextNodes(ctx context.Context) {
74
end = len(dr.links)
75
}
76
77
- for i, p := range mdag.GetNodes(ctx, dr.serv, dr.links[beg:end]) {
77
+ for i, p := range node.GetNodes(ctx, dr.serv, dr.links[beg:end]) {
78
dr.promises[beg+i] = p
79
}
80
}
unixfs/io/resolve.go
+1
-1
@@ -12,7 +12,7 @@ import (
12
13
// ResolveUnixfsOnce resolves a single hop of a path through a graph in a
14
// unixfs context. This includes handling traversing sharded directories.
15
-func ResolveUnixfsOnce(ctx context.Context, ds dag.DAGService, nd node.Node, names []string) (*node.Link, []string, error) {
15
+func ResolveUnixfsOnce(ctx context.Context, ds node.DAGService, nd node.Node, names []string) (*node.Link, []string, error) {
16
switch nd := nd.(type) {
17
case *dag.ProtoNode:
18
upb, err := ft.FromBytes(nd.Data())
unixfs/mod/dagmodifier.go
+13
-13
@@ -29,7 +29,7 @@ var writebufferSize = 1 << 21
29
// perform surgery on a DAG 'file'
30
// Dear god, please rename this to something more pleasant
31
type DagModifier struct {
32
- dagserv mdag.DAGService
32
+ dagserv node.DAGService
33
curNode node.Node
34
35
splitter chunk.SplitterGen
@@ -52,7 +52,7 @@ var ErrNotUnixfs = fmt.Errorf("dagmodifier only supports unixfs nodes (proto or
52
// created nodes will be inherted from the passed in node. If the Cid
53
// version if not 0 raw leaves will also be enabled. The Prefix and
54
// RawLeaves options can be overridden by changing them after the call.
55
-func NewDagModifier(ctx context.Context, from node.Node, serv mdag.DAGService, spl chunk.SplitterGen) (*DagModifier, error) {
55
+func NewDagModifier(ctx context.Context, from node.Node, serv node.DAGService, spl chunk.SplitterGen) (*DagModifier, error) {
56
switch from.(type) {
57
case *mdag.ProtoNode, *mdag.RawNode:
58
// ok
@@ -128,7 +128,7 @@ func (dm *DagModifier) expandSparse(size int64) error {
128
if err != nil {
129
return err
130
}
131
- _, err = dm.dagserv.Add(nnode)
131
+ err = dm.dagserv.Add(dm.ctx, nnode)
132
return err
133
}
134
@@ -216,7 +216,7 @@ func (dm *DagModifier) Sync() error {
216
return err
217
}
218
219
- _, err = dm.dagserv.Add(dm.curNode)
219
+ err = dm.dagserv.Add(dm.ctx, dm.curNode)
220
if err != nil {
221
return err
222
}
@@ -255,7 +255,7 @@ func (dm *DagModifier) modifyDag(n node.Node, offset uint64, data io.Reader) (*c
255
nd := new(mdag.ProtoNode)
256
nd.SetData(b)
257
nd.SetPrefix(&nd0.Prefix)
258
- k, err := dm.dagserv.Add(nd)
258
+ err = dm.dagserv.Add(dm.ctx, nd)
259
if err != nil {
260
return nil, false, err
261
}
@@ -266,7 +266,7 @@ func (dm *DagModifier) modifyDag(n node.Node, offset uint64, data io.Reader) (*c
266
done = true
267
}
268
269
- return k, done, nil
269
+ return nd.Cid(), done, nil
270
case *mdag.RawNode:
271
origData := nd0.RawData()
272
bytes := make([]byte, len(origData))
@@ -290,7 +290,7 @@ func (dm *DagModifier) modifyDag(n node.Node, offset uint64, data io.Reader) (*c
290
if err != nil {
291
return nil, false, err
292
}
293
- k, err := dm.dagserv.Add(nd)
293
+ err = dm.dagserv.Add(dm.ctx, nd)
294
if err != nil {
295
return nil, false, err
296
}
@@ -301,7 +301,7 @@ func (dm *DagModifier) modifyDag(n node.Node, offset uint64, data io.Reader) (*c
301
done = true
302
}
303
304
- return k, done, nil
304
+ return nd.Cid(), done, nil
305
}
306
}
307
@@ -348,8 +348,8 @@ func (dm *DagModifier) modifyDag(n node.Node, offset uint64, data io.Reader) (*c
348
cur += bs
349
}
350
351
- k, err := dm.dagserv.Add(node)
352
- return k, done, err
351
+ err = dm.dagserv.Add(dm.ctx, node)
352
+ return node.Cid(), done, err
353
}
354
355
// appendData appends the blocks from the given chan to the end of this dag
@@ -500,7 +500,7 @@ func (dm *DagModifier) Truncate(size int64) error {
500
return err
501
}
502
503
- _, err = dm.dagserv.Add(nnode)
503
+ err = dm.dagserv.Add(dm.ctx, nnode)
504
if err != nil {
505
return err
506
}
@@ -510,7 +510,7 @@ func (dm *DagModifier) Truncate(size int64) error {
510
}
511
512
// dagTruncate truncates the given node to 'size' and returns the modified Node
513
-func dagTruncate(ctx context.Context, n node.Node, size uint64, ds mdag.DAGService) (node.Node, error) {
513
+func dagTruncate(ctx context.Context, n node.Node, size uint64, ds node.DAGService) (node.Node, error) {
514
if len(n.Links()) == 0 {
515
switch nd := n.(type) {
516
case *mdag.ProtoNode:
@@ -563,7 +563,7 @@ func dagTruncate(ctx context.Context, n node.Node, size uint64, ds mdag.DAGServi
563
ndata.AddBlockSize(childsize)
564
}
565
566
- _, err := ds.Add(modified)
566
+ err := ds.Add(ctx, modified)
567
if err != nil {
568
return nil, err
569
}
unixfs/test/utils.go
+5
-5
@@ -27,7 +27,7 @@ func SizeSplitterGen(size int64) chunk.SplitterGen {
27
}
28
}
29
30
-func GetDAGServ() mdag.DAGService {
30
+func GetDAGServ() node.DAGService {
31
return mdagmock.Mock()
32
}
33
@@ -51,7 +51,7 @@ func init() {
51
UseBlake2b256.Prefix.MhLength = -1
52
}
53
54
-func GetNode(t testing.TB, dserv mdag.DAGService, data []byte, opts NodeOpts) node.Node {
54
+func GetNode(t testing.TB, dserv node.DAGService, data []byte, opts NodeOpts) node.Node {
55
in := bytes.NewReader(data)
56
57
dbp := h.DagBuilderParams{
@@ -69,11 +69,11 @@ func GetNode(t testing.TB, dserv mdag.DAGService, data []byte, opts NodeOpts) no
69
return node
70
}
71
72
-func GetEmptyNode(t testing.TB, dserv mdag.DAGService, opts NodeOpts) node.Node {
72
+func GetEmptyNode(t testing.TB, dserv node.DAGService, opts NodeOpts) node.Node {
73
return GetNode(t, dserv, []byte{}, opts)
74
}
75
76
-func GetRandomNode(t testing.TB, dserv mdag.DAGService, size int64, opts NodeOpts) ([]byte, node.Node) {
76
+func GetRandomNode(t testing.TB, dserv node.DAGService, size int64, opts NodeOpts) ([]byte, node.Node) {
77
in := io.LimitReader(u.NewTimeSeededRand(), size)
78
buf, err := ioutil.ReadAll(in)
79
if err != nil {
@@ -96,7 +96,7 @@ func ArrComp(a, b []byte) error {
96
return nil
97
}
98
99
-func PrintDag(nd *mdag.ProtoNode, ds mdag.DAGService, indent int) {
99
+func PrintDag(nd *mdag.ProtoNode, ds node.DAGService, indent int) {
100
pbd, err := ft.FromBytes(nd.Data())
101
if err != nil {
102
panic(err)