Merge pull request #631 from filecoin-project/feat/fast-vm-sync

use fast cbor-gen method for extracting links from objects

chain/vm/vm.go

@@ -1,18 +1,16 @@
 package vm
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"math/big"
 
 	block "github.com/ipfs/go-block-format"
-	bserv "github.com/ipfs/go-blockservice"
 	cid "github.com/ipfs/go-cid"
 	hamt "github.com/ipfs/go-hamt-ipld"
 	blockstore "github.com/ipfs/go-ipfs-blockstore"
-	ipld "github.com/ipfs/go-ipld-format"
 	logging "github.com/ipfs/go-log"
-	dag "github.com/ipfs/go-merkledag"
 	cbg "github.com/whyrusleeping/cbor-gen"
 	"go.opencensus.io/trace"
 	"golang.org/x/xerrors"
@@ -526,52 +524,93 @@ func (vm *VM) Flush(ctx context.Context) (cid.Cid, error) {
 	ctx, span := trace.StartSpan(ctx, "vm.Flush")
 	defer span.End()
 
-	from := dag.NewDAGService(bserv.New(vm.buf, nil))
-	to := dag.NewDAGService(bserv.New(vm.buf.Read(), nil))
+	from := vm.buf
+	to := vm.buf.Read()
 
 	root, err := vm.cstate.Flush()
 	if err != nil {
 		return cid.Undef, xerrors.Errorf("flushing vm: %w", err)
 	}
 
-	if err := Copy(ctx, from, to, root); err != nil {
+	if err := Copy(from, to, root); err != nil {
 		return cid.Undef, xerrors.Errorf("copying tree: %w", err)
 	}
 
 	return root, nil
 }
 
-func Copy(ctx context.Context, from, to ipld.DAGService, root cid.Cid) error {
+func linksForObj(blk block.Block) ([]cid.Cid, error) {
+	switch blk.Cid().Prefix().Codec {
+	case cid.DagCBOR:
+		return cbg.ScanForLinks(bytes.NewReader(blk.RawData()))
+	default:
+		return nil, xerrors.Errorf("vm flush copy method only supports dag cbor")
+	}
+}
+
+func Copy(from, to blockstore.Blockstore, root cid.Cid) error {
+	var batch []block.Block
+	batchCp := func(blk block.Block) error {
+		batch = append(batch, blk)
+		if len(batch) > 100 {
+			if err := to.PutMany(batch); err != nil {
+				return xerrors.Errorf("batch put in copy: %w", err)
+			}
+			batch = batch[:0]
+		}
+		return nil
+	}
+
+	if err := copyRec(from, to, root, batchCp); err != nil {
+		return err
+	}
+
+	if len(batch) > 0 {
+		if err := to.PutMany(batch); err != nil {
+			return xerrors.Errorf("batch put in copy: %w", err)
+		}
+	}
+
+	return nil
+}
+
+func copyRec(from, to blockstore.Blockstore, root cid.Cid, cp func(block.Block) error) error {
 	if root.Prefix().MhType == 0 {
 		// identity cid, skip
 		return nil
 	}
-	node, err := from.Get(ctx, root)
+
+	blk, err := from.Get(root)
 	if err != nil {
 		return xerrors.Errorf("get %s failed: %w", root, err)
 	}
-	links := node.Links()
-	for _, link := range links {
-		if link.Cid.Prefix().MhType == 0 {
-			continue
-		}
-		_, err := to.Get(ctx, link.Cid)
-		switch err {
-		default:
-			return err
-		case nil:
-			continue
-		case ipld.ErrNotFound:
-			// continue
-		}
-		if err := Copy(ctx, from, to, link.Cid); err != nil {
-			return err
-		}
-	}
-	err = to.Add(ctx, node)
+
+	links, err := linksForObj(blk)
 	if err != nil {
 		return err
 	}
+
+	for _, link := range links {
+		if link.Prefix().MhType == 0 {
+			continue
+		}
+
+		has, err := to.Has(link)
+		if err != nil {
+			return err
+		}
+		if has {
+			continue
+		}
+
+		if err := copyRec(from, to, link, cp); err != nil {
+			return err
+		}
+	}
+
+	if err := cp(blk); err != nil {
+		return err
+	}
 	return nil
 }
 

go.mod

@@ -90,7 +90,7 @@ require (
 	github.com/smartystreets/goconvey v0.0.0-20190731233626-505e41936337 // indirect
 	github.com/stretchr/testify v1.4.0
 	github.com/whyrusleeping/bencher v0.0.0-20190829221104-bb6607aa8bba
-	github.com/whyrusleeping/cbor-gen v0.0.0-20191107223350-6fdade89d679
+	github.com/whyrusleeping/cbor-gen v0.0.0-20191116002219-891f55cd449d
 	github.com/whyrusleeping/multiaddr-filter v0.0.0-20160516205228-e903e4adabd7
 	github.com/whyrusleeping/pubsub v0.0.0-20131020042734-02de8aa2db3d
 	go.opencensus.io v0.22.0

go.sum

@@ -550,8 +550,8 @@ github.com/whyrusleeping/bencher v0.0.0-20190829221104-bb6607aa8bba h1:X4n8JG2e2
 github.com/whyrusleeping/bencher v0.0.0-20190829221104-bb6607aa8bba/go.mod h1:CHQnYnQUEPydYCwuy8lmTHfGmdw9TKrhWV0xLx8l0oM=
 github.com/whyrusleeping/cbor-gen v0.0.0-20190910031516-c1cbffdb01bb/go.mod h1:xdlJQaiqipF0HW+Mzpg7XRM3fWbGvfgFlcppuvlkIvY=
 github.com/whyrusleeping/cbor-gen v0.0.0-20190917003517-d78d67427694/go.mod h1:xdlJQaiqipF0HW+Mzpg7XRM3fWbGvfgFlcppuvlkIvY=
-github.com/whyrusleeping/cbor-gen v0.0.0-20191107223350-6fdade89d679 h1:ct50KYdZHcdOnTAuSgppw5MZKTa3RA63FX28m0l9Aeg=
-github.com/whyrusleeping/cbor-gen v0.0.0-20191107223350-6fdade89d679/go.mod h1:xdlJQaiqipF0HW+Mzpg7XRM3fWbGvfgFlcppuvlkIvY=
+github.com/whyrusleeping/cbor-gen v0.0.0-20191116002219-891f55cd449d h1:NRa/Vs7+b91GdXrp0AqsG7pspWV6CLk5Gk7i46L4tGo=
+github.com/whyrusleeping/cbor-gen v0.0.0-20191116002219-891f55cd449d/go.mod h1:xdlJQaiqipF0HW+Mzpg7XRM3fWbGvfgFlcppuvlkIvY=
 github.com/whyrusleeping/chunker v0.0.0-20181014151217-fe64bd25879f h1:jQa4QT2UP9WYv2nzyawpKMOCl+Z/jW7djv2/J50lj9E=
 github.com/whyrusleeping/chunker v0.0.0-20181014151217-fe64bd25879f/go.mod h1:p9UJB6dDgdPgMJZs7UjUOdulKyRr9fqkS+6JKAInPy8=
 github.com/whyrusleeping/go-keyspace v0.0.0-20160322163242-5b898ac5add1 h1:EKhdznlJHPMoKr0XTrX+IlJs1LH3lyx2nfr1dOlZ79k=