move to native badger blockstore; leverage zero-copy View() to deserialize in-place

chain/gen/gen.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"sync/atomic"
 	"time"
@@ -138,12 +139,20 @@ func NewGeneratorWithSectors(numSectors int) (*ChainGen, error) {
 		return nil, xerrors.Errorf("failed to get metadata datastore: %w", err)
 	}
 
-	bds, err := lr.Datastore("/chain")
+	bs, err := lr.Blockstore(repo.BlockstoreChain)
 	if err != nil {
-		return nil, xerrors.Errorf("failed to get blocks datastore: %w", err)
+		return nil, err
 	}
 
-	bs := mybs{blockstore.NewBlockstore(bds)}
+	defer func() {
+		if c, ok := bs.(io.Closer); ok {
+			if err := c.Close(); err != nil {
+				log.Warnf("failed to close blockstore: %s", err)
+			}
+		}
+	}()
+
+	bs = mybs{bs}
 
 	ks, err := lr.KeyStore()
 	if err != nil {

chain/store/store.go

@@ -44,10 +44,10 @@ import (
 	"github.com/ipfs/go-datastore/query"
 	cbor "github.com/ipfs/go-ipld-cbor"
 	logging "github.com/ipfs/go-log/v2"
-	car "github.com/ipld/go-car"
+	"github.com/ipld/go-car"
 	carutil "github.com/ipld/go-car/util"
 	cbg "github.com/whyrusleeping/cbor-gen"
-	pubsub "github.com/whyrusleeping/pubsub"
+	"github.com/whyrusleeping/pubsub"
 	"golang.org/x/xerrors"
 )
 
@@ -108,6 +108,8 @@ type ChainStore struct {
 	localbs bstore.Blockstore
 	ds      dstore.Batching
 
+	localviewer bstore.Viewer
+
 	heaviestLk sync.Mutex
 	heaviest   *types.TipSet
 
@@ -150,6 +152,10 @@ func NewChainStore(bs bstore.Blockstore, localbs bstore.Blockstore, ds dstore.Ba
 		journal:  j,
 	}
 
+	if v, ok := localbs.(bstore.Viewer); ok {
+		cs.localviewer = v
+	}
+
 	cs.evtTypes = [1]journal.EventType{
 		evtTypeHeadChange: j.RegisterEventType("sync", "head_change"),
 	}
@@ -365,6 +371,26 @@ func (cs *ChainStore) MaybeTakeHeavierTipSet(ctx context.Context, ts *types.TipS
 	return nil
 }
 
+// ForceHeadSilent forces a chain head tipset without triggering a reorg
+// operation.
+//
+// CAUTION: Use it only for testing, such as to teleport the chain to a
+// particular tipset to carry out a benchmark, verification, etc. on a chain
+// segment.
+func (cs *ChainStore) ForceHeadSilent(_ context.Context, ts *types.TipSet) error {
+	log.Warnf("(!!!) forcing a new head silently; only use this only for testing; new head: %s", ts)
+
+	cs.heaviestLk.Lock()
+	defer cs.heaviestLk.Unlock()
+	cs.heaviest = ts
+
+	err := cs.writeHead(ts)
+	if err != nil {
+		err = xerrors.Errorf("failed to write chain head: %s", err)
+	}
+	return err
+}
+
 type reorg struct {
 	old *types.TipSet
 	new *types.TipSet
@@ -525,12 +551,20 @@ func (cs *ChainStore) Contains(ts *types.TipSet) (bool, error) {
 // GetBlock fetches a BlockHeader with the supplied CID. It returns
 // blockstore.ErrNotFound if the block was not found in the BlockStore.
 func (cs *ChainStore) GetBlock(c cid.Cid) (*types.BlockHeader, error) {
-	sb, err := cs.localbs.Get(c)
-	if err != nil {
-		return nil, err
+	if cs.localviewer == nil {
+		sb, err := cs.localbs.Get(c)
+		if err != nil {
+			return nil, err
+		}
+		return types.DecodeBlock(sb.RawData())
 	}
 
-	return types.DecodeBlock(sb.RawData())
+	var blk *types.BlockHeader
+	err := cs.localviewer.View(c, func(b []byte) (err error) {
+		blk, err = types.DecodeBlock(b)
+		return err
+	})
+	return blk, err
 }
 
 func (cs *ChainStore) LoadTipSet(tsk types.TipSetKey) (*types.TipSet, error) {
@@ -775,12 +809,7 @@ func (cs *ChainStore) GetGenesis() (*types.BlockHeader, error) {
 		return nil, err
 	}
 
-	genb, err := cs.bs.Get(c)
-	if err != nil {
-		return nil, err
-	}
-
-	return types.DecodeBlock(genb.RawData())
+	return cs.GetBlock(c)
 }
 
 func (cs *ChainStore) GetCMessage(c cid.Cid) (types.ChainMsg, error) {
@@ -796,23 +825,39 @@ func (cs *ChainStore) GetCMessage(c cid.Cid) (types.ChainMsg, error) {
 }
 
 func (cs *ChainStore) GetMessage(c cid.Cid) (*types.Message, error) {
-	sb, err := cs.localbs.Get(c)
-	if err != nil {
-		log.Errorf("get message get failed: %s: %s", c, err)
-		return nil, err
+	if cs.localviewer == nil {
+		sb, err := cs.localbs.Get(c)
+		if err != nil {
+			log.Errorf("get message get failed: %s: %s", c, err)
+			return nil, err
+		}
+		return types.DecodeMessage(sb.RawData())
 	}
 
-	return types.DecodeMessage(sb.RawData())
+	var msg *types.Message
+	err := cs.localviewer.View(c, func(b []byte) (err error) {
+		msg, err = types.DecodeMessage(b)
+		return err
+	})
+	return msg, err
 }
 
 func (cs *ChainStore) GetSignedMessage(c cid.Cid) (*types.SignedMessage, error) {
-	sb, err := cs.localbs.Get(c)
-	if err != nil {
-		log.Errorf("get message get failed: %s: %s", c, err)
-		return nil, err
+	if cs.localviewer == nil {
+		sb, err := cs.localbs.Get(c)
+		if err != nil {
+			log.Errorf("get message get failed: %s: %s", c, err)
+			return nil, err
+		}
+		return types.DecodeSignedMessage(sb.RawData())
 	}
 
-	return types.DecodeSignedMessage(sb.RawData())
+	var msg *types.SignedMessage
+	err := cs.localviewer.View(c, func(b []byte) (err error) {
+		msg, err = types.DecodeSignedMessage(b)
+		return err
+	})
+	return msg, err
 }
 
 func (cs *ChainStore) readAMTCids(root cid.Cid) ([]cid.Cid, error) {

chain/store/store_test.go

@@ -3,6 +3,7 @@ package store_test
 import (
 	"bytes"
 	"context"
+	"io"
 	"testing"
 
 	datastore "github.com/ipfs/go-datastore"
@@ -51,18 +52,24 @@ func BenchmarkGetRandomness(b *testing.B) {
 		b.Fatal(err)
 	}
 
-	bds, err := lr.Datastore("/chain")
+	bs, err := lr.Blockstore(repo.BlockstoreChain)
 	if err != nil {
 		b.Fatal(err)
 	}
 
+	defer func() {
+		if c, ok := bs.(io.Closer); ok {
+			if err := c.Close(); err != nil {
+				b.Logf("WARN: failed to close blockstore: %s", err)
+			}
+		}
+	}()
+
 	mds, err := lr.Datastore("/metadata")
 	if err != nil {
 		b.Fatal(err)
 	}
 
-	bs := blockstore.NewBlockstore(bds)
-
 	cs := store.NewChainStore(bs, bs, mds, nil, nil)
 
 	b.ResetTimer()

chain/vm/vm.go

@@ -70,13 +70,32 @@ func ResolveToKeyAddr(state types.StateTree, cst cbor.IpldStore, addr address.Ad
 }
 
 var _ cbor.IpldBlockstore = (*gasChargingBlocks)(nil)
+var _ blockstore.Viewer = (*gasChargingBlocks)(nil)
 
 type gasChargingBlocks struct {
 	chargeGas func(GasCharge)
 	pricelist Pricelist
 	under     cbor.IpldBlockstore
 }
 
+func (bs *gasChargingBlocks) View(c cid.Cid, cb func([]byte) error) error {
+	if v, ok := bs.under.(blockstore.Viewer); ok {
+		bs.chargeGas(bs.pricelist.OnIpldGet())
+		return v.View(c, func(b []byte) error {
+			// we have successfully retrieved the value; charge for it, even if the user-provided function fails.
+			bs.chargeGas(newGasCharge("OnIpldViewEnd", 0, 0).WithExtra(len(b)))
+			bs.chargeGas(gasOnActorExec)
+			return cb(b)
+		})
+	}
+	// the underlying blockstore doesn't implement the viewer interface, fall back to normal Get behaviour.
+	blk, err := bs.Get(c)
+	if err == nil && blk != nil {
+		return cb(blk.RawData())
+	}
+	return err
+}
+
 func (bs *gasChargingBlocks) Get(c cid.Cid) (block.Block, error) {
 	bs.chargeGas(bs.pricelist.OnIpldGet())
 	blk, err := bs.under.Get(c)
@@ -130,10 +149,10 @@ func (vm *VM) makeRuntime(ctx context.Context, msg *types.Message, parent *Runti
 		rt.Abortf(exitcode.SysErrForbidden, "message execution exceeds call depth")
 	}
 
-	rt.cst = &cbor.BasicIpldStore{
-		Blocks: &gasChargingBlocks{rt.chargeGasFunc(2), rt.pricelist, vm.cst.Blocks},
-		Atlas:  vm.cst.Atlas,
-	}
+	cbb := &gasChargingBlocks{rt.chargeGasFunc(2), rt.pricelist, vm.cst.Blocks}
+	cst := cbor.NewCborStore(cbb)
+	cst.Atlas = vm.cst.Atlas // associate the atlas.
+	rt.cst = cst
 
 	vmm := *msg
 	resF, ok := rt.ResolveAddress(msg.From)