Skip to content

Commit 2fd045f

Browse files
whyrusleepingwhyrusleeping
authored
Merge pull request #3255 from ipfs/kevina/getlinks
Add DAGService.GetLinks() method and use it in the GC and elsewhere.
2 parents 67a1b3e + 772164c commit 2fd045f

File tree

12 files changed

+92
-72
lines changed

12 files changed

+92
-72
lines changed

‎core/commands/dht.go

+1-5
Original file line numberDiff line numberDiff line change
@@ -370,12 +370,8 @@ func provideKeysRec(ctx context.Context, r routing.IpfsRouting, dserv dag.DAGSer
370370
provided := cid.NewSet()
371371
for _, c := range cids {
372372
kset := cid.NewSet()
373-
node, err := dserv.Get(ctx, c)
374-
if err != nil {
375-
return err
376-
}
377373

378-
err = dag.EnumerateChildrenAsync(ctx, dserv, node, kset.Visit)
374+
err := dag.EnumerateChildrenAsync(ctx, dserv, c, kset.Visit)
379375
if err != nil {
380376
return err
381377
}

‎core/commands/pin.go

+1-6
Original file line numberDiff line numberDiff line change
@@ -328,12 +328,7 @@ func pinLsAll(typeStr string, ctx context.Context, n *core.IpfsNode) (map[string
328328
if typeStr == "indirect" || typeStr == "all" {
329329
set := cid.NewSet()
330330
for _, k := range n.Pinning.RecursiveKeys() {
331-
nd, err := n.DAG.Get(ctx, k)
332-
if err != nil {
333-
return nil, err
334-
}
335-
336-
err = dag.EnumerateChildren(n.Context(), n.DAG, nd, set.Visit, false)
331+
err := dag.EnumerateChildren(n.Context(), n.DAG, k, set.Visit, false)
337332
if err != nil {
338333
return nil, err
339334
}

‎core/corerepo/gc.go

+2-2
Original file line numberDiff line numberDiff line change
@@ -90,7 +90,7 @@ func GarbageCollect(n *core.IpfsNode, ctx context.Context) error {
9090
if err != nil {
9191
return err
9292
}
93-
rmed, err := gc.GC(ctx, n.Blockstore, n.Pinning, roots)
93+
rmed, err := gc.GC(ctx, n.Blockstore, n.DAG, n.Pinning, roots)
9494
if err != nil {
9595
return err
9696
}
@@ -113,7 +113,7 @@ func GarbageCollectAsync(n *core.IpfsNode, ctx context.Context) (<-chan *KeyRemo
113113
if err != nil {
114114
return nil, err
115115
}
116-
rmed, err := gc.GC(ctx, n.Blockstore, n.Pinning, roots)
116+
rmed, err := gc.GC(ctx, n.Blockstore, n.DAG, n.Pinning, roots)
117117
if err != nil {
118118
return nil, err
119119
}

‎core/coreunix/add_test.go

+2-6
Original file line numberDiff line numberDiff line change
@@ -98,7 +98,7 @@ func TestAddGCLive(t *testing.T) {
9898
gcstarted := make(chan struct{})
9999
go func() {
100100
defer close(gcstarted)
101-
gcchan, err := gc.GC(context.Background(), node.Blockstore, node.Pinning, nil)
101+
gcchan, err := gc.GC(context.Background(), node.Blockstore, node.DAG, node.Pinning, nil)
102102
if err != nil {
103103
log.Error("GC ERROR:", err)
104104
errs <- err
@@ -156,13 +156,9 @@ func TestAddGCLive(t *testing.T) {
156156

157157
ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
158158
defer cancel()
159-
root, err := node.DAG.Get(ctx, last)
160-
if err != nil {
161-
t.Fatal(err)
162-
}
163159

164160
set := cid.NewSet()
165-
err = dag.EnumerateChildren(ctx, node.DAG, root, set.Visit, false)
161+
err = dag.EnumerateChildren(ctx, node.DAG, last, set.Visit, false)
166162
if err != nil {
167163
t.Fatal(err)
168164
}

‎exchange/bitswap/bitswap.go

+4
Original file line numberDiff line numberDiff line change
@@ -422,3 +422,7 @@ func (bs *Bitswap) GetWantlist() []key.Key {
422422
}
423423
return out
424424
}
425+
426+
func (bs *Bitswap) IsOnline() bool {
427+
return true
428+
}

‎exchange/interface.go

+2
Original file line numberDiff line numberDiff line change
@@ -22,5 +22,7 @@ type Interface interface { // type Exchanger interface
2222
// available on the network?
2323
HasBlock(blocks.Block) error
2424

25+
IsOnline() bool
26+
2527
io.Closer
2628
}

‎exchange/offline/offline.go

+4
Original file line numberDiff line numberDiff line change
@@ -67,3 +67,7 @@ func (e *offlineExchange) GetBlocks(ctx context.Context, ks []key.Key) (<-chan b
6767
}()
6868
return out, nil
6969
}
70+
71+
func (e *offlineExchange) IsOnline() bool {
72+
return false
73+
}

‎merkledag/merkledag.go

+47-16
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ import (
77
"sync"
88

99
bserv "github.com/ipfs/go-ipfs/blockservice"
10+
offline "github.com/ipfs/go-ipfs/exchange/offline"
1011
key "gx/ipfs/QmYEoKZXHoAToWfhGF3vryhMn3WWhE1o2MasQ8uzY5iDi9/go-key"
1112

1213
"context"
@@ -28,10 +29,20 @@ type DAGService interface {
2829
GetMany(context.Context, []*cid.Cid) <-chan *NodeOption
2930

3031
Batch() *Batch
32+
33+
LinkService
34+
}
35+
36+
type LinkService interface {
37+
// Return all links for a node, may be more effect than
38+
// calling Get in DAGService
39+
GetLinks(context.Context, *cid.Cid) ([]*Link, error)
40+
41+
GetOfflineLinkService() LinkService
3142
}
3243

33-
func NewDAGService(bs *bserv.BlockService) DAGService {
34-
return &dagService{bs}
44+
func NewDAGService(bs *bserv.BlockService) *dagService {
45+
return &dagService{Blocks: bs}
3546
}
3647

3748
// dagService is an IPFS Merkle DAG service.
@@ -93,13 +104,30 @@ func (n *dagService) Get(ctx context.Context, c *cid.Cid) (*Node, error) {
93104
return res, nil
94105
}
95106

107+
func (n *dagService) GetLinks(ctx context.Context, c *cid.Cid) ([]*Link, error) {
108+
node, err := n.Get(ctx, c)
109+
if err != nil {
110+
return nil, err
111+
}
112+
return node.Links, nil
113+
}
114+
115+
func (n *dagService) GetOfflineLinkService() LinkService {
116+
if n.Blocks.Exchange.IsOnline() {
117+
bsrv := bserv.New(n.Blocks.Blockstore, offline.Exchange(n.Blocks.Blockstore))
118+
return NewDAGService(bsrv)
119+
} else {
120+
return n
121+
}
122+
}
123+
96124
func (n *dagService) Remove(nd *Node) error {
97125
return n.Blocks.DeleteObject(nd)
98126
}
99127

100128
// FetchGraph fetches all nodes that are children of the given node
101-
func FetchGraph(ctx context.Context, root *Node, serv DAGService) error {
102-
return EnumerateChildrenAsync(ctx, serv, root, cid.NewSet().Visit)
129+
func FetchGraph(ctx context.Context, c *cid.Cid, serv DAGService) error {
130+
return EnumerateChildrenAsync(ctx, serv, c, cid.NewSet().Visit)
103131
}
104132

105133
// FindLinks searches this nodes links for the given key,
@@ -366,19 +394,17 @@ func legacyCidFromLink(lnk *Link) *cid.Cid {
366394
// EnumerateChildren will walk the dag below the given root node and add all
367395
// unseen children to the passed in set.
368396
// TODO: parallelize to avoid disk latency perf hits?
369-
func EnumerateChildren(ctx context.Context, ds DAGService, root *Node, visit func(*cid.Cid) bool, bestEffort bool) error {
370-
for _, lnk := range root.Links {
397+
func EnumerateChildren(ctx context.Context, ds LinkService, root *cid.Cid, visit func(*cid.Cid) bool, bestEffort bool) error {
398+
links, err := ds.GetLinks(ctx, root)
399+
if bestEffort && err == ErrNotFound {
400+
return nil
401+
} else if err != nil {
402+
return err
403+
}
404+
for _, lnk := range links {
371405
c := legacyCidFromLink(lnk)
372406
if visit(c) {
373-
child, err := ds.Get(ctx, c)
374-
if err != nil {
375-
if bestEffort && err == ErrNotFound {
376-
continue
377-
} else {
378-
return err
379-
}
380-
}
381-
err = EnumerateChildren(ctx, ds, child, visit, bestEffort)
407+
err = EnumerateChildren(ctx, ds, c, visit, bestEffort)
382408
if err != nil {
383409
return err
384410
}
@@ -387,7 +413,7 @@ func EnumerateChildren(ctx context.Context, ds DAGService, root *Node, visit fun
387413
return nil
388414
}
389415

390-
func EnumerateChildrenAsync(ctx context.Context, ds DAGService, root *Node, visit func(*cid.Cid) bool) error {
416+
func EnumerateChildrenAsync(ctx context.Context, ds DAGService, c *cid.Cid, visit func(*cid.Cid) bool) error {
391417
toprocess := make(chan []*cid.Cid, 8)
392418
nodes := make(chan *NodeOption, 8)
393419

@@ -397,6 +423,11 @@ func EnumerateChildrenAsync(ctx context.Context, ds DAGService, root *Node, visi
397423

398424
go fetchNodes(ctx, ds, toprocess, nodes)
399425

426+
root, err := ds.Get(ctx, c)
427+
if err != nil {
428+
return err
429+
}
430+
400431
nodes <- &NodeOption{Node: root}
401432
live := 1
402433

‎merkledag/merkledag_test.go

+4-4
Original file line numberDiff line numberDiff line change
@@ -231,7 +231,7 @@ func TestFetchGraph(t *testing.T) {
231231
t.Fatal(err)
232232
}
233233

234-
err = FetchGraph(context.TODO(), root, dservs[1])
234+
err = FetchGraph(context.TODO(), root.Cid(), dservs[1])
235235
if err != nil {
236236
t.Fatal(err)
237237
}
@@ -241,7 +241,7 @@ func TestFetchGraph(t *testing.T) {
241241

242242
offline_ds := NewDAGService(bs)
243243

244-
err = EnumerateChildren(context.Background(), offline_ds, root, func(_ *cid.Cid) bool { return true }, false)
244+
err = EnumerateChildren(context.Background(), offline_ds, root.Cid(), func(_ *cid.Cid) bool { return true }, false)
245245
if err != nil {
246246
t.Fatal(err)
247247
}
@@ -258,7 +258,7 @@ func TestEnumerateChildren(t *testing.T) {
258258
}
259259

260260
set := cid.NewSet()
261-
err = EnumerateChildren(context.Background(), ds, root, set.Visit, false)
261+
err = EnumerateChildren(context.Background(), ds, root.Cid(), set.Visit, false)
262262
if err != nil {
263263
t.Fatal(err)
264264
}
@@ -269,7 +269,7 @@ func TestEnumerateChildren(t *testing.T) {
269269
for _, lnk := range n.Links {
270270
c := cid.NewCidV0(lnk.Hash)
271271
if !set.Has(c) {
272-
t.Fatal("missing key in set!")
272+
t.Fatal("missing key in set! ", lnk.Hash.B58String())
273273
}
274274
child, err := ds.Get(context.Background(), c)
275275
if err != nil {

‎pin/gc/gc.go

+9-16
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,6 @@ package gc
22

33
import (
44
bstore "github.com/ipfs/go-ipfs/blocks/blockstore"
5-
bserv "github.com/ipfs/go-ipfs/blockservice"
6-
offline "github.com/ipfs/go-ipfs/exchange/offline"
75
dag "github.com/ipfs/go-ipfs/merkledag"
86
pin "github.com/ipfs/go-ipfs/pin"
97
key "gx/ipfs/QmYEoKZXHoAToWfhGF3vryhMn3WWhE1o2MasQ8uzY5iDi9/go-key"
@@ -24,13 +22,12 @@ var log = logging.Logger("gc")
2422
//
2523
// The routine then iterates over every block in the blockstore and
2624
// deletes any block that is not found in the marked set.
27-
func GC(ctx context.Context, bs bstore.GCBlockstore, pn pin.Pinner, bestEffortRoots []*cid.Cid) (<-chan key.Key, error) {
25+
func GC(ctx context.Context, bs bstore.GCBlockstore, ls dag.LinkService, pn pin.Pinner, bestEffortRoots []*cid.Cid) (<-chan key.Key, error) {
2826
unlocker := bs.GCLock()
2927

30-
bsrv := bserv.New(bs, offline.Exchange(bs))
31-
ds := dag.NewDAGService(bsrv)
28+
ls = ls.GetOfflineLinkService()
3229

33-
gcs, err := ColoredSet(ctx, pn, ds, bestEffortRoots)
30+
gcs, err := ColoredSet(ctx, pn, ls, bestEffortRoots)
3431
if err != nil {
3532
return nil, err
3633
}
@@ -71,16 +68,12 @@ func GC(ctx context.Context, bs bstore.GCBlockstore, pn pin.Pinner, bestEffortRo
7168
return output, nil
7269
}
7370

74-
func Descendants(ctx context.Context, ds dag.DAGService, set key.KeySet, roots []*cid.Cid, bestEffort bool) error {
71+
func Descendants(ctx context.Context, ls dag.LinkService, set key.KeySet, roots []*cid.Cid, bestEffort bool) error {
7572
for _, c := range roots {
7673
set.Add(key.Key(c.Hash()))
77-
nd, err := ds.Get(ctx, c)
78-
if err != nil {
79-
return err
80-
}
8174

8275
// EnumerateChildren recursively walks the dag and adds the keys to the given set
83-
err = dag.EnumerateChildren(ctx, ds, nd, func(c *cid.Cid) bool {
76+
err := dag.EnumerateChildren(ctx, ls, c, func(c *cid.Cid) bool {
8477
k := key.Key(c.Hash())
8578
seen := set.Has(k)
8679
if seen {
@@ -97,16 +90,16 @@ func Descendants(ctx context.Context, ds dag.DAGService, set key.KeySet, roots [
9790
return nil
9891
}
9992

100-
func ColoredSet(ctx context.Context, pn pin.Pinner, ds dag.DAGService, bestEffortRoots []*cid.Cid) (key.KeySet, error) {
93+
func ColoredSet(ctx context.Context, pn pin.Pinner, ls dag.LinkService, bestEffortRoots []*cid.Cid) (key.KeySet, error) {
10194
// KeySet currently implemented in memory, in the future, may be bloom filter or
10295
// disk backed to conserve memory.
10396
gcs := key.NewKeySet()
104-
err := Descendants(ctx, ds, gcs, pn.RecursiveKeys(), false)
97+
err := Descendants(ctx, ls, gcs, pn.RecursiveKeys(), false)
10598
if err != nil {
10699
return nil, err
107100
}
108101

109-
err = Descendants(ctx, ds, gcs, bestEffortRoots, true)
102+
err = Descendants(ctx, ls, gcs, bestEffortRoots, true)
110103
if err != nil {
111104
return nil, err
112105
}
@@ -115,7 +108,7 @@ func ColoredSet(ctx context.Context, pn pin.Pinner, ds dag.DAGService, bestEffor
115108
gcs.Add(key.Key(k.Hash()))
116109
}
117110

118-
err = Descendants(ctx, ds, gcs, pn.InternalPins(), false)
111+
err = Descendants(ctx, ls, gcs, pn.InternalPins(), false)
119112
if err != nil {
120113
return nil, err
121114
}

‎pin/pin.go

+11-17
Original file line numberDiff line numberDiff line change
@@ -178,7 +178,7 @@ func (p *pinner) Pin(ctx context.Context, node *mdag.Node, recurse bool) error {
178178
}
179179

180180
// fetch entire graph
181-
err := mdag.FetchGraph(ctx, node, p.dserv)
181+
err := mdag.FetchGraph(ctx, c, p.dserv)
182182
if err != nil {
183183
return err
184184
}
@@ -279,12 +279,7 @@ func (p *pinner) isPinnedWithType(c *cid.Cid, mode PinMode) (string, bool, error
279279

280280
// Default is Indirect
281281
for _, rc := range p.recursePin.Keys() {
282-
rnd, err := p.dserv.Get(context.Background(), rc)
283-
if err != nil {
284-
return "", false, err
285-
}
286-
287-
has, err := hasChild(p.dserv, rnd, k)
282+
has, err := hasChild(p.dserv, rc, k)
288283
if err != nil {
289284
return "", false, err
290285
}
@@ -317,11 +312,11 @@ func (p *pinner) CheckIfPinned(cids ...*cid.Cid) ([]Pinned, error) {
317312
// Now walk all recursive pins to check for indirect pins
318313
var checkChildren func(*cid.Cid, *cid.Cid) error
319314
checkChildren = func(rk, parentKey *cid.Cid) error {
320-
parent, err := p.dserv.Get(context.Background(), parentKey)
315+
links, err := p.dserv.GetLinks(context.Background(), parentKey)
321316
if err != nil {
322317
return err
323318
}
324-
for _, lnk := range parent.Links {
319+
for _, lnk := range links {
325320
c := cid.NewCidV0(lnk.Hash)
326321

327322
if toCheck.Has(c) {
@@ -521,19 +516,18 @@ func (p *pinner) PinWithMode(c *cid.Cid, mode PinMode) {
521516
}
522517
}
523518

524-
func hasChild(ds mdag.DAGService, root *mdag.Node, child key.Key) (bool, error) {
525-
for _, lnk := range root.Links {
519+
func hasChild(ds mdag.LinkService, root *cid.Cid, child key.Key) (bool, error) {
520+
links, err := ds.GetLinks(context.Background(), root)
521+
if err != nil {
522+
return false, err
523+
}
524+
for _, lnk := range links {
526525
c := cid.NewCidV0(lnk.Hash)
527526
if key.Key(c.Hash()) == child {
528527
return true, nil
529528
}
530529

531-
nd, err := ds.Get(context.Background(), c)
532-
if err != nil {
533-
return false, err
534-
}
535-
536-
has, err := hasChild(ds, nd, child)
530+
has, err := hasChild(ds, c, child)
537531
if err != nil {
538532
return false, err
539533
}

‎pin/pin_test.go

+5
Original file line numberDiff line numberDiff line change
@@ -225,6 +225,11 @@ func TestPinRecursiveFail(t *testing.T) {
225225
t.Fatal(err)
226226
}
227227

228+
_, err = dserv.Add(a)
229+
if err != nil {
230+
t.Fatal(err)
231+
}
232+
228233
// this one is time based... but shouldnt cause any issues
229234
mctx, _ = context.WithTimeout(ctx, time.Second)
230235
err = p.Pin(mctx, a, true)

0 commit comments

Comments
 (0)