Implement caching in GetLinks(), make EnumerateChildrenAsync use GetL…

…inks GetLinks() now is using link cache in datastore under /local/links and will not necessarily fetch linked nodes. This also affects EnumerateChildrenAsync and EnumerateChildren, and their previous behaviour can be reproduced by using FetchingVisitor from merkledag module. Add `ipfs repo flushlinkcache` License: MIT Signed-off-by: Iaroslav Gridin <[email protected]>
ipfs · Jul 14, 2017 · a047c47 · a047c47
1 parent 06c567d
commit a047c47
Show file tree

Hide file tree

Showing 19 changed files with 303 additions and 71 deletions.
diff --git a/blockservice/blockservice.go b/blockservice/blockservice.go
@@ -30,6 +30,7 @@ type BlockService interface {
 	AddBlock(o blocks.Block) (*cid.Cid, error)
 	AddBlocks(bs []blocks.Block) ([]*cid.Cid, error)
 	GetBlock(ctx context.Context, c *cid.Cid) (blocks.Block, error)
+	FetchBlock(ctx context.Context, c *cid.Cid) error
 	GetBlocks(ctx context.Context, ks []*cid.Cid) <-chan blocks.Block
 	DeleteBlock(o blocks.Block) error
 	Close() error
@@ -195,6 +196,34 @@ func getBlock(ctx context.Context, c *cid.Cid, bs blockstore.Blockstore, f excha
 	return nil, err
 }
 
+// FetchBlock ensures requested block is present in local storage
+func (bs *blockService) FetchBlock(ctx context.Context, c *cid.Cid) error {
+	log.Debugf("BlockService FetchBlock: '%s'", c)
+
+	present, err := bs.blockstore.Has(c)
+	if err != nil {
+		return err
+	}
+	if present == true {
+		return nil
+	}
+
+	if bs.exchange != nil {
+		log.Debug("Blockservice: Searching bitswap")
+		_, err := bs.exchange.GetBlock(ctx, c)
+		if err != nil {
+			if err == blockstore.ErrNotFound {
+				return ErrNotFound
+			}
+			return err
+		}
+		return nil
+	}
+
+	log.Debug("Blockservice FetchBlock: Not found")
+	return ErrNotFound
+}
+
 // GetBlocks gets a list of blocks asynchronously and returns through
 // the returned channel.
 // NB: No guarantees are made about order.

diff --git a/core/builder.go b/core/builder.go
@@ -218,9 +218,9 @@ func setupNode(ctx context.Context, n *IpfsNode, cfg *BuildCfg) error {
 	}
 
 	n.Blocks = bserv.New(n.Blockstore, n.Exchange)
-	n.DAG = dag.NewDAGService(n.Blocks)
+	n.DAG = dag.NewDAGService(n.Blocks, n.Repo.Datastore())
 
-	internalDag := dag.NewDAGService(bserv.New(n.Blockstore, offline.Exchange(n.Blockstore)))
+	internalDag := dag.NewDAGService(bserv.New(n.Blockstore, offline.Exchange(n.Blockstore)), n.Repo.Datastore())
 	n.Pinning, err = pin.LoadPinner(n.Repo.Datastore(), n.DAG, internalDag)
 	if err != nil {
 		// TODO: we should move towards only running 'NewPinner' explicity on

diff --git a/core/commands/add.go b/core/commands/add.go
@@ -227,13 +227,14 @@ You can now refer to the added file in a gateway, like so:
 		}
 
 		exch := n.Exchange
+
 		local, _, _ := req.Option("local").Bool()
 		if local {
 			exch = offline.Exchange(addblockstore)
 		}
 
 		bserv := blockservice.New(addblockstore, exch)
-		dserv := dag.NewDAGService(bserv)
+		dserv := dag.NewDAGService(bserv, n.Repo.Datastore())
 
 		fileAdder, err := coreunix.NewAdder(req.Context(), n.Pinning, n.Blockstore, dserv)
 		if err != nil {

diff --git a/core/commands/dht.go b/core/commands/dht.go
@@ -378,8 +378,7 @@ func provideKeysRec(ctx context.Context, r routing.IpfsRouting, dserv dag.DAGSer
 	provided := cid.NewSet()
 	for _, c := range cids {
 		kset := cid.NewSet()
-
-		err := dag.EnumerateChildrenAsync(ctx, dag.GetLinksDirect(dserv), c, kset.Visit)
+		err := dag.EnumerateChildrenAsync(ctx, dag.GetLinksDirect(dserv), c, dag.FetchingVisitor(ctx, kset, dserv, cid.NewSet()))
 		if err != nil {
 			return err
 		}

diff --git a/core/commands/ls.go b/core/commands/ls.go
@@ -77,7 +77,7 @@ The JSON output contains type information.
 		if !resolve {
 			offlineexch := offline.Exchange(nd.Blockstore)
 			bserv := blockservice.New(nd.Blockstore, offlineexch)
-			dserv = merkledag.NewDAGService(bserv)
+			dserv = merkledag.NewDAGService(bserv, nd.Repo.Datastore())
 		}
 
 		paths := req.Arguments()

diff --git a/core/commands/repo.go b/core/commands/repo.go
@@ -33,11 +33,12 @@ var RepoCmd = &cmds.Command{
 	},
 
 	Subcommands: map[string]*cmds.Command{
-		"gc":      repoGcCmd,
-		"stat":    repoStatCmd,
-		"fsck":    RepoFsckCmd,
-		"version": repoVersionCmd,
-		"verify":  repoVerifyCmd,
+		"gc":             repoGcCmd,
+		"stat":           repoStatCmd,
+		"fsck":           RepoFsckCmd,
+		"version":        repoVersionCmd,
+		"verify":         repoVerifyCmd,
+		"flushlinkcache": repoFlushLinkCacheCmd,
 	},
 }
 
@@ -206,6 +207,33 @@ Version         string The repo version.
 	},
 }
 
+var repoFlushLinkCacheCmd = &cmds.Command{
+	Helptext: cmds.HelpText{
+		Tagline: "Flush link cache.",
+		ShortDescription: `
+'ipfs repo flushlinkcache' is a plumbing command that will flush link cache.
+`,
+	},
+	Run: func(req cmds.Request, res cmds.Response) {
+		n, err := req.InvocContext().GetNode()
+		if err != nil {
+			res.SetError(err, cmds.ErrNormal)
+			return
+		}
+
+		err = corerepo.FlushLinkCache(req.Context(), n)
+		if err != nil {
+			res.SetError(err, cmds.ErrNormal)
+			return
+		}
+		res.SetOutput(&MessageOutput{"Link cache flushed.\n"})
+	},
+	Type: MessageOutput{},
+	Marshalers: cmds.MarshalerMap{
+		cmds.Text: MessageTextMarshaler,
+	},
+}
+
 var RepoFsckCmd = &cmds.Command{
 	Helptext: cmds.HelpText{
 		Tagline: "Remove repo lockfiles.",

diff --git a/core/corerepo/linkcache.go b/core/corerepo/linkcache.go
@@ -0,0 +1,25 @@
+package corerepo
+
+import (
+	context "context"
+	core "github.com/ipfs/go-ipfs/core"
+	ds "gx/ipfs/QmVSase1JP7cq9QkPT46oNwdp9pT6kBkG3oqS14y3QcZjG/go-datastore"
+	dsq "gx/ipfs/QmVSase1JP7cq9QkPT46oNwdp9pT6kBkG3oqS14y3QcZjG/go-datastore/query"
+)
+
+// FlushLinkCache flushes link cache, deleting all keys in it
+func FlushLinkCache(ctx context.Context, n *core.IpfsNode) error {
+	d := n.Repo.Datastore()
+	q := dsq.Query{KeysOnly: true, Prefix: "/local/links/"}
+	qr, err := d.Query(q)
+	if err != nil {
+		return err
+	}
+	for result := range qr.Next() {
+		if result.Error != nil {
+			return result.Error
+		}
+		d.Delete(ds.NewKey(result.Entry.Key))
+	}
+	return nil
+}
diff --git a/core/coreunix/add.go b/core/coreunix/add.go
@@ -552,7 +552,7 @@ func NewMemoryDagService() dag.DAGService {
 	// build mem-datastore for editor's intermediary nodes
 	bs := bstore.NewBlockstore(syncds.MutexWrap(ds.NewMapDatastore()))
 	bsrv := bserv.New(bs, offline.Exchange(bs))
-	return dag.NewDAGService(bsrv)
+	return dag.NewDAGService(bsrv, syncds.MutexWrap(ds.NewMapDatastore()))
 }
 
 // from core/commands/object.go

diff --git a/core/coreunix/add_test.go b/core/coreunix/add_test.go
@@ -23,6 +23,8 @@ import (
 	"gx/ipfs/QmVA4mafxbfH5aEvNz8fyoxC6J1xhAtw88B4GerPznSZBg/go-block-format"
 
 	cid "gx/ipfs/QmTprEaAA2A9bst5XH7exuyi5KzNMK3SEDNN8rBDnKWcUS/go-cid"
+	ds "gx/ipfs/QmVSase1JP7cq9QkPT46oNwdp9pT6kBkG3oqS14y3QcZjG/go-datastore"
+	syncds "gx/ipfs/QmVSase1JP7cq9QkPT46oNwdp9pT6kBkG3oqS14y3QcZjG/go-datastore/sync"
 )
 
 func TestAddRecursive(t *testing.T) {
@@ -148,8 +150,9 @@ func TestAddGCLive(t *testing.T) {
 	defer cancel()
 
 	set := cid.NewSet()
-	err = dag.EnumerateChildren(ctx, node.DAG.GetLinks, last, set.Visit)
-	if err != nil {
+	missing := cid.NewSet()
+	err = dag.EnumerateChildren(ctx, node.DAG.GetLinks, last, dag.FetchingVisitor(ctx, set, node.DAG, missing))
+	if err != nil || missing.Len() > 0 {
 		t.Fatal(err)
 	}
 }
@@ -170,7 +173,7 @@ func testAddWPosInfo(t *testing.T, rawLeaves bool) {
 
 	bs := &testBlockstore{GCBlockstore: node.Blockstore, expectedPath: "/tmp/foo.txt", t: t}
 	bserv := blockservice.New(bs, node.Exchange)
-	dserv := dag.NewDAGService(bserv)
+	dserv := dag.NewDAGService(bserv, syncds.MutexWrap(ds.NewMapDatastore()))
 	adder, err := NewAdder(context.Background(), node.Pinning, bs, dserv)
 	if err != nil {
 		t.Fatal(err)

diff --git a/core/coreunix/metadata_test.go b/core/coreunix/metadata_test.go
@@ -26,7 +26,7 @@ func getDagserv(t *testing.T) merkledag.DAGService {
 	db := dssync.MutexWrap(ds.NewMapDatastore())
 	bs := bstore.NewBlockstore(db)
 	blockserv := bserv.New(bs, offline.Exchange(bs))
-	return merkledag.NewDAGService(blockserv)
+	return merkledag.NewDAGService(blockserv, db)
 }
 
 func TestMetadata(t *testing.T) {

diff --git a/merkledag/coding.go b/merkledag/coding.go
@@ -16,6 +16,28 @@ import (
 // for now, we use a PBNode intermediate thing.
 // because native go objects are nice.
 
+func intermediateMarshal(l *node.Link) *pb.PBLink {
+	pbl := &pb.PBLink{}
+	pbl.Name = &l.Name
+	pbl.Tsize = &l.Size
+	if l.Cid != nil {
+		pbl.Hash = l.Cid.Bytes()
+	}
+	return pbl
+}
+
+func intermediateUnmarshal(pbl *pb.PBLink) (*node.Link, error) {
+	l := new(node.Link)
+	l.Name = pbl.GetName()
+	l.Size = pbl.GetTsize()
+	c, err := cid.Cast(pbl.GetHash())
+	if err != nil {
+		return nil, fmt.Errorf("Link hash is not valid multihash. %v", err)
+	}
+	l.Cid = c
+	return l, nil
+}
+
 // unmarshal decodes raw data into a *Node instance.
 // The conversion uses an intermediate PBNode.
 func (n *ProtoNode) unmarshal(encoded []byte) error {