@cryptotaxi247 / kubo / commits / 721df367a

Don't use a separate LinkService for DAGService.GetLinks()

Instead make LinkService a part of DAGService. The LinkService is now simply an interface that DAGService implements. Also provide a GetOfflineLinkService() method that the GC uses to get an offline instance. License: MIT Signed-off-by: Kevin Atkinson <k@kevina.org>

Kevin Atkinson committed Sep 30, 2016 at 17:06 UTC 721df367a27dfa416e1d93df23ca440fc2431f63
9 files changed +50 -42
core/core.go
+8 -9
@@ -94,15 +94,14 @@ type IpfsNode struct {
94 PrivateKey ic.PrivKey // the local node's private Key
95
96 // Services
97 - Peerstore pstore.Peerstore // storage for other Peer instances
98 - Blockstore bstore.GCBlockstore // the block store (lower level)
99 - Blocks *bserv.BlockService // the block service, get/add blocks.
100 - DAG merkledag.DAGService // the merkle dag service, get/add objects.
101 - LinkService merkledag.LinkService
102 - Resolver *path.Resolver // the path resolution system
103 - Reporter metrics.Reporter
104 - Discovery discovery.Service
105 - FilesRoot *mfs.Root
97 + Peerstore pstore.Peerstore // storage for other Peer instances
98 + Blockstore bstore.GCBlockstore // the block store (lower level)
99 + Blocks *bserv.BlockService // the block service, get/add blocks.
100 + DAG merkledag.DAGService // the merkle dag service, get/add objects.
101 + Resolver *path.Resolver // the path resolution system
102 + Reporter metrics.Reporter
103 + Discovery discovery.Service
104 + FilesRoot *mfs.Root
105
106 // Online
107 PeerHost p2phost.Host // the network host (server+client)
core/corerepo/gc.go
+2 -2
@@ -90,7 +90,7 @@ func GarbageCollect(n *core.IpfsNode, ctx context.Context) error {
90 if err != nil {
91 return err
92 }
93 - rmed, err := gc.GC(ctx, n.Blockstore, n.LinkService, n.Pinning, roots)
93 + rmed, err := gc.GC(ctx, n.Blockstore, n.DAG, n.Pinning, roots)
94 if err != nil {
95 return err
96 }
@@ -113,7 +113,7 @@ func GarbageCollectAsync(n *core.IpfsNode, ctx context.Context) (<-chan *KeyRemo
113 if err != nil {
114 return nil, err
115 }
116 - rmed, err := gc.GC(ctx, n.Blockstore, n.LinkService, n.Pinning, roots)
116 + rmed, err := gc.GC(ctx, n.Blockstore, n.DAG, n.Pinning, roots)
117 if err != nil {
118 return nil, err
119 }
core/coreunix/add_test.go
+1 -1
@@ -98,7 +98,7 @@ func TestAddGCLive(t *testing.T) {
98 gcstarted := make(chan struct{})
99 go func() {
100 defer close(gcstarted)
101 - gcchan, err := gc.GC(context.Background(), node.Blockstore, node.LinkService, node.Pinning, nil)
101 + gcchan, err := gc.GC(context.Background(), node.Blockstore, node.DAG, node.Pinning, nil)
102 if err != nil {
103 log.Error("GC ERROR:", err)
104 errs <- err
exchange/bitswap/bitswap.go
+4
@@ -422,3 +422,7 @@ func (bs *Bitswap) GetWantlist() []key.Key {
422 }
423 return out
424 }
425 +
426 +func (bs *Bitswap) IsOnline() bool {
427 + return true
428 +}
exchange/interface.go
+2
@@ -22,5 +22,7 @@ type Interface interface { // type Exchanger interface
22 // available on the network?
23 HasBlock(blocks.Block) error
24
25 + IsOnline() bool
26 +
27 io.Closer
28 }
exchange/offline/offline.go
+4
@@ -67,3 +67,7 @@ func (e *offlineExchange) GetBlocks(ctx context.Context, ks []key.Key) (<-chan b
67 }()
68 return out, nil
69 }
70 +
71 +func (e *offlineExchange) IsOnline() bool {
72 + return false
73 +}
merkledag/merkledag.go
+19 -16
@@ -7,6 +7,7 @@ import (
7 "sync"
8
9 bserv "github.com/ipfs/go-ipfs/blockservice"
10 + offline "github.com/ipfs/go-ipfs/exchange/offline"
11 key "gx/ipfs/QmYEoKZXHoAToWfhGF3vryhMn3WWhE1o2MasQ8uzY5iDi9/go-key"
12
13 "context"
@@ -23,21 +24,21 @@ type DAGService interface {
24 Get(context.Context, *cid.Cid) (*Node, error)
25 Remove(*Node) error
26
26 - // Return all links for a node, may be more effect than
27 - // calling Get
28 - GetLinks(context.Context, *cid.Cid) ([]*Link, error)
29 -
27 // GetDAG returns, in order, all the single leve child
28 // nodes of the passed in node.
29 GetMany(context.Context, []*cid.Cid) <-chan *NodeOption
30
31 Batch() *Batch
32 +
33 + LinkService
34 }
35
37 -// A LinkService returns the links for a node if they are available
38 -// locally without having to retrieve the block from the datastore.
36 type LinkService interface {
40 - Get(*cid.Cid) ([]*Link, error)
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
42 }
43
44 func NewDAGService(bs *bserv.BlockService) *dagService {
@@ -50,8 +51,7 @@ func NewDAGService(bs *bserv.BlockService) *dagService {
51 // TODO: should cache Nodes that are in memory, and be
52 // able to free some of them when vm pressure is high
53 type dagService struct {
53 - Blocks *bserv.BlockService
54 - LinkService LinkService
54 + Blocks *bserv.BlockService
55 }
56
57 // Add adds a node to the dagService, storing the block in the BlockService
@@ -105,12 +105,6 @@ func (n *dagService) Get(ctx context.Context, c *cid.Cid) (*Node, error) {
105 }
106
107 func (n *dagService) GetLinks(ctx context.Context, c *cid.Cid) ([]*Link, error) {
108 - if n.LinkService != nil {
109 - links, err := n.LinkService.Get(c)
110 - if err == nil {
111 - return links, nil
112 - }
113 - }
108 node, err := n.Get(ctx, c)
109 if err != nil {
110 return nil, err
@@ -118,6 +112,15 @@ func (n *dagService) GetLinks(ctx context.Context, c *cid.Cid) ([]*Link, error)
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 +
124 func (n *dagService) Remove(nd *Node) error {
125 return n.Blocks.DeleteObject(nd)
126 }
@@ -391,7 +394,7 @@ func legacyCidFromLink(lnk *Link) *cid.Cid {
394 // EnumerateChildren will walk the dag below the given root node and add all
395 // unseen children to the passed in set.
396 // TODO: parallelize to avoid disk latency perf hits?
394 -func EnumerateChildren(ctx context.Context, ds DAGService, links []*Link, visit func(*cid.Cid) bool, bestEffort bool) error {
397 +func EnumerateChildren(ctx context.Context, ds LinkService, links []*Link, visit func(*cid.Cid) bool, bestEffort bool) error {
398 for _, lnk := range links {
399 c := legacyCidFromLink(lnk)
400 if visit(c) {
pin/gc/gc.go
+9 -13
@@ -2,8 +2,6 @@ package gc
2
3 import (
4 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"
5 dag "github.com/ipfs/go-ipfs/merkledag"
6 pin "github.com/ipfs/go-ipfs/pin"
7 key "gx/ipfs/QmYEoKZXHoAToWfhGF3vryhMn3WWhE1o2MasQ8uzY5iDi9/go-key"
@@ -27,11 +25,9 @@ var log = logging.Logger("gc")
25 func GC(ctx context.Context, bs bstore.GCBlockstore, ls dag.LinkService, pn pin.Pinner, bestEffortRoots []*cid.Cid) (<-chan key.Key, error) {
26 unlocker := bs.GCLock()
27
30 - bsrv := bserv.New(bs, offline.Exchange(bs))
31 - ds := dag.NewDAGService(bsrv)
32 - ds.LinkService = ls
28 + ls = ls.GetOfflineLinkService()
29
34 - gcs, err := ColoredSet(ctx, pn, ds, bestEffortRoots)
30 + gcs, err := ColoredSet(ctx, pn, ls, bestEffortRoots)
31 if err != nil {
32 return nil, err
33 }
@@ -72,16 +68,16 @@ func GC(ctx context.Context, bs bstore.GCBlockstore, ls dag.LinkService, pn pin.
68 return output, nil
69 }
70
75 -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 {
72 for _, c := range roots {
73 set.Add(key.Key(c.Hash()))
78 - links, err := ds.GetLinks(ctx, c)
74 + links, err := ls.GetLinks(ctx, c)
75 if err != nil {
76 return err
77 }
78
79 // EnumerateChildren recursively walks the dag and adds the keys to the given set
84 - err = dag.EnumerateChildren(ctx, ds, links, func(c *cid.Cid) bool {
80 + err = dag.EnumerateChildren(ctx, ls, links, func(c *cid.Cid) bool {
81 k := key.Key(c.Hash())
82 seen := set.Has(k)
83 if seen {
@@ -98,16 +94,16 @@ func Descendants(ctx context.Context, ds dag.DAGService, set key.KeySet, roots [
94 return nil
95 }
96
101 -func ColoredSet(ctx context.Context, pn pin.Pinner, ds dag.DAGService, bestEffortRoots []*cid.Cid) (key.KeySet, error) {
97 +func ColoredSet(ctx context.Context, pn pin.Pinner, ls dag.LinkService, bestEffortRoots []*cid.Cid) (key.KeySet, error) {
98 // KeySet currently implemented in memory, in the future, may be bloom filter or
99 // disk backed to conserve memory.
100 gcs := key.NewKeySet()
105 - err := Descendants(ctx, ds, gcs, pn.RecursiveKeys(), false)
101 + err := Descendants(ctx, ls, gcs, pn.RecursiveKeys(), false)
102 if err != nil {
103 return nil, err
104 }
105
110 - err = Descendants(ctx, ds, gcs, bestEffortRoots, true)
106 + err = Descendants(ctx, ls, gcs, bestEffortRoots, true)
107 if err != nil {
108 return nil, err
109 }
@@ -116,7 +112,7 @@ func ColoredSet(ctx context.Context, pn pin.Pinner, ds dag.DAGService, bestEffor
112 gcs.Add(key.Key(k.Hash()))
113 }
114
119 - err = Descendants(ctx, ds, gcs, pn.InternalPins(), false)
115 + err = Descendants(ctx, ls, gcs, pn.InternalPins(), false)
116 if err != nil {
117 return nil, err
118 }
pin/pin.go
+1 -1
@@ -521,7 +521,7 @@ func (p *pinner) PinWithMode(c *cid.Cid, mode PinMode) {
521 }
522 }
523
524 -func hasChild(ds mdag.DAGService, links []*mdag.Link, child key.Key) (bool, error) {
524 +func hasChild(ds mdag.LinkService, links []*mdag.Link, child key.Key) (bool, error) {
525 for _, lnk := range links {
526 c := cid.NewCidV0(lnk.Hash)
527 if key.Key(c.Hash()) == child {