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

draft lotus storage client CARv2 changes

Browse Source
This commit is contained in:
aarshkshah1992
2021-07-03 14:12:08 +05:30
parent adba1a792a
commit 31ecce3a72
7 changed files with 343 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/impl/client/client.go
+104 -130
View File
@@ -5,6 +5,7 @@ import (
"context"
"fmt"
"io"
"io/ioutil"
"os"
"sort"
"time"
@@ -18,15 +19,11 @@ import (
"github.com/filecoin-project/go-state-types/dline"
"github.com/ipfs/go-blockservice"
"github.com/ipfs/go-cid"
"github.com/ipfs/go-cidutil"
chunker "github.com/ipfs/go-ipfs-chunker"
offline "github.com/ipfs/go-ipfs-exchange-offline"
files "github.com/ipfs/go-ipfs-files"
ipld "github.com/ipfs/go-ipld-format"
"github.com/ipfs/go-merkledag"
unixfile "github.com/ipfs/go-unixfs/file"
"github.com/ipfs/go-unixfs/importer/balanced"
ihelper "github.com/ipfs/go-unixfs/importer/helpers"
"github.com/ipld/go-car"
basicnode "github.com/ipld/go-ipld-prime/node/basic"
"github.com/ipld/go-ipld-prime/traversal/selector"
@@ -91,6 +88,8 @@ type API struct {
RetrievalStoreMgr dtypes.ClientRetrievalStoreManager
DataTransfer dtypes.ClientDataTransfer
Host host.Host
// TODO How do we inject the Repo Path here ?
}
func calcDealExpiration(minDuration uint64, md *dline.Info, startEpoch abi.ChainEpoch) abi.ChainEpoch {
@@ -120,7 +119,7 @@ func (a *API) ClientStatelessDeal(ctx context.Context, params *api.StartDealPara
}
func (a *API) dealStarter(ctx context.Context, params *api.StartDealParams, isStateless bool) (*cid.Cid, error) {
var storeID *multistore.StoreID
var CARV2FilePath string
if isStateless {
if params.Data.TransferType != storagemarket.TTManual {
return nil, xerrors.Errorf("invalid transfer type %s for stateless storage deal", params.Data.TransferType)
@@ -143,7 +142,7 @@ func (a *API) dealStarter(ctx context.Context, params *api.StartDealParams, isSt
continue
}
if c.Equals(params.Data.Root) {
storeID = &importID //nolint
CARV2FilePath = info.Labels[importmgr.LCARv2FilePath]
break
}
}
@@ -212,7 +211,7 @@ func (a *API) dealStarter(ctx context.Context, params *api.StartDealParams, isSt
Rt: st,
FastRetrieval: params.FastRetrieval,
VerifiedDeal: params.VerifiedDeal,
StoreID: storeID,
CARV2FilePath: CARV2FilePath,
})
if err != nil {
@@ -483,80 +482,129 @@ func (a *API) makeRetrievalQuery(ctx context.Context, rp rm.RetrievalPeer, paylo
}
}
func (a *API) ClientImport(ctx context.Context, ref api.FileRef) (*api.ImportRes, error) {
func (a *API) ClientImport(ctx context.Context, ref api.FileRef) (res *api.ImportRes, finalErr error) {
id, st, err := a.imgr().NewStore()
if err != nil {
return nil, err
}
// we don't need the store any more after we return from here. clean it up completely.
// we only need to retain the metadata related to this import which is identified by the storeID/importID.
defer func() {
//_ = ((*multistore.MultiStore)(a.Mds)).Delete(id)
}()
carV2File, err := a.newTempFilePath(id)
if err != nil {
return nil, xerrors.Errorf("failed to create temp CARv2 file: %w", err)
}
// make sure to remove the CARv2 file if anything goes wrong from here on.
defer func() {
if finalErr != nil {
_ = os.Remove(carV2File)
}
}()
var root cid.Cid
if ref.IsCAR {
root, err = transformCarToCARv2(ref.Path, carV2File)
if err != nil {
return nil, xerrors.Errorf("failed to import CAR file: %w", err)
}
} else {
root, err = importNormalFileToCARv2(ctx, st, ref.Path, carV2File)
if err != nil {
return nil, xerrors.Errorf("failed to import normal file to CARv2: %w", err)
}
}
if err := a.imgr().AddLabel(id, importmgr.LSource, "import"); err != nil {
return nil, err
}
if err := a.imgr().AddLabel(id, importmgr.LFileName, ref.Path); err != nil {
return nil, err
}
nd, err := a.clientImport(ctx, ref, st)
if err != nil {
if err := a.imgr().AddLabel(id, importmgr.LCARv2FilePath, carV2File); err != nil {
return nil, err
}
if err := a.imgr().AddLabel(id, importmgr.LRootCid, nd.String()); err != nil {
if err := a.imgr().AddLabel(id, importmgr.LRootCid, root.String()); err != nil {
return nil, err
}
return &api.ImportRes{
Root: nd,
Root: root,
ImportID: id,
}, nil
}
func (a *API) newTempFilePath(id multistore.StoreID) (string, error) {
// TODO Get the repo path here.
file, err := ioutil.TempFile("", fmt.Sprintf("%d", id))
if err != nil {
return "nil", xerrors.Errorf("failed to create temp CARv2 file: %w", err)
}
if err := file.Close(); err != nil {
return "", xerrors.Errorf("failed to close CARv2 file")
}
return file.Name(), nil
}
func (a *API) ClientRemoveImport(ctx context.Context, importID multistore.StoreID) error {
info, err := a.imgr().Info(importID)
if err != nil {
return xerrors.Errorf("failed to fetch multistore info: %w", err)
}
// remove the CARv2 file if we've created one.
if path := info.Labels[importmgr.LCARv2FilePath]; path != "" {
_ = os.Remove(path)
}
return a.imgr().Remove(importID)
}
func (a *API) ClientImportLocal(ctx context.Context, f io.Reader) (cid.Cid, error) {
file := files.NewReaderFile(f)
// FIXME
func (a *API) ClientImportLocal(ctx context.Context, f io.Reader) (c cid.Cid, finalErr error) {
id, st, err := a.imgr().NewStore()
if err != nil {
return cid.Undef, err
}
if err := a.imgr().AddLabel(id, "source", "import-local"); err != nil {
// we don't need the store any more after we return from here. clean it up completely.
// we only need to retain the metadata related to this import which is identified by the storeID/importID.
defer func() {
_ = ((*multistore.MultiStore)(a.Mds)).Delete(id)
}()
carV2File, err := a.newTempFilePath(id)
if err != nil {
return cid.Undef, xerrors.Errorf("failed to create temp CARv2 file: %w", err)
}
// make sure to remove the CARv2 file if anything goes wrong from here on.
defer func() {
if finalErr != nil {
_ = os.Remove(carV2File)
}
}()
// FIXME
root, err := importNormalFileToCARv2(ctx, st, "", carV2File)
if err != nil {
return root, xerrors.Errorf("failed to import to CARv2 file: %w", err)
}
if err := a.imgr().AddLabel(id, importmgr.LSource, "import-local"); err != nil {
return cid.Cid{}, err
}
bufferedDS := ipld.NewBufferedDAG(ctx, st.DAG)
prefix, err := merkledag.PrefixForCidVersion(1)
if err != nil {
return cid.Undef, err
}
prefix.MhType = DefaultHashFunction
params := ihelper.DagBuilderParams{
Maxlinks: build.UnixfsLinksPerLevel,
RawLeaves: true,
CidBuilder: cidutil.InlineBuilder{
Builder: prefix,
Limit: 126,
},
Dagserv: bufferedDS,
}
db, err := params.New(chunker.NewSizeSplitter(file, int64(build.UnixfsChunkSize)))
if err != nil {
return cid.Undef, err
}
nd, err := balanced.Layout(db)
if err != nil {
return cid.Undef, err
}
if err := a.imgr().AddLabel(id, "root", nd.Cid().String()); err != nil {
if err := a.imgr().AddLabel(id, importmgr.LRootCid, root.String()); err != nil {
return cid.Cid{}, err
}
if err := a.imgr().AddLabel(id, importmgr.LCARv2FilePath, carV2File); err != nil {
return cid.Undef, err
}
return nd.Cid(), bufferedDS.Commit()
return root, nil
}
func (a *API) ClientListImports(ctx context.Context) ([]api.Import, error) {
@@ -1036,31 +1084,28 @@ func (a *API) ClientGenCar(ctx context.Context, ref api.FileRef, outputPath stri
if err != nil {
return err
}
if err := a.imgr().AddLabel(id, "source", "gen-car"); err != nil {
return err
}
bufferedDS := ipld.NewBufferedDAG(ctx, st.DAG)
c, err := a.clientImport(ctx, ref, st)
defer func() {
// Clean up the store as we don't need it anymore.
_ = a.imgr().Remove(id)
}()
c, err := importNormalFileToUnixfsDAG(ctx, ref.Path, st.DAG)
if err != nil {
return err
return xerrors.Errorf("failed to import file to store: %w", err)
}
// TODO: does that defer mean to remove the whole blockstore?
defer bufferedDS.Remove(ctx, c) //nolint:errcheck
// generate a deterministic CARv1 payload from the UnixFS DAG by doing an IPLD
// traversal over the Unixfs DAGs in the blockstore using the "all selector" i.e the entire DAG selector.
ssb := builder.NewSelectorSpecBuilder(basicnode.Prototype.Any)
// entire DAG selector
allSelector := ssb.ExploreRecursive(selector.RecursionLimitNone(),
ssb.ExploreAll(ssb.ExploreRecursiveEdge())).Node()
sc := car.NewSelectiveCar(ctx, st.Bstore, []car.Dag{{Root: c, Selector: allSelector}})
f, err := os.Create(outputPath)
if err != nil {
return err
}
sc := car.NewSelectiveCar(ctx, st.Bstore, []car.Dag{{Root: c, Selector: allSelector}})
if err = sc.Write(f); err != nil {
return err
}
@@ -1068,77 +1113,6 @@ func (a *API) ClientGenCar(ctx context.Context, ref api.FileRef, outputPath stri
return f.Close()
}
func (a *API) clientImport(ctx context.Context, ref api.FileRef, store *multistore.Store) (cid.Cid, error) {
f, err := os.Open(ref.Path)
if err != nil {
return cid.Undef, err
}
defer f.Close() //nolint:errcheck
stat, err := f.Stat()
if err != nil {
return cid.Undef, err
}
file, err := files.NewReaderPathFile(ref.Path, f, stat)
if err != nil {
return cid.Undef, err
}
if ref.IsCAR {
var st car.Store
if store.Fstore == nil {
st = store.Bstore
} else {
st = store.Fstore
}
result, err := car.LoadCar(st, file)
if err != nil {
return cid.Undef, err
}
if len(result.Roots) != 1 {
return cid.Undef, xerrors.New("cannot import car with more than one root")
}
return result.Roots[0], nil
}
bufDs := ipld.NewBufferedDAG(ctx, store.DAG)
prefix, err := merkledag.PrefixForCidVersion(1)
if err != nil {
return cid.Undef, err
}
prefix.MhType = DefaultHashFunction
params := ihelper.DagBuilderParams{
Maxlinks: build.UnixfsLinksPerLevel,
RawLeaves: true,
CidBuilder: cidutil.InlineBuilder{
Builder: prefix,
Limit: 126,
},
Dagserv: bufDs,
NoCopy: true,
}
db, err := params.New(chunker.NewSizeSplitter(file, int64(build.UnixfsChunkSize)))
if err != nil {
return cid.Undef, err
}
nd, err := balanced.Layout(db)
if err != nil {
return cid.Undef, err
}
if err := bufDs.Commit(); err != nil {
return cid.Undef, err
}
return nd.Cid(), nil
}
func (a *API) ClientListDataTransfers(ctx context.Context) ([]api.DataTransferChannel, error) {
inProgressChannels, err := a.DataTransfer.InProgressChannels(ctx)
if err != nil {
node/impl/client/import.go
+169
View File
@@ -0,0 +1,169 @@
package client
import (
"bufio"
"context"
"fmt"
"io"
"os"
"github.com/filecoin-project/go-multistore"
"github.com/filecoin-project/lotus/build"
"github.com/ipfs/go-blockservice"
"github.com/ipfs/go-cid"
"github.com/ipfs/go-cidutil"
chunker "github.com/ipfs/go-ipfs-chunker"
offline "github.com/ipfs/go-ipfs-exchange-offline"
files2 "github.com/ipfs/go-ipfs-files"
ipld "github.com/ipfs/go-ipld-format"
"github.com/ipfs/go-merkledag"
"github.com/ipfs/go-unixfs/importer/balanced"
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 imports the client's normal file to a CARv2 file.
// It first generates a Unixfs DAG using the given store to store the resulting blocks and gets the root cid of the Unixfs DAG.
// It then writes out the Unixfs DAG to a CARv2 file by generating a Unixfs DAG again using a CARv2 read-write blockstore as the backing store
// and then finalizing the CARv2 read-write blockstore to get the backing CARv2 file.
func importNormalFileToCARv2(ctx context.Context, st *multistore.Store, inputFilePath string, outputCARv2Path string) (c cid.Cid, finalErr error) {
// create the UnixFS DAG and import the file to store to get the root.
root, err := importNormalFileToUnixfsDAG(ctx, inputFilePath, st.DAG)
if err != nil {
return cid.Undef, xerrors.Errorf("failed to import file to store: %w", err)
}
//---
// transform the file to a CARv2 file by writing out a Unixfs DAG via the CARv2 read-write blockstore.
rw, err := blockstore.NewReadWrite(outputCARv2Path, []cid.Cid{root})
if err != nil {
return cid.Undef, xerrors.Errorf("failed to create a CARv2 read-write blockstore: %w", err)
}
// make sure to call finalize on the CARv2 read-write blockstore to ensure that the blockstore flushes out a valid CARv2 file
// and releases the file handle it acquires.
defer func() {
err := rw.Finalize()
if finalErr != nil {
finalErr = xerrors.Errorf("failed to import file to CARv2, err=%w, also failed to finalize rw CARv2 blockstore, err=%w", finalErr,
err)
} else {
finalErr = err
}
}()
bsvc := blockservice.New(rw, offline.Exchange(rw))
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)
}
fmt.Printf("\n root1 is %s and root2 is %s", root, root2)
if root != root2 {
return cid.Undef, xerrors.New("roots do not match")
}
return root, nil
}
// importNormalFileToUnixfsDAG transforms a client's normal file to a UnixfsDAG and imports the DAG to the given DAG service.
func importNormalFileToUnixfsDAG(ctx context.Context, inputFilePath string, dag ipld.DAGService) (cid.Cid, error) {
f, err := os.Open(inputFilePath)
if err != nil {
return cid.Undef, xerrors.Errorf("failed to open input file: %w", err)
}
defer f.Close()
stat, err := f.Stat()
if err != nil {
return cid.Undef, xerrors.Errorf("failed to stat file :%w", err)
}
file, err := files2.NewReaderPathFile(inputFilePath, f, stat)
if err != nil {
return cid.Undef, xerrors.Errorf("failed to create reader path file: %w", err)
}
bufDs := ipld.NewBufferedDAG(ctx, dag)
prefix, err := merkledag.PrefixForCidVersion(1)
if err != nil {
return cid.Undef, err
}
prefix.MhType = DefaultHashFunction
params := ihelper.DagBuilderParams{
Maxlinks: build.UnixfsLinksPerLevel,
RawLeaves: true,
CidBuilder: cidutil.InlineBuilder{
Builder: prefix,
Limit: 126,
},
Dagserv: bufDs,
NoCopy: true,
}
db, err := params.New(chunker.NewSizeSplitter(file, int64(build.UnixfsChunkSize)))
if err != nil {
return cid.Undef, err
}
nd, err := balanced.Layout(db)
if err != nil {
return cid.Undef, err
}
if err := bufDs.Commit(); err != nil {
return cid.Undef, err
}
return nd.Cid(), nil
}
// transformCarToCARv2 transforms a client's CAR file to a CARv2 file.
func transformCarToCARv2(inputCARPath string, outputCARv2Path string) (root cid.Cid, err error) {
inputF, err := os.Open(inputCARPath)
if err != nil {
return cid.Undef, xerrors.Errorf("failed to open input CAR: %w", err)
}
defer inputF.Close() //nolint:errcheck
// read the CAR header to determine the DAG root and the CAR version.
hd, _, err := car.ReadHeader(bufio.NewReader(inputF))
if err != nil {
return cid.Undef, xerrors.Errorf("failed to read CAR header: %w", err)
}
if len(hd.Roots) != 1 {
return cid.Undef, xerrors.New("cannot import CAR with more than one root")
}
switch hd.Version {
case 2:
// This is a CARv2, we can import it as it is by simply copying it.
outF, err := os.Open(outputCARv2Path)
if err != nil {
return cid.Undef, xerrors.Errorf("failed to open output CARv2 file: %w", err)
}
defer outF.Close()
_, err = io.Copy(outF, inputF)
if err != nil {
return cid.Undef, xerrors.Errorf("failed to copy CARv2 file: %w", err)
}
case 1:
// This is a CARv1, let's transform it to a CARv2.
if err := carv2.WrapV1File(inputCARPath, outputCARv2Path); err != nil {
return cid.Undef, xerrors.Errorf("failed to transform CARv1 to CARv2: %w", err)
}
default:
return cid.Undef, xerrors.Errorf("unrecognized CAR version %d", hd.Version)
}
return hd.Roots[0], nil
}