merkledag FetchGraph and EnumerateChildren
This commit improves (fixes) the FetchGraph call for recursively fetching every descendant node of a given merkledag node. This operation should be the simplest way of ensuring that you have replicated a dag locally. This commit also implements a method in the merkledag package called EnumerateChildren, this method is used to get a set of the keys of every descendant node of the given node. All keys found are noted in the passed in KeySet, which may in the future be implemented on disk to avoid excessive memory consumption. License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com>
Jeromy committed
Jul 7, 2015 at 09:04 UTC
b9e5cfaa15ea2d105e3dba653eaa05be3a142645
3 files changed
+170
-30
core/core.go
+1
-1
@@ -90,7 +90,7 @@ type IpfsNode struct {
90
91
// Services
92
Peerstore peer.Peerstore // storage for other Peer instances
93
- Blockstore bstore.Blockstore // the block store (lower level)
93
+ Blockstore bstore.GCBlockstore // the block store (lower level)
94
Blocks *bserv.BlockService // the block service, get/add blocks.
95
DAG merkledag.DAGService // the merkle dag service, get/add objects.
96
Resolver *path.Resolver // the path resolution system
merkledag/merkledag.go
+92
-27
@@ -3,7 +3,6 @@ package merkledag
3
4
import (
5
"fmt"
6
- "sync"
6
7
"github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
8
blocks "github.com/ipfs/go-ipfs/blocks"
@@ -121,41 +120,86 @@ func (n *dagService) Remove(nd *Node) error {
120
return n.Blocks.DeleteBlock(k)
121
}
122
124
-// FetchGraph asynchronously fetches all nodes that are children of the given
125
-// node, and returns a channel that may be waited upon for the fetch to complete
126
-func FetchGraph(ctx context.Context, root *Node, serv DAGService) chan struct{} {
127
- log.Warning("Untested.")
128
- var wg sync.WaitGroup
129
- done := make(chan struct{})
123
+// FetchGraph fetches all nodes that are children of the given node
124
+func FetchGraph(ctx context.Context, root *Node, serv DAGService) error {
125
+ toprocess := make(chan []key.Key, 8)
126
+ nodes := make(chan *Node, 8)
127
+ errs := make(chan error, 1)
128
131
- for _, l := range root.Links {
132
- wg.Add(1)
133
- go func(lnk *Link) {
129
+ ctx, cancel := context.WithCancel(ctx)
130
+ defer cancel()
131
+ defer close(toprocess)
132
135
- // Signal child is done on way out
136
- defer wg.Done()
137
- select {
138
- case <-ctx.Done():
139
- return
133
+ go fetchNodes(ctx, serv, toprocess, nodes, errs)
134
+
135
+ nodes <- root
136
+ live := 1
137
+
138
+ for {
139
+ select {
140
+ case nd, ok := <-nodes:
141
+ if !ok {
142
+ return nil
143
}
144
142
- nd, err := lnk.GetNode(ctx, serv)
143
- if err != nil {
144
- log.Debug(err)
145
- return
145
+ var keys []key.Key
146
+ for _, lnk := range nd.Links {
147
+ keys = append(keys, key.Key(lnk.Hash))
148
}
149
+ keys = dedupeKeys(keys)
150
148
- // Wait for children to finish
149
- <-FetchGraph(ctx, nd, serv)
150
- }(l)
151
+ // keep track of open request, when zero, we're done
152
+ live += len(keys) - 1
153
+
154
+ if live == 0 {
155
+ return nil
156
+ }
157
+
158
+ if len(keys) > 0 {
159
+ select {
160
+ case toprocess <- keys:
161
+ case <-ctx.Done():
162
+ return ctx.Err()
163
+ }
164
+ }
165
+ case err := <-errs:
166
+ return err
167
+ case <-ctx.Done():
168
+ return ctx.Err()
169
+ }
170
}
171
+}
172
153
- go func() {
154
- wg.Wait()
155
- done <- struct{}{}
156
- }()
173
+func fetchNodes(ctx context.Context, ds DAGService, in <-chan []key.Key, out chan<- *Node, errs chan<- error) {
174
+ defer close(out)
175
+ for {
176
+ select {
177
+ case ks, ok := <-in:
178
+ if !ok {
179
+ return
180
+ }
181
158
- return done
182
+ ng := ds.GetNodes(ctx, ks)
183
+ for _, g := range ng {
184
+ go func(g NodeGetter) {
185
+ nd, err := g.Get(ctx)
186
+ if err != nil {
187
+ select {
188
+ case errs <- err:
189
+ case <-ctx.Done():
190
+ }
191
+ return
192
+ }
193
+
194
+ select {
195
+ case out <- nd:
196
+ case <-ctx.Done():
197
+ return
198
+ }
199
+ }(g)
200
+ }
201
+ }
202
+ }
203
}
204
205
// FindLinks searches this nodes links for the given key,
@@ -318,3 +362,24 @@ func (t *Batch) Commit() error {
362
t.size = 0
363
return err
364
}
365
+
366
+// EnumerateChildren will walk the dag below the given root node and add all
367
+// unseen children to the passed in set.
368
+// TODO: parallelize to avoid disk latency perf hits?
369
+func EnumerateChildren(ctx context.Context, ds DAGService, root *Node, set key.KeySet) error {
370
+ for _, lnk := range root.Links {
371
+ k := key.Key(lnk.Hash)
372
+ if !set.Has(k) {
373
+ set.Add(k)
374
+ child, err := ds.Get(ctx, k)
375
+ if err != nil {
376
+ return err
377
+ }
378
+ err = EnumerateChildren(ctx, ds, child, set)
379
+ if err != nil {
380
+ return err
381
+ }
382
+ }
383
+ }
384
+ return nil
385
+}
merkledag/merkledag_test.go
+77
-2
@@ -130,7 +130,7 @@ func SubtestNodeStat(t *testing.T, n *Node) {
130
}
131
132
if expected != *actual {
133
- t.Errorf("n.Stat incorrect.\nexpect: %s\nactual: %s", expected, actual)
133
+ t.Error("n.Stat incorrect.\nexpect: %s\nactual: %s", expected, actual)
134
} else {
135
fmt.Printf("n.Stat correct: %s\n", actual)
136
}
@@ -232,7 +232,6 @@ func runBatchFetchTest(t *testing.T, read io.Reader) {
232
}
233
}
234
}
235
-
235
func TestRecursiveAdd(t *testing.T) {
236
a := &Node{Data: []byte("A")}
237
b := &Node{Data: []byte("B")}
@@ -298,3 +297,79 @@ func TestCantGet(t *testing.T) {
297
t.Fatal("expected err not found, got: ", err)
298
}
299
}
300
+
301
+func TestFetchGraph(t *testing.T) {
302
+ bsi := bstest.Mocks(t, 1)[0]
303
+ ds := NewDAGService(bsi)
304
+
305
+ read := io.LimitReader(u.NewTimeSeededRand(), 1024*32)
306
+ spl := &chunk.SizeSplitter{512}
307
+
308
+ root, err := imp.BuildDagFromReader(read, ds, spl, nil)
309
+ if err != nil {
310
+ t.Fatal(err)
311
+ }
312
+
313
+ err = FetchGraph(context.TODO(), root, ds)
314
+ if err != nil {
315
+ t.Fatal(err)
316
+ }
317
+}
318
+
319
+func TestFetchGraphOther(t *testing.T) {
320
+ var dservs []DAGService
321
+ for _, bsi := range bstest.Mocks(t, 2) {
322
+ dservs = append(dservs, NewDAGService(bsi))
323
+ }
324
+
325
+ read := io.LimitReader(u.NewTimeSeededRand(), 1024*32)
326
+ spl := &chunk.SizeSplitter{512}
327
+
328
+ root, err := imp.BuildDagFromReader(read, dservs[0], spl, nil)
329
+ if err != nil {
330
+ t.Fatal(err)
331
+ }
332
+
333
+ err = FetchGraph(context.TODO(), root, dservs[1])
334
+ if err != nil {
335
+ t.Fatal(err)
336
+ }
337
+}
338
+
339
+func TestEnumerateChildren(t *testing.T) {
340
+ bsi := bstest.Mocks(t, 1)
341
+ ds := NewDAGService(bsi[0])
342
+
343
+ spl := &chunk.SizeSplitter{512}
344
+
345
+ read := io.LimitReader(u.NewTimeSeededRand(), 1024*1024)
346
+
347
+ root, err := imp.BuildDagFromReader(read, ds, spl, nil)
348
+ if err != nil {
349
+ t.Fatal(err)
350
+ }
351
+
352
+ ks := key.NewKeySet()
353
+ err = EnumerateChildren(context.Background(), ds, root, ks)
354
+ if err != nil {
355
+ t.Fatal(err)
356
+ }
357
+
358
+ var traverse func(n *Node)
359
+ traverse = func(n *Node) {
360
+ // traverse dag and check
361
+ for _, lnk := range n.Links {
362
+ k := key.Key(lnk.Hash)
363
+ if !ks.Has(k) {
364
+ t.Fatal("missing key in set!")
365
+ }
366
+ child, err := ds.Get(context.Background(), k)
367
+ if err != nil {
368
+ t.Fatal(err)
369
+ }
370
+ traverse(child)
371
+ }
372
+ }
373
+
374
+ traverse(root)
375
+}