Merge pull request #5484 from filecoin-project/refactor/lib/blockstore
refactor blockstores
This commit is contained in:
commit
57642697bc
@ -289,7 +289,7 @@ jobs:
|
||||
- run: cd extern/filecoin-ffi && make
|
||||
- run:
|
||||
name: "go get lotus@master"
|
||||
command: cd testplans/lotus-soup && go get github.com/filecoin-project/lotus@master
|
||||
command: cd testplans/lotus-soup && go mod edit -replace=github.com/filecoin-project/lotus=../..
|
||||
- run:
|
||||
name: "build lotus-soup testplan"
|
||||
command: pushd testplans/lotus-soup && go build -tags=testground .
|
||||
|
@ -34,6 +34,12 @@ import (
|
||||
|
||||
//go:generate go run github.com/golang/mock/mockgen -destination=mocks/mock_full.go -package=mocks . FullNode
|
||||
|
||||
// ChainIO abstracts operations for accessing raw IPLD objects.
|
||||
type ChainIO interface {
|
||||
ChainReadObj(context.Context, cid.Cid) ([]byte, error)
|
||||
ChainHasObj(context.Context, cid.Cid) (bool, error)
|
||||
}
|
||||
|
||||
// FullNode API is a low-level interface to the Filecoin network full node
|
||||
type FullNode interface {
|
||||
Common
|
||||
|
@ -1,68 +0,0 @@
|
||||
package apibstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
blocks "github.com/ipfs/go-block-format"
|
||||
"github.com/ipfs/go-cid"
|
||||
"golang.org/x/xerrors"
|
||||
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
)
|
||||
|
||||
type ChainIO interface {
|
||||
ChainReadObj(context.Context, cid.Cid) ([]byte, error)
|
||||
ChainHasObj(context.Context, cid.Cid) (bool, error)
|
||||
}
|
||||
|
||||
type apiBStore struct {
|
||||
api ChainIO
|
||||
}
|
||||
|
||||
func NewAPIBlockstore(cio ChainIO) blockstore.Blockstore {
|
||||
return &apiBStore{
|
||||
api: cio,
|
||||
}
|
||||
}
|
||||
|
||||
func (a *apiBStore) DeleteBlock(cid.Cid) error {
|
||||
return xerrors.New("not supported")
|
||||
}
|
||||
|
||||
func (a *apiBStore) Has(c cid.Cid) (bool, error) {
|
||||
return a.api.ChainHasObj(context.TODO(), c)
|
||||
}
|
||||
|
||||
func (a *apiBStore) Get(c cid.Cid) (blocks.Block, error) {
|
||||
bb, err := a.api.ChainReadObj(context.TODO(), c)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return blocks.NewBlockWithCid(bb, c)
|
||||
}
|
||||
|
||||
func (a *apiBStore) GetSize(c cid.Cid) (int, error) {
|
||||
bb, err := a.api.ChainReadObj(context.TODO(), c)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return len(bb), nil
|
||||
}
|
||||
|
||||
func (a *apiBStore) Put(blocks.Block) error {
|
||||
return xerrors.New("not supported")
|
||||
}
|
||||
|
||||
func (a *apiBStore) PutMany([]blocks.Block) error {
|
||||
return xerrors.New("not supported")
|
||||
}
|
||||
|
||||
func (a *apiBStore) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) {
|
||||
return nil, xerrors.New("not supported")
|
||||
}
|
||||
|
||||
func (a *apiBStore) HashOnRead(enabled bool) {
|
||||
return
|
||||
}
|
||||
|
||||
var _ blockstore.Blockstore = &apiBStore{}
|
@ -15,7 +15,7 @@ import (
|
||||
cbor "github.com/ipfs/go-ipld-cbor"
|
||||
|
||||
"github.com/filecoin-project/lotus/api"
|
||||
"github.com/filecoin-project/lotus/api/apibstore"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/build"
|
||||
"github.com/filecoin-project/lotus/chain/actors/adt"
|
||||
"github.com/filecoin-project/lotus/chain/actors/builtin"
|
||||
@ -132,7 +132,7 @@ func TestPaymentChannels(t *testing.T, b APIBuilder, blocktime time.Duration) {
|
||||
t.Fatal("Unable to settle payment channel")
|
||||
}
|
||||
|
||||
creatorStore := adt.WrapStore(ctx, cbor.NewCborStore(apibstore.NewAPIBlockstore(paymentCreator)))
|
||||
creatorStore := adt.WrapStore(ctx, cbor.NewCborStore(blockstore.NewAPIBlockstore(paymentCreator)))
|
||||
|
||||
// wait for the receiver to submit their vouchers
|
||||
ev := events.NewEvents(ctx, paymentCreator)
|
||||
|
66
blockstore/api.go
Normal file
66
blockstore/api.go
Normal file
@ -0,0 +1,66 @@
|
||||
package blockstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
blocks "github.com/ipfs/go-block-format"
|
||||
"github.com/ipfs/go-cid"
|
||||
"golang.org/x/xerrors"
|
||||
)
|
||||
|
||||
type ChainIO interface {
|
||||
ChainReadObj(context.Context, cid.Cid) ([]byte, error)
|
||||
ChainHasObj(context.Context, cid.Cid) (bool, error)
|
||||
}
|
||||
|
||||
type apiBlockstore struct {
|
||||
api ChainIO
|
||||
}
|
||||
|
||||
// This blockstore is adapted in the constructor.
|
||||
var _ BasicBlockstore = (*apiBlockstore)(nil)
|
||||
|
||||
func NewAPIBlockstore(cio ChainIO) Blockstore {
|
||||
bs := &apiBlockstore{api: cio}
|
||||
return Adapt(bs) // return an adapted blockstore.
|
||||
}
|
||||
|
||||
func (a *apiBlockstore) DeleteBlock(cid.Cid) error {
|
||||
return xerrors.New("not supported")
|
||||
}
|
||||
|
||||
func (a *apiBlockstore) Has(c cid.Cid) (bool, error) {
|
||||
return a.api.ChainHasObj(context.TODO(), c)
|
||||
}
|
||||
|
||||
func (a *apiBlockstore) Get(c cid.Cid) (blocks.Block, error) {
|
||||
bb, err := a.api.ChainReadObj(context.TODO(), c)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return blocks.NewBlockWithCid(bb, c)
|
||||
}
|
||||
|
||||
func (a *apiBlockstore) GetSize(c cid.Cid) (int, error) {
|
||||
bb, err := a.api.ChainReadObj(context.TODO(), c)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return len(bb), nil
|
||||
}
|
||||
|
||||
func (a *apiBlockstore) Put(blocks.Block) error {
|
||||
return xerrors.New("not supported")
|
||||
}
|
||||
|
||||
func (a *apiBlockstore) PutMany([]blocks.Block) error {
|
||||
return xerrors.New("not supported")
|
||||
}
|
||||
|
||||
func (a *apiBlockstore) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) {
|
||||
return nil, xerrors.New("not supported")
|
||||
}
|
||||
|
||||
func (a *apiBlockstore) HashOnRead(enabled bool) {
|
||||
return
|
||||
}
|
@ -16,7 +16,7 @@ import (
|
||||
logger "github.com/ipfs/go-log/v2"
|
||||
pool "github.com/libp2p/go-buffer-pool"
|
||||
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
)
|
||||
|
||||
var (
|
@ -6,8 +6,9 @@ import (
|
||||
"testing"
|
||||
|
||||
blocks "github.com/ipfs/go-block-format"
|
||||
blockstore "github.com/ipfs/go-ipfs-blockstore"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
)
|
||||
|
||||
func TestBadgerBlockstore(t *testing.T) {
|
@ -8,7 +8,7 @@ import (
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
blocks "github.com/ipfs/go-block-format"
|
||||
"github.com/ipfs/go-cid"
|
||||
u "github.com/ipfs/go-ipfs-util"
|
66
blockstore/blockstore.go
Normal file
66
blockstore/blockstore.go
Normal file
@ -0,0 +1,66 @@
|
||||
package blockstore
|
||||
|
||||
import (
|
||||
"github.com/ipfs/go-cid"
|
||||
ds "github.com/ipfs/go-datastore"
|
||||
logging "github.com/ipfs/go-log/v2"
|
||||
|
||||
blockstore "github.com/ipfs/go-ipfs-blockstore"
|
||||
)
|
||||
|
||||
var log = logging.Logger("blockstore")
|
||||
|
||||
var ErrNotFound = blockstore.ErrNotFound
|
||||
|
||||
// Blockstore is the blockstore interface used by Lotus. It is the union
|
||||
// of the basic go-ipfs blockstore, with other capabilities required by Lotus,
|
||||
// e.g. View or Sync.
|
||||
type Blockstore interface {
|
||||
blockstore.Blockstore
|
||||
blockstore.Viewer
|
||||
}
|
||||
|
||||
// BasicBlockstore is an alias to the original IPFS Blockstore.
|
||||
type BasicBlockstore = blockstore.Blockstore
|
||||
|
||||
type Viewer = blockstore.Viewer
|
||||
|
||||
// WrapIDStore wraps the underlying blockstore in an "identity" blockstore.
|
||||
// The ID store filters out all puts for blocks with CIDs using the "identity"
|
||||
// hash function. It also extracts inlined blocks from CIDs using the identity
|
||||
// hash function and returns them on get/has, ignoring the contents of the
|
||||
// blockstore.
|
||||
func WrapIDStore(bstore blockstore.Blockstore) Blockstore {
|
||||
return blockstore.NewIdStore(bstore).(Blockstore)
|
||||
}
|
||||
|
||||
// FromDatastore creates a new blockstore backed by the given datastore.
|
||||
func FromDatastore(dstore ds.Batching) Blockstore {
|
||||
return WrapIDStore(blockstore.NewBlockstore(dstore))
|
||||
}
|
||||
|
||||
type adaptedBlockstore struct {
|
||||
blockstore.Blockstore
|
||||
}
|
||||
|
||||
var _ Blockstore = (*adaptedBlockstore)(nil)
|
||||
|
||||
func (a *adaptedBlockstore) View(cid cid.Cid, callback func([]byte) error) error {
|
||||
blk, err := a.Get(cid)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return callback(blk.RawData())
|
||||
}
|
||||
|
||||
// Adapt adapts a standard blockstore to a Lotus blockstore by
|
||||
// enriching it with the extra methods that Lotus requires (e.g. View, Sync).
|
||||
//
|
||||
// View proxies over to Get and calls the callback with the value supplied by Get.
|
||||
// Sync noops.
|
||||
func Adapt(bs blockstore.Blockstore) Blockstore {
|
||||
if ret, ok := bs.(Blockstore); ok {
|
||||
return ret
|
||||
}
|
||||
return &adaptedBlockstore{bs}
|
||||
}
|
166
blockstore/buffered.go
Normal file
166
blockstore/buffered.go
Normal file
@ -0,0 +1,166 @@
|
||||
package blockstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
|
||||
block "github.com/ipfs/go-block-format"
|
||||
"github.com/ipfs/go-cid"
|
||||
)
|
||||
|
||||
// buflog is a logger for the buffered blockstore. It is subscoped from the
|
||||
// blockstore logger.
|
||||
var buflog = log.Named("buf")
|
||||
|
||||
type BufferedBlockstore struct {
|
||||
read Blockstore
|
||||
write Blockstore
|
||||
}
|
||||
|
||||
func NewBuffered(base Blockstore) *BufferedBlockstore {
|
||||
var buf Blockstore
|
||||
if os.Getenv("LOTUS_DISABLE_VM_BUF") == "iknowitsabadidea" {
|
||||
buflog.Warn("VM BLOCKSTORE BUFFERING IS DISABLED")
|
||||
buf = base
|
||||
} else {
|
||||
buf = NewMemory()
|
||||
}
|
||||
|
||||
bs := &BufferedBlockstore{
|
||||
read: base,
|
||||
write: buf,
|
||||
}
|
||||
return bs
|
||||
}
|
||||
|
||||
func NewTieredBstore(r Blockstore, w Blockstore) *BufferedBlockstore {
|
||||
return &BufferedBlockstore{
|
||||
read: r,
|
||||
write: w,
|
||||
}
|
||||
}
|
||||
|
||||
var (
|
||||
_ Blockstore = (*BufferedBlockstore)(nil)
|
||||
_ Viewer = (*BufferedBlockstore)(nil)
|
||||
)
|
||||
|
||||
func (bs *BufferedBlockstore) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) {
|
||||
a, err := bs.read.AllKeysChan(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
b, err := bs.write.AllKeysChan(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
out := make(chan cid.Cid)
|
||||
go func() {
|
||||
defer close(out)
|
||||
for a != nil || b != nil {
|
||||
select {
|
||||
case val, ok := <-a:
|
||||
if !ok {
|
||||
a = nil
|
||||
} else {
|
||||
select {
|
||||
case out <- val:
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
case val, ok := <-b:
|
||||
if !ok {
|
||||
b = nil
|
||||
} else {
|
||||
select {
|
||||
case out <- val:
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (bs *BufferedBlockstore) DeleteBlock(c cid.Cid) error {
|
||||
if err := bs.read.DeleteBlock(c); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return bs.write.DeleteBlock(c)
|
||||
}
|
||||
|
||||
func (bs *BufferedBlockstore) View(c cid.Cid, callback func([]byte) error) error {
|
||||
// both stores are viewable.
|
||||
if err := bs.write.View(c, callback); err == ErrNotFound {
|
||||
// not found in write blockstore; fall through.
|
||||
} else {
|
||||
return err // propagate errors, or nil, i.e. found.
|
||||
}
|
||||
return bs.read.View(c, callback)
|
||||
}
|
||||
|
||||
func (bs *BufferedBlockstore) Get(c cid.Cid) (block.Block, error) {
|
||||
if out, err := bs.write.Get(c); err != nil {
|
||||
if err != ErrNotFound {
|
||||
return nil, err
|
||||
}
|
||||
} else {
|
||||
return out, nil
|
||||
}
|
||||
|
||||
return bs.read.Get(c)
|
||||
}
|
||||
|
||||
func (bs *BufferedBlockstore) GetSize(c cid.Cid) (int, error) {
|
||||
s, err := bs.read.GetSize(c)
|
||||
if err == ErrNotFound || s == 0 {
|
||||
return bs.write.GetSize(c)
|
||||
}
|
||||
|
||||
return s, err
|
||||
}
|
||||
|
||||
func (bs *BufferedBlockstore) Put(blk block.Block) error {
|
||||
has, err := bs.read.Has(blk.Cid()) // TODO: consider dropping this check
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if has {
|
||||
return nil
|
||||
}
|
||||
|
||||
return bs.write.Put(blk)
|
||||
}
|
||||
|
||||
func (bs *BufferedBlockstore) Has(c cid.Cid) (bool, error) {
|
||||
has, err := bs.write.Has(c)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
if has {
|
||||
return true, nil
|
||||
}
|
||||
|
||||
return bs.read.Has(c)
|
||||
}
|
||||
|
||||
func (bs *BufferedBlockstore) HashOnRead(hor bool) {
|
||||
bs.read.HashOnRead(hor)
|
||||
bs.write.HashOnRead(hor)
|
||||
}
|
||||
|
||||
func (bs *BufferedBlockstore) PutMany(blks []block.Block) error {
|
||||
return bs.write.PutMany(blks)
|
||||
}
|
||||
|
||||
func (bs *BufferedBlockstore) Read() Blockstore {
|
||||
return bs.read
|
||||
}
|
25
blockstore/cached.go
Normal file
25
blockstore/cached.go
Normal file
@ -0,0 +1,25 @@
|
||||
package blockstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
blockstore "github.com/ipfs/go-ipfs-blockstore"
|
||||
)
|
||||
|
||||
type CacheOpts = blockstore.CacheOpts
|
||||
|
||||
func DefaultCacheOpts() CacheOpts {
|
||||
return CacheOpts{
|
||||
HasBloomFilterSize: 0,
|
||||
HasBloomFilterHashes: 0,
|
||||
HasARCCacheSize: 512 << 10,
|
||||
}
|
||||
}
|
||||
|
||||
func CachedBlockstore(ctx context.Context, bs Blockstore, opts CacheOpts) (Blockstore, error) {
|
||||
cached, err := blockstore.CachedBlockstore(ctx, bs, opts)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return WrapIDStore(cached), nil
|
||||
}
|
9
blockstore/doc.go
Normal file
9
blockstore/doc.go
Normal file
@ -0,0 +1,9 @@
|
||||
// Package blockstore and subpackages contain most of the blockstore
|
||||
// implementations used by Lotus.
|
||||
//
|
||||
// Blockstores not ultimately constructed out of the building blocks in this
|
||||
// package may not work properly.
|
||||
//
|
||||
// This package re-exports parts of the go-ipfs-blockstore package such that
|
||||
// no other package needs to import it directly, for ergonomics and traceability.
|
||||
package blockstore
|
@ -9,24 +9,24 @@ import (
|
||||
|
||||
blocks "github.com/ipfs/go-block-format"
|
||||
"github.com/ipfs/go-cid"
|
||||
blockstore "github.com/ipfs/go-ipfs-blockstore"
|
||||
logging "github.com/ipfs/go-log"
|
||||
)
|
||||
|
||||
var log = logging.Logger("blockstore")
|
||||
|
||||
type FallbackStore struct {
|
||||
blockstore.Blockstore
|
||||
Blockstore
|
||||
|
||||
fallbackGetBlock func(context.Context, cid.Cid) (blocks.Block, error)
|
||||
lk sync.RWMutex
|
||||
lk sync.RWMutex
|
||||
// missFn is the function that will be invoked on a local miss to pull the
|
||||
// block from elsewhere.
|
||||
missFn func(context.Context, cid.Cid) (blocks.Block, error)
|
||||
}
|
||||
|
||||
func (fbs *FallbackStore) SetFallback(fg func(context.Context, cid.Cid) (blocks.Block, error)) {
|
||||
var _ Blockstore = (*FallbackStore)(nil)
|
||||
|
||||
func (fbs *FallbackStore) SetFallback(missFn func(context.Context, cid.Cid) (blocks.Block, error)) {
|
||||
fbs.lk.Lock()
|
||||
defer fbs.lk.Unlock()
|
||||
|
||||
fbs.fallbackGetBlock = fg
|
||||
fbs.missFn = missFn
|
||||
}
|
||||
|
||||
func (fbs *FallbackStore) getFallback(c cid.Cid) (blocks.Block, error) {
|
||||
@ -34,23 +34,23 @@ func (fbs *FallbackStore) getFallback(c cid.Cid) (blocks.Block, error) {
|
||||
fbs.lk.RLock()
|
||||
defer fbs.lk.RUnlock()
|
||||
|
||||
if fbs.fallbackGetBlock == nil {
|
||||
if fbs.missFn == nil {
|
||||
// FallbackStore wasn't configured yet (chainstore/bitswap aren't up yet)
|
||||
// Wait for a bit and retry
|
||||
fbs.lk.RUnlock()
|
||||
time.Sleep(5 * time.Second)
|
||||
fbs.lk.RLock()
|
||||
|
||||
if fbs.fallbackGetBlock == nil {
|
||||
log.Errorw("fallbackstore: fallbackGetBlock not configured yet")
|
||||
return nil, blockstore.ErrNotFound
|
||||
if fbs.missFn == nil {
|
||||
log.Errorw("fallbackstore: missFn not configured yet")
|
||||
return nil, ErrNotFound
|
||||
}
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.TODO(), 120*time.Second)
|
||||
defer cancel()
|
||||
|
||||
b, err := fbs.fallbackGetBlock(ctx, c)
|
||||
b, err := fbs.missFn(ctx, c)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@ -69,7 +69,7 @@ func (fbs *FallbackStore) Get(c cid.Cid) (blocks.Block, error) {
|
||||
switch err {
|
||||
case nil:
|
||||
return b, nil
|
||||
case blockstore.ErrNotFound:
|
||||
case ErrNotFound:
|
||||
return fbs.getFallback(c)
|
||||
default:
|
||||
return b, err
|
||||
@ -81,7 +81,7 @@ func (fbs *FallbackStore) GetSize(c cid.Cid) (int, error) {
|
||||
switch err {
|
||||
case nil:
|
||||
return sz, nil
|
||||
case blockstore.ErrNotFound:
|
||||
case ErrNotFound:
|
||||
b, err := fbs.getFallback(c)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
@ -91,5 +91,3 @@ func (fbs *FallbackStore) GetSize(c cid.Cid) (int, error) {
|
||||
return sz, err
|
||||
}
|
||||
}
|
||||
|
||||
var _ blockstore.Blockstore = &FallbackStore{}
|
@ -1,4 +1,4 @@
|
||||
package ipfsbstore
|
||||
package blockstore
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
@ -16,16 +16,16 @@ import (
|
||||
iface "github.com/ipfs/interface-go-ipfs-core"
|
||||
"github.com/ipfs/interface-go-ipfs-core/options"
|
||||
"github.com/ipfs/interface-go-ipfs-core/path"
|
||||
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
)
|
||||
|
||||
type IpfsBstore struct {
|
||||
type IPFSBlockstore struct {
|
||||
ctx context.Context
|
||||
api, offlineAPI iface.CoreAPI
|
||||
}
|
||||
|
||||
func NewIpfsBstore(ctx context.Context, onlineMode bool) (*IpfsBstore, error) {
|
||||
var _ BasicBlockstore = (*IPFSBlockstore)(nil)
|
||||
|
||||
func NewLocalIPFSBlockstore(ctx context.Context, onlineMode bool) (Blockstore, error) {
|
||||
localApi, err := httpapi.NewLocalApi()
|
||||
if err != nil {
|
||||
return nil, xerrors.Errorf("getting local ipfs api: %w", err)
|
||||
@ -34,6 +34,7 @@ func NewIpfsBstore(ctx context.Context, onlineMode bool) (*IpfsBstore, error) {
|
||||
if err != nil {
|
||||
return nil, xerrors.Errorf("setting offline mode: %s", err)
|
||||
}
|
||||
|
||||
offlineAPI := api
|
||||
if onlineMode {
|
||||
offlineAPI, err = localApi.WithOptions(options.Api.Offline(true))
|
||||
@ -42,14 +43,16 @@ func NewIpfsBstore(ctx context.Context, onlineMode bool) (*IpfsBstore, error) {
|
||||
}
|
||||
}
|
||||
|
||||
return &IpfsBstore{
|
||||
bs := &IPFSBlockstore{
|
||||
ctx: ctx,
|
||||
api: api,
|
||||
offlineAPI: offlineAPI,
|
||||
}, nil
|
||||
}
|
||||
|
||||
return Adapt(bs), nil
|
||||
}
|
||||
|
||||
func NewRemoteIpfsBstore(ctx context.Context, maddr multiaddr.Multiaddr, onlineMode bool) (*IpfsBstore, error) {
|
||||
func NewRemoteIPFSBlockstore(ctx context.Context, maddr multiaddr.Multiaddr, onlineMode bool) (Blockstore, error) {
|
||||
httpApi, err := httpapi.NewApi(maddr)
|
||||
if err != nil {
|
||||
return nil, xerrors.Errorf("setting remote ipfs api: %w", err)
|
||||
@ -58,6 +61,7 @@ func NewRemoteIpfsBstore(ctx context.Context, maddr multiaddr.Multiaddr, onlineM
|
||||
if err != nil {
|
||||
return nil, xerrors.Errorf("applying offline mode: %s", err)
|
||||
}
|
||||
|
||||
offlineAPI := api
|
||||
if onlineMode {
|
||||
offlineAPI, err = httpApi.WithOptions(options.Api.Offline(true))
|
||||
@ -66,18 +70,20 @@ func NewRemoteIpfsBstore(ctx context.Context, maddr multiaddr.Multiaddr, onlineM
|
||||
}
|
||||
}
|
||||
|
||||
return &IpfsBstore{
|
||||
bs := &IPFSBlockstore{
|
||||
ctx: ctx,
|
||||
api: api,
|
||||
offlineAPI: offlineAPI,
|
||||
}, nil
|
||||
}
|
||||
|
||||
return Adapt(bs), nil
|
||||
}
|
||||
|
||||
func (i *IpfsBstore) DeleteBlock(cid cid.Cid) error {
|
||||
func (i *IPFSBlockstore) DeleteBlock(cid cid.Cid) error {
|
||||
return xerrors.Errorf("not supported")
|
||||
}
|
||||
|
||||
func (i *IpfsBstore) Has(cid cid.Cid) (bool, error) {
|
||||
func (i *IPFSBlockstore) Has(cid cid.Cid) (bool, error) {
|
||||
_, err := i.offlineAPI.Block().Stat(i.ctx, path.IpldPath(cid))
|
||||
if err != nil {
|
||||
// The underlying client is running in Offline mode.
|
||||
@ -93,7 +99,7 @@ func (i *IpfsBstore) Has(cid cid.Cid) (bool, error) {
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func (i *IpfsBstore) Get(cid cid.Cid) (blocks.Block, error) {
|
||||
func (i *IPFSBlockstore) Get(cid cid.Cid) (blocks.Block, error) {
|
||||
rd, err := i.api.Block().Get(i.ctx, path.IpldPath(cid))
|
||||
if err != nil {
|
||||
return nil, xerrors.Errorf("getting ipfs block: %w", err)
|
||||
@ -107,7 +113,7 @@ func (i *IpfsBstore) Get(cid cid.Cid) (blocks.Block, error) {
|
||||
return blocks.NewBlockWithCid(data, cid)
|
||||
}
|
||||
|
||||
func (i *IpfsBstore) GetSize(cid cid.Cid) (int, error) {
|
||||
func (i *IPFSBlockstore) GetSize(cid cid.Cid) (int, error) {
|
||||
st, err := i.api.Block().Stat(i.ctx, path.IpldPath(cid))
|
||||
if err != nil {
|
||||
return 0, xerrors.Errorf("getting ipfs block: %w", err)
|
||||
@ -116,7 +122,7 @@ func (i *IpfsBstore) GetSize(cid cid.Cid) (int, error) {
|
||||
return st.Size(), nil
|
||||
}
|
||||
|
||||
func (i *IpfsBstore) Put(block blocks.Block) error {
|
||||
func (i *IPFSBlockstore) Put(block blocks.Block) error {
|
||||
mhd, err := multihash.Decode(block.Cid().Hash())
|
||||
if err != nil {
|
||||
return err
|
||||
@ -128,7 +134,7 @@ func (i *IpfsBstore) Put(block blocks.Block) error {
|
||||
return err
|
||||
}
|
||||
|
||||
func (i *IpfsBstore) PutMany(blocks []blocks.Block) error {
|
||||
func (i *IPFSBlockstore) PutMany(blocks []blocks.Block) error {
|
||||
// TODO: could be done in parallel
|
||||
|
||||
for _, block := range blocks {
|
||||
@ -140,12 +146,10 @@ func (i *IpfsBstore) PutMany(blocks []blocks.Block) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (i *IpfsBstore) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) {
|
||||
func (i *IPFSBlockstore) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) {
|
||||
return nil, xerrors.Errorf("not supported")
|
||||
}
|
||||
|
||||
func (i *IpfsBstore) HashOnRead(enabled bool) {
|
||||
func (i *IPFSBlockstore) HashOnRead(enabled bool) {
|
||||
return // TODO: We could technically support this, but..
|
||||
}
|
||||
|
||||
var _ blockstore.Blockstore = &IpfsBstore{}
|
@ -7,20 +7,25 @@ import (
|
||||
"github.com/ipfs/go-cid"
|
||||
)
|
||||
|
||||
// MemStore is a terminal blockstore that keeps blocks in memory.
|
||||
type MemStore map[cid.Cid]blocks.Block
|
||||
// NewMemory returns a temporary memory-backed blockstore.
|
||||
func NewMemory() MemBlockstore {
|
||||
return make(MemBlockstore)
|
||||
}
|
||||
|
||||
func (m MemStore) DeleteBlock(k cid.Cid) error {
|
||||
// MemBlockstore is a terminal blockstore that keeps blocks in memory.
|
||||
type MemBlockstore map[cid.Cid]blocks.Block
|
||||
|
||||
func (m MemBlockstore) DeleteBlock(k cid.Cid) error {
|
||||
delete(m, k)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m MemStore) Has(k cid.Cid) (bool, error) {
|
||||
func (m MemBlockstore) Has(k cid.Cid) (bool, error) {
|
||||
_, ok := m[k]
|
||||
return ok, nil
|
||||
}
|
||||
|
||||
func (m MemStore) View(k cid.Cid, callback func([]byte) error) error {
|
||||
func (m MemBlockstore) View(k cid.Cid, callback func([]byte) error) error {
|
||||
b, ok := m[k]
|
||||
if !ok {
|
||||
return ErrNotFound
|
||||
@ -28,7 +33,7 @@ func (m MemStore) View(k cid.Cid, callback func([]byte) error) error {
|
||||
return callback(b.RawData())
|
||||
}
|
||||
|
||||
func (m MemStore) Get(k cid.Cid) (blocks.Block, error) {
|
||||
func (m MemBlockstore) Get(k cid.Cid) (blocks.Block, error) {
|
||||
b, ok := m[k]
|
||||
if !ok {
|
||||
return nil, ErrNotFound
|
||||
@ -37,7 +42,7 @@ func (m MemStore) Get(k cid.Cid) (blocks.Block, error) {
|
||||
}
|
||||
|
||||
// GetSize returns the CIDs mapped BlockSize
|
||||
func (m MemStore) GetSize(k cid.Cid) (int, error) {
|
||||
func (m MemBlockstore) GetSize(k cid.Cid) (int, error) {
|
||||
b, ok := m[k]
|
||||
if !ok {
|
||||
return 0, ErrNotFound
|
||||
@ -46,7 +51,7 @@ func (m MemStore) GetSize(k cid.Cid) (int, error) {
|
||||
}
|
||||
|
||||
// Put puts a given block to the underlying datastore
|
||||
func (m MemStore) Put(b blocks.Block) error {
|
||||
func (m MemBlockstore) Put(b blocks.Block) error {
|
||||
// Convert to a basic block for safety, but try to reuse the existing
|
||||
// block if it's already a basic block.
|
||||
k := b.Cid()
|
||||
@ -64,7 +69,7 @@ func (m MemStore) Put(b blocks.Block) error {
|
||||
|
||||
// PutMany puts a slice of blocks at the same time using batching
|
||||
// capabilities of the underlying datastore whenever possible.
|
||||
func (m MemStore) PutMany(bs []blocks.Block) error {
|
||||
func (m MemBlockstore) PutMany(bs []blocks.Block) error {
|
||||
for _, b := range bs {
|
||||
_ = m.Put(b) // can't fail
|
||||
}
|
||||
@ -74,7 +79,7 @@ func (m MemStore) PutMany(bs []blocks.Block) error {
|
||||
// AllKeysChan returns a channel from which
|
||||
// the CIDs in the Blockstore can be read. It should respect
|
||||
// the given context, closing the channel if it becomes Done.
|
||||
func (m MemStore) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) {
|
||||
func (m MemBlockstore) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) {
|
||||
ch := make(chan cid.Cid, len(m))
|
||||
for k := range m {
|
||||
ch <- k
|
||||
@ -85,6 +90,6 @@ func (m MemStore) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) {
|
||||
|
||||
// HashOnRead specifies if every read block should be
|
||||
// rehashed to make sure it matches its CID.
|
||||
func (m MemStore) HashOnRead(enabled bool) {
|
||||
func (m MemBlockstore) HashOnRead(enabled bool) {
|
||||
// no-op
|
||||
}
|
75
blockstore/sync.go
Normal file
75
blockstore/sync.go
Normal file
@ -0,0 +1,75 @@
|
||||
package blockstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
|
||||
blocks "github.com/ipfs/go-block-format"
|
||||
"github.com/ipfs/go-cid"
|
||||
)
|
||||
|
||||
// NewMemorySync returns a thread-safe in-memory blockstore.
|
||||
func NewMemorySync() *SyncBlockstore {
|
||||
return &SyncBlockstore{bs: make(MemBlockstore)}
|
||||
}
|
||||
|
||||
// SyncBlockstore is a terminal blockstore that is a synchronized version
|
||||
// of MemBlockstore.
|
||||
type SyncBlockstore struct {
|
||||
mu sync.RWMutex
|
||||
bs MemBlockstore // specifically use a memStore to save indirection overhead.
|
||||
}
|
||||
|
||||
func (m *SyncBlockstore) DeleteBlock(k cid.Cid) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
return m.bs.DeleteBlock(k)
|
||||
}
|
||||
|
||||
func (m *SyncBlockstore) Has(k cid.Cid) (bool, error) {
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
return m.bs.Has(k)
|
||||
}
|
||||
|
||||
func (m *SyncBlockstore) View(k cid.Cid, callback func([]byte) error) error {
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
|
||||
return m.bs.View(k, callback)
|
||||
}
|
||||
|
||||
func (m *SyncBlockstore) Get(k cid.Cid) (blocks.Block, error) {
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
return m.bs.Get(k)
|
||||
}
|
||||
|
||||
func (m *SyncBlockstore) GetSize(k cid.Cid) (int, error) {
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
return m.bs.GetSize(k)
|
||||
}
|
||||
|
||||
func (m *SyncBlockstore) Put(b blocks.Block) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
return m.bs.Put(b)
|
||||
}
|
||||
|
||||
func (m *SyncBlockstore) PutMany(bs []blocks.Block) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
return m.bs.PutMany(bs)
|
||||
}
|
||||
|
||||
func (m *SyncBlockstore) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) {
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
// this blockstore implementation doesn't do any async work.
|
||||
return m.bs.AllKeysChan(ctx)
|
||||
}
|
||||
|
||||
func (m *SyncBlockstore) HashOnRead(enabled bool) {
|
||||
// noop
|
||||
}
|
@ -1,4 +1,4 @@
|
||||
package timedbs
|
||||
package blockstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
@ -10,37 +10,37 @@ import (
|
||||
"github.com/ipfs/go-cid"
|
||||
"github.com/raulk/clock"
|
||||
"go.uber.org/multierr"
|
||||
|
||||
"github.com/filecoin-project/lotus/build"
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
)
|
||||
|
||||
// TimedCacheBS is a blockstore that keeps blocks for at least the specified
|
||||
// caching interval before discarding them. Garbage collection must be started
|
||||
// and stopped by calling Start/Stop.
|
||||
// TimedCacheBlockstore is a blockstore that keeps blocks for at least the
|
||||
// specified caching interval before discarding them. Garbage collection must
|
||||
// be started and stopped by calling Start/Stop.
|
||||
//
|
||||
// Under the covers, it's implemented with an active and an inactive blockstore
|
||||
// that are rotated every cache time interval. This means all blocks will be
|
||||
// stored at most 2x the cache interval.
|
||||
type TimedCacheBS struct {
|
||||
//
|
||||
// Create a new instance by calling the NewTimedCacheBlockstore constructor.
|
||||
type TimedCacheBlockstore struct {
|
||||
mu sync.RWMutex
|
||||
active, inactive blockstore.MemStore
|
||||
active, inactive MemBlockstore
|
||||
clock clock.Clock
|
||||
interval time.Duration
|
||||
closeCh chan struct{}
|
||||
doneRotatingCh chan struct{}
|
||||
}
|
||||
|
||||
func NewTimedCacheBS(cacheTime time.Duration) *TimedCacheBS {
|
||||
return &TimedCacheBS{
|
||||
active: blockstore.NewTemporary(),
|
||||
inactive: blockstore.NewTemporary(),
|
||||
interval: cacheTime,
|
||||
clock: build.Clock,
|
||||
func NewTimedCacheBlockstore(interval time.Duration) *TimedCacheBlockstore {
|
||||
b := &TimedCacheBlockstore{
|
||||
active: NewMemory(),
|
||||
inactive: NewMemory(),
|
||||
interval: interval,
|
||||
clock: clock.New(),
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
||||
func (t *TimedCacheBS) Start(ctx context.Context) error {
|
||||
func (t *TimedCacheBlockstore) Start(_ context.Context) error {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
if t.closeCh != nil {
|
||||
@ -65,11 +65,11 @@ func (t *TimedCacheBS) Start(ctx context.Context) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (t *TimedCacheBS) Stop(ctx context.Context) error {
|
||||
func (t *TimedCacheBlockstore) Stop(_ context.Context) error {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
if t.closeCh == nil {
|
||||
return fmt.Errorf("not started started")
|
||||
return fmt.Errorf("not started")
|
||||
}
|
||||
select {
|
||||
case <-t.closeCh:
|
||||
@ -80,15 +80,15 @@ func (t *TimedCacheBS) Stop(ctx context.Context) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (t *TimedCacheBS) rotate() {
|
||||
newBs := blockstore.NewTemporary()
|
||||
func (t *TimedCacheBlockstore) rotate() {
|
||||
newBs := NewMemory()
|
||||
|
||||
t.mu.Lock()
|
||||
t.inactive, t.active = t.active, newBs
|
||||
t.mu.Unlock()
|
||||
}
|
||||
|
||||
func (t *TimedCacheBS) Put(b blocks.Block) error {
|
||||
func (t *TimedCacheBlockstore) Put(b blocks.Block) error {
|
||||
// Don't check the inactive set here. We want to keep this block for at
|
||||
// least one interval.
|
||||
t.mu.Lock()
|
||||
@ -96,33 +96,43 @@ func (t *TimedCacheBS) Put(b blocks.Block) error {
|
||||
return t.active.Put(b)
|
||||
}
|
||||
|
||||
func (t *TimedCacheBS) PutMany(bs []blocks.Block) error {
|
||||
func (t *TimedCacheBlockstore) PutMany(bs []blocks.Block) error {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
return t.active.PutMany(bs)
|
||||
}
|
||||
|
||||
func (t *TimedCacheBS) Get(k cid.Cid) (blocks.Block, error) {
|
||||
func (t *TimedCacheBlockstore) View(k cid.Cid, callback func([]byte) error) error {
|
||||
t.mu.RLock()
|
||||
defer t.mu.RUnlock()
|
||||
err := t.active.View(k, callback)
|
||||
if err == ErrNotFound {
|
||||
err = t.inactive.View(k, callback)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func (t *TimedCacheBlockstore) Get(k cid.Cid) (blocks.Block, error) {
|
||||
t.mu.RLock()
|
||||
defer t.mu.RUnlock()
|
||||
b, err := t.active.Get(k)
|
||||
if err == blockstore.ErrNotFound {
|
||||
if err == ErrNotFound {
|
||||
b, err = t.inactive.Get(k)
|
||||
}
|
||||
return b, err
|
||||
}
|
||||
|
||||
func (t *TimedCacheBS) GetSize(k cid.Cid) (int, error) {
|
||||
func (t *TimedCacheBlockstore) GetSize(k cid.Cid) (int, error) {
|
||||
t.mu.RLock()
|
||||
defer t.mu.RUnlock()
|
||||
size, err := t.active.GetSize(k)
|
||||
if err == blockstore.ErrNotFound {
|
||||
if err == ErrNotFound {
|
||||
size, err = t.inactive.GetSize(k)
|
||||
}
|
||||
return size, err
|
||||
}
|
||||
|
||||
func (t *TimedCacheBS) Has(k cid.Cid) (bool, error) {
|
||||
func (t *TimedCacheBlockstore) Has(k cid.Cid) (bool, error) {
|
||||
t.mu.RLock()
|
||||
defer t.mu.RUnlock()
|
||||
if has, err := t.active.Has(k); err != nil {
|
||||
@ -133,17 +143,17 @@ func (t *TimedCacheBS) Has(k cid.Cid) (bool, error) {
|
||||
return t.inactive.Has(k)
|
||||
}
|
||||
|
||||
func (t *TimedCacheBS) HashOnRead(_ bool) {
|
||||
func (t *TimedCacheBlockstore) HashOnRead(_ bool) {
|
||||
// no-op
|
||||
}
|
||||
|
||||
func (t *TimedCacheBS) DeleteBlock(k cid.Cid) error {
|
||||
func (t *TimedCacheBlockstore) DeleteBlock(k cid.Cid) error {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
return multierr.Combine(t.active.DeleteBlock(k), t.inactive.DeleteBlock(k))
|
||||
}
|
||||
|
||||
func (t *TimedCacheBS) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) {
|
||||
func (t *TimedCacheBlockstore) AllKeysChan(_ context.Context) (<-chan cid.Cid, error) {
|
||||
t.mu.RLock()
|
||||
defer t.mu.RUnlock()
|
||||
|
@ -1,4 +1,4 @@
|
||||
package timedbs
|
||||
package blockstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
@ -12,8 +12,8 @@ import (
|
||||
"github.com/ipfs/go-cid"
|
||||
)
|
||||
|
||||
func TestTimedBSSimple(t *testing.T) {
|
||||
tc := NewTimedCacheBS(10 * time.Millisecond)
|
||||
func TestTimedCacheBlockstoreSimple(t *testing.T) {
|
||||
tc := NewTimedCacheBlockstore(10 * time.Millisecond)
|
||||
mClock := clock.NewMock()
|
||||
mClock.Set(time.Now())
|
||||
tc.clock = mClock
|
@ -16,7 +16,7 @@ import (
|
||||
builtin2 "github.com/filecoin-project/specs-actors/v2/actors/builtin"
|
||||
adt2 "github.com/filecoin-project/specs-actors/v2/actors/util/adt"
|
||||
|
||||
bstore "github.com/filecoin-project/lotus/lib/blockstore"
|
||||
bstore "github.com/filecoin-project/lotus/blockstore"
|
||||
)
|
||||
|
||||
func TestDiffAdtArray(t *testing.T) {
|
||||
@ -295,7 +295,7 @@ func (t *TestDiffArray) Remove(key uint64, val *typegen.Deferred) error {
|
||||
|
||||
func newContextStore() Store {
|
||||
ctx := context.Background()
|
||||
bs := bstore.NewTemporarySync()
|
||||
bs := bstore.NewMemorySync()
|
||||
store := cbornode.NewCborStore(bs)
|
||||
return WrapStore(ctx, store)
|
||||
}
|
||||
|
@ -5,8 +5,8 @@ import (
|
||||
"sync"
|
||||
|
||||
"github.com/filecoin-project/go-address"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
"github.com/ipfs/go-cid"
|
||||
"golang.org/x/xerrors"
|
||||
)
|
||||
|
@ -3,6 +3,7 @@ package state
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/filecoin-project/lotus/api"
|
||||
"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
|
||||
|
||||
"github.com/filecoin-project/go-address"
|
||||
@ -10,7 +11,7 @@ import (
|
||||
"github.com/filecoin-project/go-state-types/big"
|
||||
cbor "github.com/ipfs/go-ipld-cbor"
|
||||
|
||||
"github.com/filecoin-project/lotus/api/apibstore"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/chain/actors/adt"
|
||||
init_ "github.com/filecoin-project/lotus/chain/actors/builtin/init"
|
||||
"github.com/filecoin-project/lotus/chain/actors/builtin/market"
|
||||
@ -23,7 +24,7 @@ type UserData interface{}
|
||||
|
||||
// ChainAPI abstracts out calls made by this class to external APIs
|
||||
type ChainAPI interface {
|
||||
apibstore.ChainIO
|
||||
api.ChainIO
|
||||
StateGetActor(ctx context.Context, actor address.Address, tsk types.TipSetKey) (*types.Actor, error)
|
||||
}
|
||||
|
||||
@ -36,7 +37,7 @@ type StatePredicates struct {
|
||||
func NewStatePredicates(api ChainAPI) *StatePredicates {
|
||||
return &StatePredicates{
|
||||
api: api,
|
||||
cst: cbor.NewCborStore(apibstore.NewAPIBlockstore(api)),
|
||||
cst: cbor.NewCborStore(blockstore.NewAPIBlockstore(api)),
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -23,9 +23,9 @@ import (
|
||||
adt2 "github.com/filecoin-project/specs-actors/v2/actors/util/adt"
|
||||
tutils "github.com/filecoin-project/specs-actors/v2/support/testing"
|
||||
|
||||
bstore "github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/chain/actors/builtin/market"
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
bstore "github.com/filecoin-project/lotus/lib/blockstore"
|
||||
)
|
||||
|
||||
var dummyCid cid.Cid
|
||||
@ -36,7 +36,7 @@ func init() {
|
||||
|
||||
func TestMarketPredicates(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
bs := bstore.NewTemporarySync()
|
||||
bs := bstore.NewMemorySync()
|
||||
store := adt2.WrapStore(ctx, cbornode.NewCborStore(bs))
|
||||
|
||||
oldDeal1 := &market2.DealState{
|
||||
@ -334,7 +334,7 @@ func TestMarketPredicates(t *testing.T) {
|
||||
|
||||
func TestMinerSectorChange(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
bs := bstore.NewTemporarySync()
|
||||
bs := bstore.NewMemorySync()
|
||||
store := adt2.WrapStore(ctx, cbornode.NewCborStore(bs))
|
||||
|
||||
nextID := uint64(0)
|
||||
|
@ -27,6 +27,7 @@ import (
|
||||
proof2 "github.com/filecoin-project/specs-actors/v2/actors/runtime/proof"
|
||||
|
||||
"github.com/filecoin-project/lotus/api"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/build"
|
||||
"github.com/filecoin-project/lotus/chain/actors/policy"
|
||||
"github.com/filecoin-project/lotus/chain/beacon"
|
||||
@ -40,7 +41,6 @@ import (
|
||||
"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper"
|
||||
"github.com/filecoin-project/lotus/genesis"
|
||||
"github.com/filecoin-project/lotus/journal"
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
"github.com/filecoin-project/lotus/lib/sigs"
|
||||
"github.com/filecoin-project/lotus/node/repo"
|
||||
)
|
||||
|
@ -8,8 +8,8 @@ import (
|
||||
"github.com/filecoin-project/specs-actors/actors/builtin"
|
||||
cbor "github.com/ipfs/go-ipld-cbor"
|
||||
|
||||
bstore "github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
bstore "github.com/filecoin-project/lotus/lib/blockstore"
|
||||
)
|
||||
|
||||
func SetupSystemActor(bs bstore.Blockstore) (*types.Actor, error) {
|
||||
|
@ -16,9 +16,9 @@ import (
|
||||
cbg "github.com/whyrusleeping/cbor-gen"
|
||||
"golang.org/x/xerrors"
|
||||
|
||||
bstore "github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
"github.com/filecoin-project/lotus/genesis"
|
||||
bstore "github.com/filecoin-project/lotus/lib/blockstore"
|
||||
)
|
||||
|
||||
func SetupInitActor(bs bstore.Blockstore, netname string, initialActors []genesis.Actor, rootVerifier genesis.Actor, remainder genesis.Actor) (int64, *types.Actor, map[address.Address]address.Address, error) {
|
||||
|
@ -9,9 +9,9 @@ import (
|
||||
reward0 "github.com/filecoin-project/specs-actors/actors/builtin/reward"
|
||||
cbor "github.com/ipfs/go-ipld-cbor"
|
||||
|
||||
bstore "github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/build"
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
bstore "github.com/filecoin-project/lotus/lib/blockstore"
|
||||
)
|
||||
|
||||
func SetupRewardActor(bs bstore.Blockstore, qaPower big.Int) (*types.Actor, error) {
|
||||
|
@ -7,8 +7,8 @@ import (
|
||||
"github.com/filecoin-project/specs-actors/actors/builtin/cron"
|
||||
cbor "github.com/ipfs/go-ipld-cbor"
|
||||
|
||||
bstore "github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
bstore "github.com/filecoin-project/lotus/lib/blockstore"
|
||||
)
|
||||
|
||||
func SetupCronActor(bs bstore.Blockstore) (*types.Actor, error) {
|
||||
|
@ -9,8 +9,8 @@ import (
|
||||
power0 "github.com/filecoin-project/specs-actors/actors/builtin/power"
|
||||
cbor "github.com/ipfs/go-ipld-cbor"
|
||||
|
||||
bstore "github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
bstore "github.com/filecoin-project/lotus/lib/blockstore"
|
||||
)
|
||||
|
||||
func SetupStoragePowerActor(bs bstore.Blockstore) (*types.Actor, error) {
|
||||
|
@ -8,8 +8,8 @@ import (
|
||||
"github.com/filecoin-project/specs-actors/actors/util/adt"
|
||||
cbor "github.com/ipfs/go-ipld-cbor"
|
||||
|
||||
bstore "github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
bstore "github.com/filecoin-project/lotus/lib/blockstore"
|
||||
)
|
||||
|
||||
func SetupStorageMarketActor(bs bstore.Blockstore) (*types.Actor, error) {
|
||||
|
@ -10,8 +10,8 @@ import (
|
||||
verifreg0 "github.com/filecoin-project/specs-actors/actors/builtin/verifreg"
|
||||
"github.com/filecoin-project/specs-actors/actors/util/adt"
|
||||
|
||||
bstore "github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
bstore "github.com/filecoin-project/lotus/lib/blockstore"
|
||||
)
|
||||
|
||||
var RootVerifierID address.Address
|
||||
|
@ -26,13 +26,13 @@ import (
|
||||
verifreg0 "github.com/filecoin-project/specs-actors/actors/builtin/verifreg"
|
||||
adt0 "github.com/filecoin-project/specs-actors/actors/util/adt"
|
||||
|
||||
bstore "github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/build"
|
||||
"github.com/filecoin-project/lotus/chain/state"
|
||||
"github.com/filecoin-project/lotus/chain/store"
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
"github.com/filecoin-project/lotus/chain/vm"
|
||||
"github.com/filecoin-project/lotus/genesis"
|
||||
bstore "github.com/filecoin-project/lotus/lib/blockstore"
|
||||
"github.com/filecoin-project/lotus/lib/sigs"
|
||||
)
|
||||
|
||||
|
@ -15,6 +15,7 @@ import (
|
||||
"github.com/filecoin-project/go-state-types/abi"
|
||||
"github.com/filecoin-project/go-state-types/big"
|
||||
"github.com/filecoin-project/go-state-types/network"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/build"
|
||||
"github.com/filecoin-project/lotus/chain/actors/adt"
|
||||
"github.com/filecoin-project/lotus/chain/actors/builtin"
|
||||
@ -24,8 +25,6 @@ import (
|
||||
"github.com/filecoin-project/lotus/chain/store"
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
"github.com/filecoin-project/lotus/chain/vm"
|
||||
bstore "github.com/filecoin-project/lotus/lib/blockstore"
|
||||
"github.com/filecoin-project/lotus/lib/bufbstore"
|
||||
builtin0 "github.com/filecoin-project/specs-actors/actors/builtin"
|
||||
miner0 "github.com/filecoin-project/specs-actors/actors/builtin/miner"
|
||||
multisig0 "github.com/filecoin-project/specs-actors/actors/builtin/multisig"
|
||||
@ -793,7 +792,7 @@ func UpgradeRefuel(ctx context.Context, sm *StateManager, _ MigrationCache, cb E
|
||||
}
|
||||
|
||||
func UpgradeActorsV2(ctx context.Context, sm *StateManager, _ MigrationCache, cb ExecCallback, root cid.Cid, epoch abi.ChainEpoch, ts *types.TipSet) (cid.Cid, error) {
|
||||
buf := bufbstore.NewTieredBstore(sm.cs.Blockstore(), bstore.NewTemporarySync())
|
||||
buf := blockstore.NewTieredBstore(sm.cs.Blockstore(), blockstore.NewMemorySync())
|
||||
store := store.ActorStore(ctx, buf)
|
||||
|
||||
info, err := store.Put(ctx, new(types.StateInfo0))
|
||||
@ -1010,7 +1009,7 @@ func upgradeActorsV3Common(
|
||||
root cid.Cid, epoch abi.ChainEpoch, ts *types.TipSet,
|
||||
config nv10.Config,
|
||||
) (cid.Cid, error) {
|
||||
buf := bufbstore.NewTieredBstore(sm.cs.Blockstore(), bstore.NewTemporarySync())
|
||||
buf := blockstore.NewTieredBstore(sm.cs.Blockstore(), blockstore.NewMemorySync())
|
||||
store := store.ActorStore(ctx, buf)
|
||||
|
||||
// Load the state root.
|
||||
|
@ -6,10 +6,10 @@ import (
|
||||
"testing"
|
||||
|
||||
"github.com/filecoin-project/go-state-types/abi"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/chain/gen"
|
||||
"github.com/filecoin-project/lotus/chain/store"
|
||||
"github.com/filecoin-project/lotus/chain/types/mock"
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
datastore "github.com/ipfs/go-datastore"
|
||||
syncds "github.com/ipfs/go-datastore/sync"
|
||||
"github.com/stretchr/testify/assert"
|
||||
@ -30,7 +30,7 @@ func TestIndexSeeks(t *testing.T) {
|
||||
|
||||
ctx := context.TODO()
|
||||
|
||||
nbs := blockstore.NewTemporarySync()
|
||||
nbs := blockstore.NewMemorySync()
|
||||
cs := store.NewChainStore(nbs, nbs, syncds.MutexWrap(datastore.NewMapDatastore()), nil, nil)
|
||||
defer cs.Close() //nolint:errcheck
|
||||
|
||||
|
@ -23,12 +23,12 @@ import (
|
||||
blockadt "github.com/filecoin-project/specs-actors/actors/util/adt"
|
||||
|
||||
"github.com/filecoin-project/lotus/api"
|
||||
bstore "github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/build"
|
||||
"github.com/filecoin-project/lotus/chain/actors/adt"
|
||||
"github.com/filecoin-project/lotus/chain/actors/builtin"
|
||||
"github.com/filecoin-project/lotus/chain/vm"
|
||||
"github.com/filecoin-project/lotus/journal"
|
||||
bstore "github.com/filecoin-project/lotus/lib/blockstore"
|
||||
"github.com/filecoin-project/lotus/metrics"
|
||||
|
||||
"go.opencensus.io/stats"
|
||||
|
@ -11,12 +11,12 @@ import (
|
||||
"github.com/filecoin-project/go-state-types/abi"
|
||||
"github.com/filecoin-project/go-state-types/crypto"
|
||||
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/chain/actors/policy"
|
||||
"github.com/filecoin-project/lotus/chain/gen"
|
||||
"github.com/filecoin-project/lotus/chain/stmgr"
|
||||
"github.com/filecoin-project/lotus/chain/store"
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
"github.com/filecoin-project/lotus/node/repo"
|
||||
)
|
||||
|
||||
@ -104,7 +104,7 @@ func TestChainExportImport(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
nbs := blockstore.NewTemporary()
|
||||
nbs := blockstore.NewMemory()
|
||||
cs := store.NewChainStore(nbs, nbs, datastore.NewMapDatastore(), nil, nil)
|
||||
defer cs.Close() //nolint:errcheck
|
||||
|
||||
@ -139,7 +139,7 @@ func TestChainExportImportFull(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
nbs := blockstore.NewTemporary()
|
||||
nbs := blockstore.NewMemory()
|
||||
cs := store.NewChainStore(nbs, nbs, datastore.NewMapDatastore(), nil, nil)
|
||||
defer cs.Close() //nolint:errcheck
|
||||
|
||||
|
@ -7,13 +7,13 @@ import (
|
||||
"time"
|
||||
|
||||
address "github.com/filecoin-project/go-address"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/build"
|
||||
"github.com/filecoin-project/lotus/chain"
|
||||
"github.com/filecoin-project/lotus/chain/messagepool"
|
||||
"github.com/filecoin-project/lotus/chain/stmgr"
|
||||
"github.com/filecoin-project/lotus/chain/store"
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
"github.com/filecoin-project/lotus/lib/sigs"
|
||||
"github.com/filecoin-project/lotus/metrics"
|
||||
"github.com/filecoin-project/lotus/node/impl/client"
|
||||
@ -392,7 +392,7 @@ func (bv *BlockValidator) isChainNearSynced() bool {
|
||||
func (bv *BlockValidator) validateMsgMeta(ctx context.Context, msg *types.BlockMsg) error {
|
||||
// TODO there has to be a simpler way to do this without the blockstore dance
|
||||
// block headers use adt0
|
||||
store := blockadt.WrapStore(ctx, cbor.NewCborStore(blockstore.NewTemporary()))
|
||||
store := blockadt.WrapStore(ctx, cbor.NewCborStore(blockstore.NewMemory()))
|
||||
bmArr := blockadt.MakeEmptyArray(store)
|
||||
smArr := blockadt.MakeEmptyArray(store)
|
||||
|
||||
|
@ -44,6 +44,7 @@ import (
|
||||
proof2 "github.com/filecoin-project/specs-actors/v2/actors/runtime/proof"
|
||||
|
||||
"github.com/filecoin-project/lotus/api"
|
||||
bstore "github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/build"
|
||||
"github.com/filecoin-project/lotus/chain/actors/builtin/power"
|
||||
"github.com/filecoin-project/lotus/chain/beacon"
|
||||
@ -54,7 +55,6 @@ import (
|
||||
"github.com/filecoin-project/lotus/chain/store"
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
"github.com/filecoin-project/lotus/chain/vm"
|
||||
bstore "github.com/filecoin-project/lotus/lib/blockstore"
|
||||
"github.com/filecoin-project/lotus/lib/sigs"
|
||||
"github.com/filecoin-project/lotus/metrics"
|
||||
)
|
||||
@ -321,7 +321,7 @@ func (syncer *Syncer) ValidateMsgMeta(fblk *types.FullBlock) error {
|
||||
|
||||
// We use a temporary bstore here to avoid writing intermediate pieces
|
||||
// into the blockstore.
|
||||
blockstore := bstore.NewTemporary()
|
||||
blockstore := bstore.NewMemory()
|
||||
cst := cbor.NewCborStore(blockstore)
|
||||
|
||||
var bcids, scids []cid.Cid
|
||||
@ -1102,7 +1102,7 @@ func (syncer *Syncer) checkBlockMessages(ctx context.Context, b *types.FullBlock
|
||||
}
|
||||
|
||||
// Validate message arrays in a temporary blockstore.
|
||||
tmpbs := bstore.NewTemporary()
|
||||
tmpbs := bstore.NewMemory()
|
||||
tmpstore := blockadt.WrapStore(ctx, cbor.NewCborStore(tmpbs))
|
||||
|
||||
bmArr := blockadt.MakeEmptyArray(tmpstore)
|
||||
@ -1553,7 +1553,7 @@ func (syncer *Syncer) iterFullTipsets(ctx context.Context, headers []*types.TipS
|
||||
|
||||
for bsi := 0; bsi < len(bstout); bsi++ {
|
||||
// temp storage so we don't persist data we dont want to
|
||||
bs := bstore.NewTemporary()
|
||||
bs := bstore.NewMemory()
|
||||
blks := cbor.NewCborStore(bs)
|
||||
|
||||
this := headers[i-bsi]
|
||||
|
@ -28,6 +28,7 @@ import (
|
||||
"github.com/filecoin-project/go-state-types/exitcode"
|
||||
"github.com/filecoin-project/go-state-types/network"
|
||||
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/build"
|
||||
"github.com/filecoin-project/lotus/chain/actors/adt"
|
||||
"github.com/filecoin-project/lotus/chain/actors/aerrors"
|
||||
@ -36,9 +37,6 @@ import (
|
||||
"github.com/filecoin-project/lotus/chain/actors/builtin/reward"
|
||||
"github.com/filecoin-project/lotus/chain/state"
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
bstore "github.com/filecoin-project/lotus/lib/blockstore"
|
||||
"github.com/filecoin-project/lotus/lib/bufbstore"
|
||||
)
|
||||
|
||||
const MaxCallDepth = 4096
|
||||
@ -208,7 +206,7 @@ type VM struct {
|
||||
cstate *state.StateTree
|
||||
base cid.Cid
|
||||
cst *cbor.BasicIpldStore
|
||||
buf *bufbstore.BufferedBS
|
||||
buf *blockstore.BufferedBlockstore
|
||||
blockHeight abi.ChainEpoch
|
||||
areg *ActorRegistry
|
||||
rand Rand
|
||||
@ -224,7 +222,7 @@ type VMOpts struct {
|
||||
StateBase cid.Cid
|
||||
Epoch abi.ChainEpoch
|
||||
Rand Rand
|
||||
Bstore bstore.Blockstore
|
||||
Bstore blockstore.Blockstore
|
||||
Syscalls SyscallBuilder
|
||||
CircSupplyCalc CircSupplyCalculator
|
||||
NtwkVersion NtwkVersionGetter // TODO: stebalien: In what cases do we actually need this? It seems like even when creating new networks we want to use the 'global'/build-default version getter
|
||||
@ -233,7 +231,7 @@ type VMOpts struct {
|
||||
}
|
||||
|
||||
func NewVM(ctx context.Context, opts *VMOpts) (*VM, error) {
|
||||
buf := bufbstore.NewBufferedBstore(opts.Bstore)
|
||||
buf := blockstore.NewBuffered(opts.Bstore)
|
||||
cst := cbor.NewCborStore(buf)
|
||||
state, err := state.LoadStateTree(cst, opts.StateBase)
|
||||
if err != nil {
|
||||
|
@ -29,7 +29,7 @@ import (
|
||||
init2 "github.com/filecoin-project/specs-actors/v2/actors/builtin/init"
|
||||
msig2 "github.com/filecoin-project/specs-actors/v2/actors/builtin/multisig"
|
||||
|
||||
"github.com/filecoin-project/lotus/api/apibstore"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/build"
|
||||
"github.com/filecoin-project/lotus/chain/actors/adt"
|
||||
"github.com/filecoin-project/lotus/chain/actors/builtin/multisig"
|
||||
@ -202,7 +202,7 @@ var msigInspectCmd = &cli.Command{
|
||||
defer closer()
|
||||
ctx := ReqContext(cctx)
|
||||
|
||||
store := adt.WrapStore(ctx, cbor.NewCborStore(apibstore.NewAPIBlockstore(api)))
|
||||
store := adt.WrapStore(ctx, cbor.NewCborStore(blockstore.NewAPIBlockstore(api)))
|
||||
|
||||
maddr, err := address.NewFromString(cctx.Args().First())
|
||||
if err != nil {
|
||||
|
@ -20,8 +20,8 @@ import (
|
||||
cbor "github.com/ipfs/go-ipld-cbor"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/filecoin-project/lotus/api/apibstore"
|
||||
"github.com/filecoin-project/lotus/api/test"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/build"
|
||||
"github.com/filecoin-project/lotus/chain/events"
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
@ -400,7 +400,7 @@ func getPaychState(ctx context.Context, t *testing.T, node test.TestNode, chAddr
|
||||
act, err := node.StateGetActor(ctx, chAddr, types.EmptyTSK)
|
||||
require.NoError(t, err)
|
||||
|
||||
store := cbor.NewCborStore(apibstore.NewAPIBlockstore(node))
|
||||
store := cbor.NewCborStore(blockstore.NewAPIBlockstore(node))
|
||||
chState, err := paych.Load(adt.WrapStore(ctx, store), act)
|
||||
require.NoError(t, err)
|
||||
|
||||
|
@ -34,7 +34,7 @@ import (
|
||||
|
||||
"github.com/filecoin-project/lotus/api"
|
||||
lapi "github.com/filecoin-project/lotus/api"
|
||||
"github.com/filecoin-project/lotus/api/apibstore"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/build"
|
||||
"github.com/filecoin-project/lotus/chain/state"
|
||||
"github.com/filecoin-project/lotus/chain/stmgr"
|
||||
@ -1010,7 +1010,7 @@ var stateComputeStateCmd = &cli.Command{
|
||||
}
|
||||
|
||||
if cctx.Bool("html") {
|
||||
st, err := state.LoadStateTree(cbor.NewCborStore(apibstore.NewAPIBlockstore(api)), stout.Root)
|
||||
st, err := state.LoadStateTree(cbor.NewCborStore(blockstore.NewAPIBlockstore(api)), stout.Root)
|
||||
if err != nil {
|
||||
return xerrors.Errorf("loading state tree: %w", err)
|
||||
}
|
||||
|
@ -25,13 +25,13 @@ import (
|
||||
"github.com/prometheus/client_golang/prometheus/promauto"
|
||||
|
||||
"github.com/filecoin-project/lotus/api"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
badgerbs "github.com/filecoin-project/lotus/blockstore/badger"
|
||||
"github.com/filecoin-project/lotus/chain/stmgr"
|
||||
"github.com/filecoin-project/lotus/chain/store"
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
"github.com/filecoin-project/lotus/chain/vm"
|
||||
lcli "github.com/filecoin-project/lotus/cli"
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
badgerbs "github.com/filecoin-project/lotus/lib/blockstore/badger"
|
||||
_ "github.com/filecoin-project/lotus/lib/sigs/bls"
|
||||
_ "github.com/filecoin-project/lotus/lib/sigs/secp"
|
||||
"github.com/filecoin-project/lotus/node/repo"
|
||||
@ -229,7 +229,7 @@ var importBenchCmd = &cli.Command{
|
||||
if ds != nil {
|
||||
ds = measure.New("dsbench", ds)
|
||||
defer ds.Close() //nolint:errcheck
|
||||
bs = blockstore.NewBlockstore(ds)
|
||||
bs = blockstore.FromDatastore(ds)
|
||||
}
|
||||
|
||||
if c, ok := bs.(io.Closer); ok {
|
||||
|
@ -15,7 +15,7 @@ import (
|
||||
"github.com/filecoin-project/go-state-types/big"
|
||||
|
||||
"github.com/filecoin-project/lotus/api"
|
||||
"github.com/filecoin-project/lotus/api/apibstore"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
|
||||
"github.com/filecoin-project/lotus/chain/actors/builtin/power"
|
||||
"github.com/filecoin-project/lotus/chain/events/state"
|
||||
@ -202,7 +202,7 @@ func (p *Processor) processMiners(ctx context.Context, minerTips map[types.TipSe
|
||||
log.Debugw("Processed Miners", "duration", time.Since(start).String())
|
||||
}()
|
||||
|
||||
stor := store.ActorStore(ctx, apibstore.NewAPIBlockstore(p.node))
|
||||
stor := store.ActorStore(ctx, blockstore.NewAPIBlockstore(p.node))
|
||||
|
||||
var out []minerActorInfo
|
||||
// TODO add parallel calls if this becomes slow
|
||||
@ -649,7 +649,7 @@ func (p *Processor) getMinerStateAt(ctx context.Context, maddr address.Address,
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return miner.Load(store.ActorStore(ctx, apibstore.NewAPIBlockstore(p.node)), prevActor)
|
||||
return miner.Load(store.ActorStore(ctx, blockstore.NewAPIBlockstore(p.node)), prevActor)
|
||||
}
|
||||
|
||||
func (p *Processor) getMinerPreCommitChanges(ctx context.Context, m minerActorInfo) (*miner.PreCommitChanges, error) {
|
||||
|
@ -9,10 +9,10 @@ import (
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/chain/vm"
|
||||
"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper"
|
||||
"github.com/filecoin-project/lotus/journal"
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
"github.com/filecoin-project/lotus/node/modules/testing"
|
||||
"github.com/google/uuid"
|
||||
"github.com/mitchellh/go-homedir"
|
||||
@ -327,7 +327,7 @@ var genesisCarCmd = &cli.Command{
|
||||
}
|
||||
ofile := c.String("out")
|
||||
jrnl := journal.NilJournal()
|
||||
bstor := blockstore.NewTemporarySync()
|
||||
bstor := blockstore.NewMemorySync()
|
||||
sbldr := vm.Syscalls(ffiwrapper.ProofVerifier)
|
||||
_, err := testing.MakeGenesis(ofile, c.Args().First())(bstor, sbldr, jrnl)()
|
||||
return err
|
||||
|
@ -17,6 +17,7 @@ import (
|
||||
"github.com/filecoin-project/go-address"
|
||||
"github.com/filecoin-project/go-state-types/big"
|
||||
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/build"
|
||||
"github.com/filecoin-project/lotus/chain/actors/adt"
|
||||
"github.com/filecoin-project/lotus/chain/actors/builtin/account"
|
||||
@ -26,7 +27,6 @@ import (
|
||||
"github.com/filecoin-project/lotus/chain/stmgr"
|
||||
"github.com/filecoin-project/lotus/chain/store"
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
)
|
||||
|
||||
type addrInfo struct {
|
||||
@ -50,7 +50,7 @@ var genesisVerifyCmd = &cli.Command{
|
||||
if !cctx.Args().Present() {
|
||||
return fmt.Errorf("must pass genesis car file")
|
||||
}
|
||||
bs := blockstore.NewBlockstore(datastore.NewMapDatastore())
|
||||
bs := blockstore.FromDatastore(datastore.NewMapDatastore())
|
||||
|
||||
cs := store.NewChainStore(bs, bs, datastore.NewMapDatastore(), nil, nil)
|
||||
defer cs.Close() //nolint:errcheck
|
||||
|
@ -11,10 +11,10 @@ import (
|
||||
"github.com/urfave/cli/v2"
|
||||
"golang.org/x/xerrors"
|
||||
|
||||
badgerbs "github.com/filecoin-project/lotus/blockstore/badger"
|
||||
"github.com/filecoin-project/lotus/chain/store"
|
||||
"github.com/filecoin-project/lotus/chain/vm"
|
||||
"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper"
|
||||
badgerbs "github.com/filecoin-project/lotus/lib/blockstore/badger"
|
||||
"github.com/filecoin-project/lotus/node/repo"
|
||||
)
|
||||
|
||||
|
@ -13,7 +13,7 @@ import (
|
||||
|
||||
verifreg2 "github.com/filecoin-project/specs-actors/v2/actors/builtin/verifreg"
|
||||
|
||||
"github.com/filecoin-project/lotus/api/apibstore"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/build"
|
||||
"github.com/filecoin-project/lotus/chain/actors"
|
||||
"github.com/filecoin-project/lotus/chain/actors/adt"
|
||||
@ -190,7 +190,7 @@ var verifRegListVerifiersCmd = &cli.Command{
|
||||
return err
|
||||
}
|
||||
|
||||
apibs := apibstore.NewAPIBlockstore(api)
|
||||
apibs := blockstore.NewAPIBlockstore(api)
|
||||
store := adt.WrapStore(ctx, cbor.NewCborStore(apibs))
|
||||
|
||||
st, err := verifreg.Load(store, act)
|
||||
@ -220,7 +220,7 @@ var verifRegListClientsCmd = &cli.Command{
|
||||
return err
|
||||
}
|
||||
|
||||
apibs := apibstore.NewAPIBlockstore(api)
|
||||
apibs := blockstore.NewAPIBlockstore(api)
|
||||
store := adt.WrapStore(ctx, cbor.NewCborStore(apibs))
|
||||
|
||||
st, err := verifreg.Load(store, act)
|
||||
@ -303,7 +303,7 @@ var verifRegCheckVerifierCmd = &cli.Command{
|
||||
return err
|
||||
}
|
||||
|
||||
apibs := apibstore.NewAPIBlockstore(api)
|
||||
apibs := blockstore.NewAPIBlockstore(api)
|
||||
store := adt.WrapStore(ctx, cbor.NewCborStore(apibs))
|
||||
|
||||
st, err := verifreg.Load(store, act)
|
||||
|
@ -19,7 +19,7 @@ import (
|
||||
|
||||
miner2 "github.com/filecoin-project/specs-actors/v2/actors/builtin/miner"
|
||||
|
||||
"github.com/filecoin-project/lotus/api/apibstore"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/build"
|
||||
"github.com/filecoin-project/lotus/chain/actors"
|
||||
"github.com/filecoin-project/lotus/chain/actors/adt"
|
||||
@ -307,7 +307,7 @@ var actorRepayDebtCmd = &cli.Command{
|
||||
return err
|
||||
}
|
||||
|
||||
store := adt.WrapStore(ctx, cbor.NewCborStore(apibstore.NewAPIBlockstore(api)))
|
||||
store := adt.WrapStore(ctx, cbor.NewCborStore(blockstore.NewAPIBlockstore(api)))
|
||||
|
||||
mst, err := miner.Load(store, mact)
|
||||
if err != nil {
|
||||
|
@ -18,14 +18,12 @@ import (
|
||||
sealing "github.com/filecoin-project/lotus/extern/storage-sealing"
|
||||
|
||||
"github.com/filecoin-project/lotus/api"
|
||||
"github.com/filecoin-project/lotus/api/apibstore"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/build"
|
||||
"github.com/filecoin-project/lotus/chain/actors/adt"
|
||||
"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
lcli "github.com/filecoin-project/lotus/cli"
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
"github.com/filecoin-project/lotus/lib/bufbstore"
|
||||
)
|
||||
|
||||
var infoCmd = &cli.Command{
|
||||
@ -102,7 +100,7 @@ func infoCmdAct(cctx *cli.Context) error {
|
||||
return err
|
||||
}
|
||||
|
||||
tbs := bufbstore.NewTieredBstore(apibstore.NewAPIBlockstore(api), blockstore.NewTemporary())
|
||||
tbs := blockstore.NewTieredBstore(blockstore.NewAPIBlockstore(api), blockstore.NewMemory())
|
||||
mas, err := miner.Load(adt.WrapStore(ctx, cbor.NewCborStore(tbs)), mact)
|
||||
if err != nil {
|
||||
return err
|
||||
|
@ -12,7 +12,7 @@ import (
|
||||
|
||||
"github.com/filecoin-project/go-address"
|
||||
"github.com/filecoin-project/go-state-types/abi"
|
||||
"github.com/filecoin-project/lotus/api/apibstore"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
|
||||
"github.com/filecoin-project/lotus/chain/store"
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
@ -52,7 +52,7 @@ var provingFaultsCmd = &cli.Command{
|
||||
|
||||
ctx := lcli.ReqContext(cctx)
|
||||
|
||||
stor := store.ActorStore(ctx, apibstore.NewAPIBlockstore(api))
|
||||
stor := store.ActorStore(ctx, blockstore.NewAPIBlockstore(api))
|
||||
|
||||
maddr, err := getActorAddress(ctx, nodeApi, cctx.String("actor"))
|
||||
if err != nil {
|
||||
@ -127,7 +127,7 @@ var provingInfoCmd = &cli.Command{
|
||||
return err
|
||||
}
|
||||
|
||||
stor := store.ActorStore(ctx, apibstore.NewAPIBlockstore(api))
|
||||
stor := store.ActorStore(ctx, blockstore.NewAPIBlockstore(api))
|
||||
|
||||
mas, err := miner.Load(stor, mact)
|
||||
if err != nil {
|
||||
|
@ -15,8 +15,8 @@ import (
|
||||
"github.com/libp2p/go-libp2p-core/peer"
|
||||
pubsub "github.com/libp2p/go-libp2p-pubsub"
|
||||
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/build"
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
)
|
||||
|
||||
var topic = "/fil/headnotifs/"
|
||||
@ -28,7 +28,7 @@ func init() {
|
||||
return
|
||||
}
|
||||
|
||||
bs := blockstore.NewTemporary()
|
||||
bs := blockstore.NewMemory()
|
||||
|
||||
c, err := car.LoadCar(bs, bytes.NewReader(genBytes))
|
||||
if err != nil {
|
||||
|
@ -17,10 +17,10 @@ import (
|
||||
|
||||
"github.com/filecoin-project/test-vectors/schema"
|
||||
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/chain/state"
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
"github.com/filecoin-project/lotus/conformance"
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
)
|
||||
|
||||
var execFlags struct {
|
||||
|
@ -9,7 +9,7 @@ import (
|
||||
dssync "github.com/ipfs/go-datastore/sync"
|
||||
|
||||
"github.com/filecoin-project/lotus/api"
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
|
||||
"github.com/filecoin-project/lotus/chain/actors/adt"
|
||||
|
||||
@ -45,7 +45,7 @@ func NewProxyingStores(ctx context.Context, api api.FullNode) *Stores {
|
||||
bs := &proxyingBlockstore{
|
||||
ctx: ctx,
|
||||
api: api,
|
||||
Blockstore: blockstore.NewBlockstore(ds),
|
||||
Blockstore: blockstore.FromDatastore(ds),
|
||||
}
|
||||
return NewStores(ctx, ds, bs)
|
||||
}
|
||||
|
@ -5,6 +5,7 @@ import (
|
||||
gobig "math/big"
|
||||
"os"
|
||||
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/chain/state"
|
||||
"github.com/filecoin-project/lotus/chain/stmgr"
|
||||
"github.com/filecoin-project/lotus/chain/store"
|
||||
@ -12,7 +13,6 @@ import (
|
||||
"github.com/filecoin-project/lotus/chain/vm"
|
||||
"github.com/filecoin-project/lotus/conformance/chaos"
|
||||
"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper"
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
|
||||
_ "github.com/filecoin-project/lotus/lib/sigs/bls" // enable bls signatures
|
||||
_ "github.com/filecoin-project/lotus/lib/sigs/secp" // enable secp signatures
|
||||
|
@ -26,9 +26,9 @@ import (
|
||||
|
||||
"github.com/filecoin-project/test-vectors/schema"
|
||||
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
"github.com/filecoin-project/lotus/chain/vm"
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
)
|
||||
|
||||
// FallbackBlockstoreGetter is a fallback blockstore to use for resolving CIDs
|
||||
@ -306,7 +306,7 @@ func writeStateToTempCAR(bs blockstore.Blockstore, roots ...cid.Cid) (string, er
|
||||
}
|
||||
|
||||
func LoadBlockstore(vectorCAR schema.Base64EncodedBytes) (blockstore.Blockstore, error) {
|
||||
bs := blockstore.Blockstore(blockstore.NewTemporary())
|
||||
bs := blockstore.Blockstore(blockstore.NewMemory())
|
||||
|
||||
// Read the base64-encoded CAR from the vector, and inflate the gzip.
|
||||
buf := bytes.NewReader(vectorCAR)
|
||||
|
@ -1,66 +0,0 @@
|
||||
// blockstore contains all the basic blockstore constructors used by lotus. Any
|
||||
// blockstores not ultimately constructed out of the building blocks in this
|
||||
// package may not work properly.
|
||||
//
|
||||
// * This package correctly wraps blockstores with the IdBlockstore. This blockstore:
|
||||
// * Filters out all puts for blocks with CIDs using the "identity" hash function.
|
||||
// * Extracts inlined blocks from CIDs using the identity hash function and
|
||||
// returns them on get/has, ignoring the contents of the blockstore.
|
||||
// * In the future, this package may enforce additional restrictions on block
|
||||
// sizes, CID validity, etc.
|
||||
//
|
||||
// To make auditing for misuse of blockstores tractable, this package re-exports
|
||||
// parts of the go-ipfs-blockstore package such that no other package needs to
|
||||
// import it directly.
|
||||
package blockstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
ds "github.com/ipfs/go-datastore"
|
||||
|
||||
blockstore "github.com/ipfs/go-ipfs-blockstore"
|
||||
)
|
||||
|
||||
// NewTemporary returns a temporary blockstore.
|
||||
func NewTemporary() MemStore {
|
||||
return make(MemStore)
|
||||
}
|
||||
|
||||
// NewTemporarySync returns a thread-safe temporary blockstore.
|
||||
func NewTemporarySync() *SyncStore {
|
||||
return &SyncStore{bs: make(MemStore)}
|
||||
}
|
||||
|
||||
// WrapIDStore wraps the underlying blockstore in an "identity" blockstore.
|
||||
func WrapIDStore(bstore blockstore.Blockstore) blockstore.Blockstore {
|
||||
return blockstore.NewIdStore(bstore)
|
||||
}
|
||||
|
||||
// NewBlockstore creates a new blockstore wrapped by the given datastore.
|
||||
func NewBlockstore(dstore ds.Batching) blockstore.Blockstore {
|
||||
return WrapIDStore(blockstore.NewBlockstore(dstore))
|
||||
}
|
||||
|
||||
// Alias so other packages don't have to import go-ipfs-blockstore
|
||||
type Blockstore = blockstore.Blockstore
|
||||
type Viewer = blockstore.Viewer
|
||||
type CacheOpts = blockstore.CacheOpts
|
||||
|
||||
var ErrNotFound = blockstore.ErrNotFound
|
||||
|
||||
func DefaultCacheOpts() CacheOpts {
|
||||
return CacheOpts{
|
||||
HasBloomFilterSize: 0,
|
||||
HasBloomFilterHashes: 0,
|
||||
HasARCCacheSize: 512 << 10,
|
||||
}
|
||||
}
|
||||
|
||||
func CachedBlockstore(ctx context.Context, bs Blockstore, opts CacheOpts) (Blockstore, error) {
|
||||
bs, err := blockstore.CachedBlockstore(ctx, bs, opts)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return WrapIDStore(bs), nil
|
||||
}
|
@ -1,79 +0,0 @@
|
||||
package blockstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
|
||||
blocks "github.com/ipfs/go-block-format"
|
||||
"github.com/ipfs/go-cid"
|
||||
)
|
||||
|
||||
// SyncStore is a terminal blockstore that is a synchronized version
|
||||
// of MemStore.
|
||||
type SyncStore struct {
|
||||
mu sync.RWMutex
|
||||
bs MemStore // specifically use a memStore to save indirection overhead.
|
||||
}
|
||||
|
||||
func (m *SyncStore) DeleteBlock(k cid.Cid) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
return m.bs.DeleteBlock(k)
|
||||
}
|
||||
|
||||
func (m *SyncStore) Has(k cid.Cid) (bool, error) {
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
return m.bs.Has(k)
|
||||
}
|
||||
|
||||
func (m *SyncStore) View(k cid.Cid, callback func([]byte) error) error {
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
|
||||
return m.bs.View(k, callback)
|
||||
}
|
||||
|
||||
func (m *SyncStore) Get(k cid.Cid) (blocks.Block, error) {
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
return m.bs.Get(k)
|
||||
}
|
||||
|
||||
// GetSize returns the CIDs mapped BlockSize
|
||||
func (m *SyncStore) GetSize(k cid.Cid) (int, error) {
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
return m.bs.GetSize(k)
|
||||
}
|
||||
|
||||
// Put puts a given block to the underlying datastore
|
||||
func (m *SyncStore) Put(b blocks.Block) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
return m.bs.Put(b)
|
||||
}
|
||||
|
||||
// PutMany puts a slice of blocks at the same time using batching
|
||||
// capabilities of the underlying datastore whenever possible.
|
||||
func (m *SyncStore) PutMany(bs []blocks.Block) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
return m.bs.PutMany(bs)
|
||||
}
|
||||
|
||||
// AllKeysChan returns a channel from which
|
||||
// the CIDs in the Blockstore can be read. It should respect
|
||||
// the given context, closing the channel if it becomes Done.
|
||||
func (m *SyncStore) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) {
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
// this blockstore implementation doesn't do any async work.
|
||||
return m.bs.AllKeysChan(ctx)
|
||||
}
|
||||
|
||||
// HashOnRead specifies if every read block should be
|
||||
// rehashed to make sure it matches its CID.
|
||||
func (m *SyncStore) HashOnRead(enabled bool) {
|
||||
// noop
|
||||
}
|
@ -1,186 +0,0 @@
|
||||
package bufbstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
|
||||
block "github.com/ipfs/go-block-format"
|
||||
"github.com/ipfs/go-cid"
|
||||
logging "github.com/ipfs/go-log/v2"
|
||||
|
||||
bstore "github.com/filecoin-project/lotus/lib/blockstore"
|
||||
)
|
||||
|
||||
var log = logging.Logger("bufbs")
|
||||
|
||||
type BufferedBS struct {
|
||||
read bstore.Blockstore
|
||||
write bstore.Blockstore
|
||||
|
||||
readviewer bstore.Viewer
|
||||
writeviewer bstore.Viewer
|
||||
}
|
||||
|
||||
func NewBufferedBstore(base bstore.Blockstore) *BufferedBS {
|
||||
var buf bstore.Blockstore
|
||||
if os.Getenv("LOTUS_DISABLE_VM_BUF") == "iknowitsabadidea" {
|
||||
log.Warn("VM BLOCKSTORE BUFFERING IS DISABLED")
|
||||
buf = base
|
||||
} else {
|
||||
buf = bstore.NewTemporary()
|
||||
}
|
||||
|
||||
bs := &BufferedBS{
|
||||
read: base,
|
||||
write: buf,
|
||||
}
|
||||
if v, ok := base.(bstore.Viewer); ok {
|
||||
bs.readviewer = v
|
||||
}
|
||||
if v, ok := buf.(bstore.Viewer); ok {
|
||||
bs.writeviewer = v
|
||||
}
|
||||
if (bs.writeviewer == nil) != (bs.readviewer == nil) {
|
||||
log.Warnf("one of the stores is not viewable; running less efficiently")
|
||||
}
|
||||
return bs
|
||||
}
|
||||
|
||||
func NewTieredBstore(r bstore.Blockstore, w bstore.Blockstore) *BufferedBS {
|
||||
return &BufferedBS{
|
||||
read: r,
|
||||
write: w,
|
||||
}
|
||||
}
|
||||
|
||||
var _ bstore.Blockstore = (*BufferedBS)(nil)
|
||||
var _ bstore.Viewer = (*BufferedBS)(nil)
|
||||
|
||||
func (bs *BufferedBS) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) {
|
||||
a, err := bs.read.AllKeysChan(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
b, err := bs.write.AllKeysChan(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
out := make(chan cid.Cid)
|
||||
go func() {
|
||||
defer close(out)
|
||||
for a != nil || b != nil {
|
||||
select {
|
||||
case val, ok := <-a:
|
||||
if !ok {
|
||||
a = nil
|
||||
} else {
|
||||
select {
|
||||
case out <- val:
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
case val, ok := <-b:
|
||||
if !ok {
|
||||
b = nil
|
||||
} else {
|
||||
select {
|
||||
case out <- val:
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (bs *BufferedBS) DeleteBlock(c cid.Cid) error {
|
||||
if err := bs.read.DeleteBlock(c); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return bs.write.DeleteBlock(c)
|
||||
}
|
||||
|
||||
func (bs *BufferedBS) View(c cid.Cid, callback func([]byte) error) error {
|
||||
if bs.writeviewer == nil || bs.readviewer == nil {
|
||||
// one of the stores isn't Viewer; fall back to pure Get behaviour.
|
||||
blk, err := bs.Get(c)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return callback(blk.RawData())
|
||||
}
|
||||
|
||||
// both stores are viewable.
|
||||
if err := bs.writeviewer.View(c, callback); err == bstore.ErrNotFound {
|
||||
// not found in write blockstore; fall through.
|
||||
} else {
|
||||
return err // propagate errors, or nil, i.e. found.
|
||||
}
|
||||
return bs.readviewer.View(c, callback)
|
||||
}
|
||||
|
||||
func (bs *BufferedBS) Get(c cid.Cid) (block.Block, error) {
|
||||
if out, err := bs.write.Get(c); err != nil {
|
||||
if err != bstore.ErrNotFound {
|
||||
return nil, err
|
||||
}
|
||||
} else {
|
||||
return out, nil
|
||||
}
|
||||
|
||||
return bs.read.Get(c)
|
||||
}
|
||||
|
||||
func (bs *BufferedBS) GetSize(c cid.Cid) (int, error) {
|
||||
s, err := bs.read.GetSize(c)
|
||||
if err == bstore.ErrNotFound || s == 0 {
|
||||
return bs.write.GetSize(c)
|
||||
}
|
||||
|
||||
return s, err
|
||||
}
|
||||
|
||||
func (bs *BufferedBS) Put(blk block.Block) error {
|
||||
has, err := bs.read.Has(blk.Cid()) // TODO: consider dropping this check
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if has {
|
||||
return nil
|
||||
}
|
||||
|
||||
return bs.write.Put(blk)
|
||||
}
|
||||
|
||||
func (bs *BufferedBS) Has(c cid.Cid) (bool, error) {
|
||||
has, err := bs.write.Has(c)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
if has {
|
||||
return true, nil
|
||||
}
|
||||
|
||||
return bs.read.Has(c)
|
||||
}
|
||||
|
||||
func (bs *BufferedBS) HashOnRead(hor bool) {
|
||||
bs.read.HashOnRead(hor)
|
||||
bs.write.HashOnRead(hor)
|
||||
}
|
||||
|
||||
func (bs *BufferedBS) PutMany(blks []block.Block) error {
|
||||
return bs.write.PutMany(blks)
|
||||
}
|
||||
|
||||
func (bs *BufferedBS) Read() bstore.Blockstore {
|
||||
return bs.read
|
||||
}
|
@ -10,7 +10,7 @@ import (
|
||||
"github.com/filecoin-project/go-address"
|
||||
"github.com/filecoin-project/lotus/chain/actors/adt"
|
||||
|
||||
"github.com/filecoin-project/lotus/api/apibstore"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
)
|
||||
@ -24,7 +24,7 @@ type apiWrapper struct {
|
||||
}
|
||||
|
||||
func (ca *apiWrapper) diffPreCommits(ctx context.Context, actor address.Address, pre, cur types.TipSetKey) (*miner.PreCommitChanges, error) {
|
||||
store := adt.WrapStore(ctx, cbor.NewCborStore(apibstore.NewAPIBlockstore(ca.api)))
|
||||
store := adt.WrapStore(ctx, cbor.NewCborStore(blockstore.NewAPIBlockstore(ca.api)))
|
||||
|
||||
preAct, err := ca.api.StateGetActor(ctx, actor, pre)
|
||||
if err != nil {
|
||||
|
@ -14,8 +14,8 @@ import (
|
||||
|
||||
"github.com/filecoin-project/go-address"
|
||||
"github.com/filecoin-project/go-state-types/abi"
|
||||
bstore "github.com/filecoin-project/lotus/blockstore"
|
||||
test "github.com/filecoin-project/lotus/chain/events/state/mock"
|
||||
bstore "github.com/filecoin-project/lotus/lib/blockstore"
|
||||
builtin2 "github.com/filecoin-project/specs-actors/v2/actors/builtin"
|
||||
|
||||
market2 "github.com/filecoin-project/specs-actors/v2/actors/builtin/market"
|
||||
@ -28,7 +28,7 @@ import (
|
||||
|
||||
func TestDealStateMatcher(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
bs := bstore.NewTemporarySync()
|
||||
bs := bstore.NewMemorySync()
|
||||
store := adt2.WrapStore(ctx, cbornode.NewCborStore(bs))
|
||||
|
||||
deal1 := &market2.DealState{
|
||||
|
@ -31,10 +31,10 @@ import (
|
||||
"github.com/filecoin-project/specs-actors/actors/util/adt"
|
||||
|
||||
"github.com/filecoin-project/lotus/api"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/chain/store"
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
"github.com/filecoin-project/lotus/chain/vm"
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
)
|
||||
|
||||
var log = logging.Logger("fullnode")
|
||||
|
@ -20,6 +20,7 @@ import (
|
||||
"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper"
|
||||
"github.com/filecoin-project/lotus/journal"
|
||||
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/build"
|
||||
"github.com/filecoin-project/lotus/chain"
|
||||
"github.com/filecoin-project/lotus/chain/beacon"
|
||||
@ -30,9 +31,6 @@ import (
|
||||
"github.com/filecoin-project/lotus/chain/store"
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
"github.com/filecoin-project/lotus/chain/vm"
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
"github.com/filecoin-project/lotus/lib/bufbstore"
|
||||
"github.com/filecoin-project/lotus/lib/timedbs"
|
||||
"github.com/filecoin-project/lotus/node/modules/dtypes"
|
||||
"github.com/filecoin-project/lotus/node/modules/helpers"
|
||||
"github.com/filecoin-project/lotus/node/repo"
|
||||
@ -46,10 +44,10 @@ func ChainBitswap(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host, rt r
|
||||
|
||||
// Write all incoming bitswap blocks into a temporary blockstore for two
|
||||
// block times. If they validate, they'll be persisted later.
|
||||
cache := timedbs.NewTimedCacheBS(2 * time.Duration(build.BlockDelaySecs) * time.Second)
|
||||
cache := blockstore.NewTimedCacheBlockstore(2 * time.Duration(build.BlockDelaySecs) * time.Second)
|
||||
lc.Append(fx.Hook{OnStop: cache.Stop, OnStart: cache.Start})
|
||||
|
||||
bitswapBs := bufbstore.NewTieredBstore(bs, cache)
|
||||
bitswapBs := blockstore.NewTieredBstore(bs, cache)
|
||||
|
||||
// Use just exch.Close(), closing the context is not needed
|
||||
exch := bitswap.New(mctx, bitswapNetwork, bitswapBs, bitswapOptions...)
|
||||
|
@ -30,9 +30,9 @@ import (
|
||||
"github.com/ipfs/go-datastore/namespace"
|
||||
"github.com/libp2p/go-libp2p-core/host"
|
||||
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/chain/market"
|
||||
"github.com/filecoin-project/lotus/journal"
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
"github.com/filecoin-project/lotus/markets"
|
||||
marketevents "github.com/filecoin-project/lotus/markets/loggers"
|
||||
"github.com/filecoin-project/lotus/markets/retrievaladapter"
|
||||
|
@ -14,7 +14,7 @@ import (
|
||||
"github.com/filecoin-project/go-fil-markets/piecestore"
|
||||
"github.com/filecoin-project/go-statestore"
|
||||
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/node/repo/importmgr"
|
||||
"github.com/filecoin-project/lotus/node/repo/retrievalstoremgr"
|
||||
)
|
||||
|
@ -6,8 +6,7 @@ import (
|
||||
|
||||
"github.com/multiformats/go-multiaddr"
|
||||
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
"github.com/filecoin-project/lotus/lib/ipfsbstore"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/node/modules/dtypes"
|
||||
"github.com/filecoin-project/lotus/node/modules/helpers"
|
||||
)
|
||||
@ -26,9 +25,9 @@ func IpfsClientBlockstore(ipfsMaddr string, onlineMode bool) func(helpers.Metric
|
||||
if err != nil {
|
||||
return nil, xerrors.Errorf("parsing ipfs multiaddr: %w", err)
|
||||
}
|
||||
ipfsbs, err = ipfsbstore.NewRemoteIpfsBstore(helpers.LifecycleCtx(mctx, lc), ma, onlineMode)
|
||||
ipfsbs, err = blockstore.NewRemoteIPFSBlockstore(helpers.LifecycleCtx(mctx, lc), ma, onlineMode)
|
||||
} else {
|
||||
ipfsbs, err = ipfsbstore.NewIpfsBstore(helpers.LifecycleCtx(mctx, lc), onlineMode)
|
||||
ipfsbs, err = blockstore.NewLocalIPFSBlockstore(helpers.LifecycleCtx(mctx, lc), onlineMode)
|
||||
}
|
||||
if err != nil {
|
||||
return nil, xerrors.Errorf("constructing ipfs blockstore: %w", err)
|
||||
|
@ -7,7 +7,7 @@ import (
|
||||
|
||||
"github.com/filecoin-project/go-address"
|
||||
"github.com/filecoin-project/lotus/api"
|
||||
"github.com/filecoin-project/lotus/api/apibstore"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/chain/actors/adt"
|
||||
"github.com/filecoin-project/lotus/chain/actors/builtin/paych"
|
||||
"github.com/filecoin-project/lotus/chain/stmgr"
|
||||
@ -21,7 +21,7 @@ type RPCStateManager struct {
|
||||
}
|
||||
|
||||
func NewRPCStateManager(api api.GatewayAPI) *RPCStateManager {
|
||||
cstore := cbor.NewCborStore(apibstore.NewAPIBlockstore(api))
|
||||
cstore := cbor.NewCborStore(blockstore.NewAPIBlockstore(api))
|
||||
return &RPCStateManager{gapi: api, cstore: cstore}
|
||||
}
|
||||
|
||||
|
@ -56,6 +56,7 @@ import (
|
||||
"github.com/filecoin-project/lotus/extern/storage-sealing/sealiface"
|
||||
|
||||
lapi "github.com/filecoin-project/lotus/api"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/build"
|
||||
"github.com/filecoin-project/lotus/chain/actors/builtin"
|
||||
"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
|
||||
@ -63,7 +64,6 @@ import (
|
||||
"github.com/filecoin-project/lotus/chain/gen/slashfilter"
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
"github.com/filecoin-project/lotus/journal"
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
"github.com/filecoin-project/lotus/markets"
|
||||
marketevents "github.com/filecoin-project/lotus/markets/loggers"
|
||||
"github.com/filecoin-project/lotus/markets/retrievaladapter"
|
||||
@ -392,7 +392,7 @@ func StagingBlockstore(lc fx.Lifecycle, mctx helpers.MetricsCtx, r repo.LockedRe
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return blockstore.NewBlockstore(stagingds), nil
|
||||
return blockstore.FromDatastore(stagingds), nil
|
||||
}
|
||||
|
||||
// StagingDAG is a DAGService for the StagingBlockstore
|
||||
|
@ -1,6 +1,6 @@
|
||||
package repo
|
||||
|
||||
import badgerbs "github.com/filecoin-project/lotus/lib/blockstore/badger"
|
||||
import badgerbs "github.com/filecoin-project/lotus/blockstore/badger"
|
||||
|
||||
// BadgerBlockstoreOptions returns the badger options to apply for the provided
|
||||
// domain.
|
||||
|
@ -13,7 +13,7 @@ import (
|
||||
"sync"
|
||||
|
||||
"github.com/BurntSushi/toml"
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/ipfs/go-datastore"
|
||||
fslock "github.com/ipfs/go-fs-lock"
|
||||
logging "github.com/ipfs/go-log/v2"
|
||||
@ -22,10 +22,10 @@ import (
|
||||
"github.com/multiformats/go-multiaddr"
|
||||
"golang.org/x/xerrors"
|
||||
|
||||
lblockstore "github.com/filecoin-project/lotus/blockstore"
|
||||
badgerbs "github.com/filecoin-project/lotus/blockstore/badger"
|
||||
"github.com/filecoin-project/lotus/extern/sector-storage/fsutil"
|
||||
"github.com/filecoin-project/lotus/extern/sector-storage/stores"
|
||||
lblockstore "github.com/filecoin-project/lotus/lib/blockstore"
|
||||
badgerbs "github.com/filecoin-project/lotus/lib/blockstore/badger"
|
||||
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
"github.com/filecoin-project/lotus/node/config"
|
||||
|
@ -7,7 +7,7 @@ import (
|
||||
"golang.org/x/xerrors"
|
||||
|
||||
"github.com/filecoin-project/go-multistore"
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/ipfs/go-datastore"
|
||||
"github.com/ipfs/go-datastore/namespace"
|
||||
)
|
||||
@ -31,7 +31,7 @@ const (
|
||||
func New(mds *multistore.MultiStore, ds datastore.Batching) *Mgr {
|
||||
return &Mgr{
|
||||
mds: mds,
|
||||
Blockstore: mds.MultiReadBlockstore(),
|
||||
Blockstore: blockstore.Adapt(mds.MultiReadBlockstore()),
|
||||
|
||||
ds: datastore.NewLogDatastore(namespace.Wrap(ds, datastore.NewKey("/stores")), "storess"),
|
||||
}
|
||||
|
@ -4,7 +4,7 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/ipfs/go-datastore"
|
||||
"github.com/multiformats/go-multiaddr"
|
||||
|
||||
|
@ -15,10 +15,10 @@ import (
|
||||
"github.com/multiformats/go-multiaddr"
|
||||
"golang.org/x/xerrors"
|
||||
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/chain/types"
|
||||
"github.com/filecoin-project/lotus/extern/sector-storage/fsutil"
|
||||
"github.com/filecoin-project/lotus/extern/sector-storage/stores"
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
"github.com/filecoin-project/lotus/node/config"
|
||||
)
|
||||
|
||||
@ -161,7 +161,7 @@ func NewMemory(opts *MemRepoOptions) *MemRepo {
|
||||
|
||||
return &MemRepo{
|
||||
repoLock: make(chan struct{}, 1),
|
||||
blockstore: blockstore.WrapIDStore(blockstore.NewTemporarySync()),
|
||||
blockstore: blockstore.WrapIDStore(blockstore.NewMemorySync()),
|
||||
datastore: opts.Ds,
|
||||
configF: opts.ConfigF,
|
||||
keystore: opts.KeyStore,
|
||||
|
@ -4,7 +4,7 @@ import (
|
||||
"errors"
|
||||
|
||||
"github.com/filecoin-project/go-multistore"
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/node/repo/importmgr"
|
||||
"github.com/ipfs/go-blockservice"
|
||||
offline "github.com/ipfs/go-ipfs-exchange-offline"
|
||||
|
@ -15,7 +15,7 @@ import (
|
||||
|
||||
"github.com/filecoin-project/go-multistore"
|
||||
|
||||
"github.com/filecoin-project/lotus/lib/blockstore"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/node/repo/importmgr"
|
||||
"github.com/filecoin-project/lotus/node/repo/retrievalstoremgr"
|
||||
)
|
||||
@ -71,7 +71,7 @@ func TestMultistoreRetrievalStoreManager(t *testing.T) {
|
||||
func TestBlockstoreRetrievalStoreManager(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
ds := dss.MutexWrap(datastore.NewMapDatastore())
|
||||
bs := blockstore.NewBlockstore(ds)
|
||||
bs := blockstore.FromDatastore(ds)
|
||||
retrievalStoreMgr := retrievalstoremgr.NewBlockstoreRetrievalStoreManager(bs)
|
||||
var stores []retrievalstoremgr.RetrievalStore
|
||||
var cids []cid.Cid
|
||||
|
@ -18,7 +18,7 @@ import (
|
||||
market2 "github.com/filecoin-project/specs-actors/v2/actors/builtin/market"
|
||||
|
||||
"github.com/filecoin-project/lotus/api"
|
||||
"github.com/filecoin-project/lotus/api/apibstore"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/build"
|
||||
"github.com/filecoin-project/lotus/chain/actors"
|
||||
"github.com/filecoin-project/lotus/chain/actors/builtin/market"
|
||||
@ -188,7 +188,7 @@ func (s SealingAPIAdapter) StateSectorPreCommitInfo(ctx context.Context, maddr a
|
||||
return nil, xerrors.Errorf("handleSealFailed(%d): temp error: %+v", sectorNumber, err)
|
||||
}
|
||||
|
||||
stor := store.ActorStore(ctx, apibstore.NewAPIBlockstore(s.delegate))
|
||||
stor := store.ActorStore(ctx, blockstore.NewAPIBlockstore(s.delegate))
|
||||
|
||||
state, err := miner.Load(stor, act)
|
||||
if err != nil {
|
||||
|
@ -14,7 +14,7 @@ import (
|
||||
|
||||
"github.com/filecoin-project/go-address"
|
||||
"github.com/filecoin-project/go-state-types/big"
|
||||
"github.com/filecoin-project/lotus/api/apibstore"
|
||||
"github.com/filecoin-project/lotus/blockstore"
|
||||
"github.com/filecoin-project/lotus/build"
|
||||
|
||||
"github.com/filecoin-project/lotus/api"
|
||||
@ -699,7 +699,7 @@ func info(t *testkit.TestEnvironment, m *testkit.LotusMiner, maddr address.Addre
|
||||
i.FaultyBytes = types.BigMul(types.NewInt(nfaults), types.NewInt(uint64(mi.SectorSize)))
|
||||
}
|
||||
|
||||
stor := store.ActorStore(ctx, apibstore.NewAPIBlockstore(api))
|
||||
stor := store.ActorStore(ctx, blockstore.NewAPIBlockstore(api))
|
||||
mas, err := miner.Load(stor, mact)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
Loading…
Reference in New Issue
Block a user