cerc-io/lotus
11
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 27 Wiki Activity

replace badger in storage client

Browse Source
This commit is contained in:
aarshkshah1992
2021-07-15 14:26:02 +05:30
parent 872843e491
commit 46b78406b1
8 changed files with 66 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/builder_miner.go
-4
View File
@@ -7,8 +7,6 @@ import (
"go.uber.org/fx"
"golang.org/x/xerrors"
"github.com/filecoin-project/dagstore"
"github.com/filecoin-project/dagstore/mount"
mktdagstore "github.com/filecoin-project/go-fil-markets/dagstore"
"github.com/filecoin-project/go-fil-markets/retrievalmarket"
rmnet "github.com/filecoin-project/go-fil-markets/retrievalmarket/network"
@@ -149,8 +147,6 @@ func ConfigStorageMiner(c interface{}) Option {
Override(new(dtypes.RetrievalPricingFunc), modules.RetrievalPricingFunc(cfg.Dealmaking)),
// DAG Store
Override(new(*mount.Registry), modules.DAGStoreRegistry),
Override(new(*dagstore.DAGStore), modules.DAGStore),
Override(new(mktdagstore.DagStoreWrapper), modules.DAGStoreWrapper),
// Markets (retrieval)
node/impl/client/client.go
+29 -28
View File
@@ -10,6 +10,7 @@ import (
"sort"
"time"
"github.com/filecoin-project/go-fil-markets/filestorecaradapter"
"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
carv2 "github.com/ipld/go-car/v2"
"github.com/ipld/go-car/v2/blockstore"
@@ -124,14 +125,14 @@ func (a *API) dealStarter(ctx context.Context, params *api.StartDealParams, isSt
return nil, xerrors.New("stateless storage deals can only be initiated with storage price of 0")
}
} else if params.Data.TransferType == storagemarket.TTGraphsync {
c, err := a.imgr().CARV2FilePathFor(params.Data.Root)
fc, err := a.imgr().FilestoreCARV2FilePathFor(params.Data.Root)
if err != nil {
return nil, xerrors.Errorf("failed to find CARv2 file path: %w", err)
}
if c == "" {
if fc == "" {
return nil, xerrors.New("no CARv2 file path for deal")
}
CARV2FilePath = c
CARV2FilePath = fc
}
walletKey, err := a.StateAccountKey(ctx, params.Wallet, types.EmptyTSK)
@@ -187,17 +188,17 @@ func (a *API) dealStarter(ctx context.Context, params *api.StartDealParams, isSt
providerInfo := utils.NewStorageProviderInfo(params.Miner, mi.Worker, mi.SectorSize, *mi.PeerId, mi.Multiaddrs)
result, err := a.SMDealClient.ProposeStorageDeal(ctx, storagemarket.ProposeStorageDealParams{
Addr: params.Wallet,
Info: &providerInfo,
Data: params.Data,
StartEpoch: dealStart,
EndEpoch: calcDealExpiration(params.MinBlocksDuration, md, dealStart),
Price: params.EpochPrice,
Collateral: params.ProviderCollateral,
Rt: st,
FastRetrieval: params.FastRetrieval,
VerifiedDeal: params.VerifiedDeal,
CARV2FilePath: CARV2FilePath,
Addr: params.Wallet,
Info: &providerInfo,
Data: params.Data,
StartEpoch: dealStart,
EndEpoch: calcDealExpiration(params.MinBlocksDuration, md, dealStart),
Price: params.EpochPrice,
Collateral: params.ProviderCollateral,
Rt: st,
FastRetrieval: params.FastRetrieval,
VerifiedDeal: params.VerifiedDeal,
FilestoreCARv2FilePath: CARV2FilePath,
})
if err != nil {
@@ -398,11 +399,11 @@ func (a *API) newDealInfoWithTransfer(transferCh *api.DataTransferChannel, v sto
func (a *API) ClientHasLocal(ctx context.Context, root cid.Cid) (bool, error) {
// TODO: check if we have the ENTIRE dag
carv2Path, err := a.imgr().CARV2FilePathFor(root)
fc, err := a.imgr().FilestoreCARV2FilePathFor(root)
if err != nil {
return false, err
}
if len(carv2Path) == 0 {
if len(fc) == 0 {
return false, nil
}
return true, nil
@@ -490,7 +491,7 @@ func (a *API) ClientImport(ctx context.Context, ref api.FileRef) (res *api.Impor
return nil, xerrors.Errorf("failed to import CAR file: %w", err)
}
} else {
root, err = a.importNormalFileToCARv2(ctx, id, ref.Path, carV2File)
root, err = a.importNormalFileToFilestoreCARv2(ctx, id, ref.Path, carV2File)
if err != nil {
return nil, xerrors.Errorf("failed to import normal file to CARv2: %w", err)
}
@@ -502,7 +503,7 @@ func (a *API) ClientImport(ctx context.Context, ref api.FileRef) (res *api.Impor
if err := a.imgr().AddLabel(id, importmgr.LFileName, ref.Path); err != nil {
return nil, err
}
if err := a.imgr().AddLabel(id, importmgr.LCARv2FilePath, carV2File); err != nil {
if err := a.imgr().AddLabel(id, importmgr.LFileStoreCARv2FilePath, carV2File); err != nil {
return nil, err
}
if err := a.imgr().AddLabel(id, importmgr.LRootCid, root.String()); err != nil {
@@ -522,7 +523,7 @@ func (a *API) ClientRemoveImport(ctx context.Context, importID importmgr.ImportI
}
// remove the CARv2 file if we've created one.
if path := info.Labels[importmgr.LCARv2FilePath]; path != "" {
if path := info.Labels[importmgr.LFileStoreCARv2FilePath]; path != "" {
_ = os.Remove(path)
}
@@ -579,7 +580,7 @@ func (a *API) ClientListImports(ctx context.Context) ([]api.Import, error) {
Key: id,
Source: info.Labels[importmgr.LSource],
FilePath: info.Labels[importmgr.LFileName],
CARv2FilePath: info.Labels[importmgr.LCARv2FilePath],
CARv2FilePath: info.Labels[importmgr.LFileStoreCARv2FilePath],
}
if info.Labels[importmgr.LRootCid] != "" {
@@ -999,17 +1000,17 @@ func (w *lenWriter) Write(p []byte) (n int, err error) {
}
func (a *API) ClientDealSize(ctx context.Context, root cid.Cid) (api.DataSize, error) {
carv2FilePath, err := a.imgr().CARV2FilePathFor(root)
fc, err := a.imgr().FilestoreCARV2FilePathFor(root)
if err != nil {
return api.DataSize{}, xerrors.Errorf("failed to find CARv2 file for root: %w", err)
}
if len(carv2FilePath) == 0 {
if len(fc) == 0 {
return api.DataSize{}, xerrors.New("no CARv2 file for root")
}
rdOnly, err := blockstore.OpenReadOnly(carv2FilePath)
rdOnly, err := filestorecaradapter.NewReadOnlyFileStore(fc)
if err != nil {
return api.DataSize{}, xerrors.Errorf("failed to open read only blockstore: %w", err)
return api.DataSize{}, xerrors.Errorf("failed to open read only filestore: %w", err)
}
defer rdOnly.Close() //nolint:errcheck
@@ -1030,15 +1031,15 @@ func (a *API) ClientDealSize(ctx context.Context, root cid.Cid) (api.DataSize, e
}
func (a *API) ClientDealPieceCID(ctx context.Context, root cid.Cid) (api.DataCIDSize, error) {
carv2FilePath, err := a.imgr().CARV2FilePathFor(root)
fc, err := a.imgr().FilestoreCARV2FilePathFor(root)
if err != nil {
return api.DataCIDSize{}, xerrors.Errorf("failed to find CARv2 file for root: %w", err)
}
if len(carv2FilePath) == 0 {
if len(fc) == 0 {
return api.DataCIDSize{}, xerrors.New("no CARv2 file for root")
}
rdOnly, err := blockstore.OpenReadOnly(carv2FilePath)
rdOnly, err := filestorecaradapter.NewReadOnlyFileStore(fc)
if err != nil {
return api.DataCIDSize{}, xerrors.Errorf("failed to open read only blockstore: %w", err)
}
@@ -1069,7 +1070,7 @@ func (a *API) ClientGenCar(ctx context.Context, ref api.FileRef, outputPath stri
}
defer os.Remove(tmpCARv2File) //nolint:errcheck
root, err := a.importNormalFileToCARv2(ctx, id, ref.Path, tmpCARv2File)
root, err := a.importNormalFileToFilestoreCARv2(ctx, id, ref.Path, tmpCARv2File)
if err != nil {
return xerrors.Errorf("failed to import normal file to CARv2")
}
node/impl/client/import.go
+9 -8
View File
@@ -6,6 +6,7 @@ import (
"io"
"os"
"github.com/filecoin-project/go-fil-markets/filestorecaradapter"
bstore "github.com/filecoin-project/lotus/blockstore"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/node/repo/importmgr"
@@ -24,12 +25,12 @@ import (
ihelper "github.com/ipfs/go-unixfs/importer/helpers"
"github.com/ipld/go-car"
carv2 "github.com/ipld/go-car/v2"
"github.com/ipld/go-car/v2/blockstore"
"golang.org/x/xerrors"
)
// importNormalFileToCARv2 transforms the client's "normal file" to a Unixfs IPLD DAG and writes out the DAG to a CARv2 file at the given output path.
func (a *API) importNormalFileToCARv2(ctx context.Context, importID importmgr.ImportID, inputFilePath string, outputCARv2Path string) (c cid.Cid, finalErr error) {
// importNormalFileToFilestoreCARv2 transforms the client's "normal file" to a Unixfs IPLD DAG and writes out the DAG to a CARv2 file
// that can be used to back a filestore.
func (a *API) importNormalFileToFilestoreCARv2(ctx context.Context, importID importmgr.ImportID, inputFilePath string, outputCARv2Path string) (c cid.Cid, finalErr error) {
// TODO: We've currently put in a hack to create the Unixfs DAG as a CARv2 without using Badger.
// We first create the Unixfs DAG using a filestore to get the root of the Unixfs DAG.
@@ -38,7 +39,6 @@ func (a *API) importNormalFileToCARv2(ctx context.Context, importID importmgr.Im
//
// In the second pass, we create a CARv2 file with the root present using the root node we get in the above step.
// This hack should be fixed when CARv2 allows specifying the root AFTER finishing the CARv2 streaming write.
// TODO How do we clean this up ?
fm := filestore.NewFileManager(ds_sync.MutexWrap(datastore.NewMapDatastore()), "/")
fm.AllowFiles = true
fstore := filestore.NewFilestore(bstore.NewMemorySync(), fm)
@@ -51,14 +51,15 @@ func (a *API) importNormalFileToCARv2(ctx context.Context, importID importmgr.Im
return cid.Undef, xerrors.Errorf("failed to import file to store: %w", err)
}
//------ Second Pass --- Now that we have the root of the Unixfs DAG -> write out the Unixfs DAG to a CARv2 file with the root present by using a read-write CARv2 blockstore.
rw, err := blockstore.NewReadWrite(outputCARv2Path, []cid.Cid{root})
//------ Second Pass --- Now that we have the root of the Unixfs DAG -> write out the Unixfs DAG to a CARv2 file with the root present by using a
// filestore backed by a read-write CARv2 blockstore.
fsb, err := filestorecaradapter.NewReadWriteFileStore(outputCARv2Path, []cid.Cid{root})
if err != nil {
return cid.Undef, xerrors.Errorf("failed to create a CARv2 read-write blockstore: %w", err)
}
defer rw.Finalize() //nolint:errcheck
defer fsb.Close() //nolint:errcheck
bsvc = blockservice.New(rw, offline.Exchange(rw))
bsvc = blockservice.New(fsb, offline.Exchange(fsb))
root2, err := importNormalFileToUnixfsDAG(ctx, inputFilePath, merkledag.NewDAGService(bsvc))
if err != nil {
return cid.Undef, xerrors.Errorf("failed to create Unixfs DAG with CARv2 blockstore: %w", err)
node/impl/client/import_test.go
+3 -2
View File
@@ -10,6 +10,7 @@ import (
"strings"
"testing"
"github.com/filecoin-project/go-fil-markets/filestorecaradapter"
bstore "github.com/filecoin-project/lotus/blockstore"
"github.com/filecoin-project/lotus/node/repo/importmgr"
"github.com/ipfs/go-blockservice"
@@ -81,12 +82,12 @@ func TestImportNormalFileToCARv2(t *testing.T) {
outputCARv2 := genTmpFile(t)
defer os.Remove(outputCARv2) //nolint:errcheck
root, err := a.importNormalFileToCARv2(ctx, importID, inputFilePath, outputCARv2)
root, err := a.importNormalFileToFilestoreCARv2(ctx, importID, inputFilePath, outputCARv2)
require.NoError(t, err)
require.NotEqual(t, cid.Undef, root)
// convert the CARv2 to a normal file again and ensure the contents match
readOnly, err := blockstore.OpenReadOnly(outputCARv2)
readOnly, err := filestorecaradapter.NewReadOnlyFileStore(outputCARv2)
require.NoError(t, err)
defer readOnly.Close() //nolint:errcheck
dag := merkledag.NewDAGService(blockservice.New(readOnly, offline.Exchange(readOnly)))
node/modules/storageminer.go
+9 -19
View File
@@ -28,8 +28,6 @@ import (
"github.com/libp2p/go-libp2p-core/host"
"github.com/libp2p/go-libp2p-core/routing"
"github.com/filecoin-project/dagstore"
"github.com/filecoin-project/dagstore/mount"
"github.com/filecoin-project/go-address"
dtimpl "github.com/filecoin-project/go-data-transfer/impl"
dtnet "github.com/filecoin-project/go-data-transfer/network"
@@ -568,29 +566,21 @@ func BasicDealFilter(user dtypes.StorageDealFilter) func(onlineOk dtypes.Conside
}
}
func DAGStoreRegistry() *mount.Registry {
return mount.NewRegistry()
}
func DAGStore(r repo.LockedRepo, ds dtypes.MetadataDS, dsRegistry *mount.Registry) (*dagstore.DAGStore, error) {
func DAGStoreWrapper(
ds dtypes.MetadataDS,
r repo.LockedRepo,
pieceStore dtypes.ProviderPieceStore,
rpn retrievalmarket.RetrievalProviderNode,
) (mktdagstore.DagStoreWrapper, error) {
dagStoreDir := filepath.Join(r.Path(), "dagstore")
dagStoreDS := namespace.Wrap(ds, datastore.NewKey("/dagstore/provider"))
return dagstore.NewDAGStore(dagstore.Config{
cfg := mktdagstore.MarketDAGStoreConfig{
TransientsDir: filepath.Join(dagStoreDir, "transients"),
IndexDir: filepath.Join(dagStoreDir, "index"),
Datastore: dagStoreDS,
MountRegistry: dsRegistry,
})
}
func DAGStoreWrapper(
pieceStore dtypes.ProviderPieceStore,
rpn retrievalmarket.RetrievalProviderNode,
dsRegistry *mount.Registry,
dagStore *dagstore.DAGStore,
) (mktdagstore.DagStoreWrapper, error) {
}
mountApi := mktdagstore.NewLotusMountAPI(pieceStore, rpn)
return mktdagstore.NewDagStoreWrapper(dsRegistry, dagStore, mountApi)
return mktdagstore.NewDagStoreWrapper(cfg, mountApi)
}
func StorageProvider(minerAddress dtypes.MinerAddress,
node/repo/importmgr/mgr.go
+6 -6
View File
@@ -29,10 +29,10 @@ type Mgr struct {
type Label string
const (
LSource = "source" // Function which created the import
LRootCid = "root" // Root CID
LFileName = "filename" // Local file path
LCARv2FilePath = "CARv2Path" // path of the CARv2 file.
LSource = "source" // Function which created the import
LRootCid = "root" // Root CID
LFileName = "filename" // Local file path
LFileStoreCARv2FilePath = "CARv2Path" // path of the full CARv2 file or a CARv2 file that can serve as the backing store for a Filestore.
)
func New(ds datastore.Batching, repoPath string) *Mgr {
@@ -129,7 +129,7 @@ func (m *Mgr) Remove(id ImportID) error {
return nil
}
func (m *Mgr) CARV2FilePathFor(dagRoot cid.Cid) (string, error) {
func (m *Mgr) FilestoreCARV2FilePathFor(dagRoot cid.Cid) (string, error) {
importIDs, err := m.List()
if err != nil {
return "", xerrors.Errorf("failed to fetch import IDs: %w", err)
@@ -150,7 +150,7 @@ func (m *Mgr) CARV2FilePathFor(dagRoot cid.Cid) (string, error) {
continue
}
if c.Equals(dagRoot) {
return info.Labels[LCARv2FilePath], nil
return info.Labels[LFileStoreCARv2FilePath], nil
}
}