Refactor EnumerateChildrenAsync to take in a function to get the links.
For now it is always called with the helper function GetLinksDirect to avoid any change in behaviour. License: MIT Signed-off-by: Kevin Atkinson <k@kevina.org>
Kevin Atkinson committed
Feb 16, 2017 at 21:29 UTC
c3346ad9d5a2f5c652029ac52fd5e3e1a81114fa
3 files changed
+24
-11
core/commands/dht.go
+1
-1
@@ -370,7 +370,7 @@ func provideKeysRec(ctx context.Context, r routing.IpfsRouting, dserv dag.DAGSer
370
for _, c := range cids {
371
kset := cid.NewSet()
372
373
- err := dag.EnumerateChildrenAsync(ctx, dserv, c, kset.Visit)
373
+ err := dag.EnumerateChildrenAsync(ctx, dag.GetLinksDirect(dserv), c, kset.Visit)
374
if err != nil {
375
return err
376
}
merkledag/merkledag.go
+22
-9
@@ -138,11 +138,23 @@ func (n *dagService) Remove(nd node.Node) error {
138
return n.Blocks.DeleteBlock(nd)
139
}
140
141
+// get the links for a node, from the node, bypassing the
142
+// LinkService
143
+func GetLinksDirect(serv DAGService) GetLinks {
144
+ return func(ctx context.Context, c *cid.Cid) ([]*node.Link, error) {
145
+ node, err := serv.Get(ctx, c)
146
+ if err != nil {
147
+ return nil, err
148
+ }
149
+ return node.Links(), nil
150
+ }
151
+}
152
+
153
// FetchGraph fetches all nodes that are children of the given node
154
func FetchGraph(ctx context.Context, root *cid.Cid, serv DAGService) error {
155
v, _ := ctx.Value("progress").(*ProgressTracker)
156
if v == nil {
145
- return EnumerateChildrenAsync(ctx, serv, root, cid.NewSet().Visit)
157
+ return EnumerateChildrenAsync(ctx, GetLinksDirect(serv), root, cid.NewSet().Visit)
158
}
159
set := cid.NewSet()
160
visit := func(c *cid.Cid) bool {
@@ -153,7 +165,7 @@ func FetchGraph(ctx context.Context, root *cid.Cid, serv DAGService) error {
165
return false
166
}
167
}
156
- return EnumerateChildrenAsync(ctx, serv, root, visit)
168
+ return EnumerateChildrenAsync(ctx, GetLinksDirect(serv), root, visit)
169
}
170
171
// FindLinks searches this nodes links for the given key,
@@ -380,10 +392,11 @@ func (t *Batch) Commit() error {
392
return err
393
}
394
395
+type GetLinks func(context.Context, *cid.Cid) ([]*node.Link, error)
396
+
397
// EnumerateChildren will walk the dag below the given root node and add all
398
// unseen children to the passed in set.
399
// TODO: parallelize to avoid disk latency perf hits?
386
-type GetLinks func(context.Context, *cid.Cid) ([]*node.Link, error)
400
func EnumerateChildren(ctx context.Context, getLinks GetLinks, root *cid.Cid, visit func(*cid.Cid) bool) error {
401
links, err := getLinks(ctx, root)
402
if err != nil {
@@ -426,9 +439,9 @@ func (p *ProgressTracker) Value() int {
439
// 'fetchNodes' will start at a time
440
var FetchGraphConcurrency = 8
441
429
-func EnumerateChildrenAsync(ctx context.Context, ds DAGService, c *cid.Cid, visit func(*cid.Cid) bool) error {
442
+func EnumerateChildrenAsync(ctx context.Context, getLinks GetLinks, c *cid.Cid, visit func(*cid.Cid) bool) error {
443
feed := make(chan *cid.Cid)
431
- out := make(chan node.Node)
444
+ out := make(chan []*node.Link)
445
done := make(chan struct{})
446
447
var setlk sync.Mutex
@@ -441,7 +454,7 @@ func EnumerateChildrenAsync(ctx context.Context, ds DAGService, c *cid.Cid, visi
454
for i := 0; i < FetchGraphConcurrency; i++ {
455
go func() {
456
for ic := range feed {
444
- n, err := ds.Get(ctx, ic)
457
+ links, err := getLinks(ctx, ic)
458
if err != nil {
459
errChan <- err
460
return
@@ -453,7 +466,7 @@ func EnumerateChildrenAsync(ctx context.Context, ds DAGService, c *cid.Cid, visi
466
467
if unseen {
468
select {
456
- case out <- n:
469
+ case out <- links:
470
case <-fetchersCtx.Done():
471
return
472
}
@@ -488,8 +501,8 @@ func EnumerateChildrenAsync(ctx context.Context, ds DAGService, c *cid.Cid, visi
501
if inProgress == 0 && next == nil {
502
return nil
503
}
491
- case nd := <-out:
492
- for _, lnk := range nd.Links() {
504
+ case links := <-out:
505
+ for _, lnk := range links {
506
if next == nil {
507
next = lnk.Cid
508
send = feed
merkledag/merkledag_test.go
+1
-1
@@ -543,7 +543,7 @@ func TestEnumerateAsyncFailsNotFound(t *testing.T) {
543
}
544
545
cset := cid.NewSet()
546
- err = EnumerateChildrenAsync(context.Background(), ds, pcid, cset.Visit)
546
+ err = EnumerateChildrenAsync(context.Background(), GetLinksDirect(ds), pcid, cset.Visit)
547
if err == nil {
548
t.Fatal("this should have failed")
549
}