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

Merge remote-tracking branch 'origin/master' into feat/post-worker

Browse Source
This commit is contained in:
Łukasz Magiera
2022-03-21 12:09:47 +01:00
parent cca69a6c16 d118ef9202
commit 92e4caa151
23 changed files with 525 additions and 309 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/impl/client/client.go
+6 -6
View File
@@ -42,7 +42,6 @@ import (
"github.com/libp2p/go-libp2p-core/host"
"github.com/libp2p/go-libp2p-core/peer"
"github.com/multiformats/go-multibase"
mh "github.com/multiformats/go-multihash"
"go.uber.org/fx"
"github.com/filecoin-project/go-address"
@@ -56,6 +55,7 @@ import (
"github.com/filecoin-project/go-fil-markets/storagemarket/network"
"github.com/filecoin-project/go-fil-markets/stores"
"github.com/filecoin-project/lotus/lib/unixfs"
"github.com/filecoin-project/lotus/markets/retrievaladapter"
"github.com/filecoin-project/lotus/markets/storageadapter"
@@ -79,7 +79,7 @@ import (
var log = logging.Logger("client")
var DefaultHashFunction = uint64(mh.BLAKE2B_MIN + 31)
var DefaultHashFunction = unixfs.DefaultHashFunction
// 8 days ~= SealDuration + PreCommit + MaxProveCommitDuration + 8 hour buffer
const dealStartBufferHours uint64 = 8 * 24
@@ -548,7 +548,7 @@ func (a *API) ClientImport(ctx context.Context, ref api.FileRef) (res *api.Impor
}()
// perform the unixfs chunking.
root, err = a.createUnixFSFilestore(ctx, ref.Path, carPath)
root, err = unixfs.CreateFilestore(ctx, ref.Path, carPath)
if err != nil {
return nil, xerrors.Errorf("failed to import file using unixfs: %w", err)
}
@@ -618,7 +618,7 @@ func (a *API) ClientImportLocal(ctx context.Context, r io.Reader) (cid.Cid, erro
// once the DAG is formed and the root is calculated, we overwrite the
// inner carv1 header with the final root.
b, err := unixFSCidBuilder()
b, err := unixfs.CidBuilder()
if err != nil {
return cid.Undef, err
}
@@ -635,7 +635,7 @@ func (a *API) ClientImportLocal(ctx context.Context, r io.Reader) (cid.Cid, erro
return cid.Undef, xerrors.Errorf("failed to create carv2 read/write blockstore: %w", err)
}
root, err := buildUnixFS(ctx, file, bs, false)
root, err := unixfs.Build(ctx, file, bs, false)
if err != nil {
return cid.Undef, xerrors.Errorf("failed to build unixfs dag: %w", err)
}
@@ -1364,7 +1364,7 @@ func (a *API) ClientGenCar(ctx context.Context, ref api.FileRef, outputPath stri
defer os.Remove(tmp) //nolint:errcheck
// generate and import the UnixFS DAG into a filestore (positional reference) CAR.
root, err := a.createUnixFSFilestore(ctx, ref.Path, tmp)
root, err := unixfs.CreateFilestore(ctx, ref.Path, tmp)
if err != nil {
return xerrors.Errorf("failed to import file using unixfs: %w", err)
}
node/impl/client/import.go
-156
View File
@@ -1,156 +0,0 @@
package client
import (
"context"
"fmt"
"io"
"io/ioutil"
"os"
"github.com/filecoin-project/go-fil-markets/stores"
"github.com/ipfs/go-blockservice"
"github.com/ipfs/go-cid"
"github.com/ipfs/go-cidutil"
bstore "github.com/ipfs/go-ipfs-blockstore"
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"
"github.com/ipfs/go-unixfs/importer/balanced"
ihelper "github.com/ipfs/go-unixfs/importer/helpers"
"golang.org/x/xerrors"
"github.com/filecoin-project/lotus/build"
)
func unixFSCidBuilder() (cid.Builder, error) {
prefix, err := merkledag.PrefixForCidVersion(1)
if err != nil {
return nil, fmt.Errorf("failed to initialize UnixFS CID Builder: %w", err)
}
prefix.MhType = DefaultHashFunction
b := cidutil.InlineBuilder{
Builder: prefix,
Limit: 126,
}
return b, nil
}
// createUnixFSFilestore takes a standard file whose path is src, forms a UnixFS DAG, and
// writes a CARv2 file with positional mapping (backed by the go-filestore library).
func (a *API) createUnixFSFilestore(ctx context.Context, srcPath string, dstPath string) (cid.Cid, error) {
// This method uses a two-phase approach with a staging CAR blockstore and
// a final CAR blockstore.
//
// This is necessary because of https://github.com/ipld/go-car/issues/196
//
// TODO: do we need to chunk twice? Isn't the first output already in the
// right order? Can't we just copy the CAR file and replace the header?
src, err := os.Open(srcPath)
if err != nil {
return cid.Undef, xerrors.Errorf("failed to open input file: %w", err)
}
defer src.Close() //nolint:errcheck
stat, err := src.Stat()
if err != nil {
return cid.Undef, xerrors.Errorf("failed to stat file :%w", err)
}
file, err := files.NewReaderPathFile(srcPath, src, stat)
if err != nil {
return cid.Undef, xerrors.Errorf("failed to create reader path file: %w", err)
}
f, err := ioutil.TempFile("", "")
if err != nil {
return cid.Undef, xerrors.Errorf("failed to create temp file: %w", err)
}
_ = f.Close() // close; we only want the path.
tmp := f.Name()
defer os.Remove(tmp) //nolint:errcheck
// Step 1. Compute the UnixFS DAG and write it to a CARv2 file to get
// the root CID of the DAG.
fstore, err := stores.ReadWriteFilestore(tmp)
if err != nil {
return cid.Undef, xerrors.Errorf("failed to create temporary filestore: %w", err)
}
finalRoot1, err := buildUnixFS(ctx, file, fstore, true)
if err != nil {
_ = fstore.Close()
return cid.Undef, xerrors.Errorf("failed to import file to store to compute root: %w", err)
}
if err := fstore.Close(); err != nil {
return cid.Undef, xerrors.Errorf("failed to finalize car filestore: %w", err)
}
// Step 2. We now have the root of the UnixFS DAG, and we can write the
// final CAR for real under `dst`.
bs, err := stores.ReadWriteFilestore(dstPath, finalRoot1)
if err != nil {
return cid.Undef, xerrors.Errorf("failed to create a carv2 read/write filestore: %w", err)
}
// rewind file to the beginning.
if _, err := src.Seek(0, 0); err != nil {
return cid.Undef, xerrors.Errorf("failed to rewind file: %w", err)
}
finalRoot2, err := buildUnixFS(ctx, file, bs, true)
if err != nil {
_ = bs.Close()
return cid.Undef, xerrors.Errorf("failed to create UnixFS DAG with carv2 blockstore: %w", err)
}
if err := bs.Close(); err != nil {
return cid.Undef, xerrors.Errorf("failed to finalize car blockstore: %w", err)
}
if finalRoot1 != finalRoot2 {
return cid.Undef, xerrors.New("roots do not match")
}
return finalRoot1, nil
}
// buildUnixFS builds a UnixFS DAG out of the supplied reader,
// and imports the DAG into the supplied service.
func buildUnixFS(ctx context.Context, reader io.Reader, into bstore.Blockstore, filestore bool) (cid.Cid, error) {
b, err := unixFSCidBuilder()
if err != nil {
return cid.Undef, err
}
bsvc := blockservice.New(into, offline.Exchange(into))
dags := merkledag.NewDAGService(bsvc)
bufdag := ipld.NewBufferedDAG(ctx, dags)
params := ihelper.DagBuilderParams{
Maxlinks: build.UnixfsLinksPerLevel,
RawLeaves: true,
CidBuilder: b,
Dagserv: bufdag,
NoCopy: filestore,
}
db, err := params.New(chunker.NewSizeSplitter(reader, int64(build.UnixfsChunkSize)))
if err != nil {
return cid.Undef, err
}
nd, err := balanced.Layout(db)
if err != nil {
return cid.Undef, err
}
if err := bufdag.Commit(); err != nil {
return cid.Undef, err
}
return nd.Cid(), nil
}
node/impl/client/import_test.go
-136
View File
@@ -1,136 +0,0 @@
//stm: #unit
package client
import (
"bytes"
"context"
"io"
"io/ioutil"
"os"
"strings"
"testing"
"github.com/ipfs/go-blockservice"
"github.com/ipfs/go-cid"
offline "github.com/ipfs/go-ipfs-exchange-offline"
files "github.com/ipfs/go-ipfs-files"
"github.com/ipfs/go-merkledag"
unixfile "github.com/ipfs/go-unixfs/file"
carv2 "github.com/ipld/go-car/v2"
"github.com/ipld/go-car/v2/blockstore"
"github.com/stretchr/testify/require"
"github.com/filecoin-project/go-fil-markets/stores"
"github.com/filecoin-project/lotus/node/repo/imports"
)
// This test uses a full "dense" CARv2, and not a filestore (positional mapping).
func TestRoundtripUnixFS_Dense(t *testing.T) {
//stm: @CLIENT_DATA_IMPORT_002
ctx := context.Background()
inputPath, inputContents := genInputFile(t)
defer os.Remove(inputPath) //nolint:errcheck
carv2File := newTmpFile(t)
defer os.Remove(carv2File) //nolint:errcheck
// import a file to a Unixfs DAG using a CARv2 read/write blockstore.
bs, err := blockstore.OpenReadWrite(carv2File, nil,
carv2.ZeroLengthSectionAsEOF(true),
blockstore.UseWholeCIDs(true))
require.NoError(t, err)
root, err := buildUnixFS(ctx, bytes.NewBuffer(inputContents), bs, false)
require.NoError(t, err)
require.NotEqual(t, cid.Undef, root)
require.NoError(t, bs.Finalize())
// reconstruct the file.
readOnly, err := blockstore.OpenReadOnly(carv2File,
carv2.ZeroLengthSectionAsEOF(true),
blockstore.UseWholeCIDs(true))
require.NoError(t, err)
defer readOnly.Close() //nolint:errcheck
dags := merkledag.NewDAGService(blockservice.New(readOnly, offline.Exchange(readOnly)))
nd, err := dags.Get(ctx, root)
require.NoError(t, err)
file, err := unixfile.NewUnixfsFile(ctx, dags, nd)
require.NoError(t, err)
tmpOutput := newTmpFile(t)
defer os.Remove(tmpOutput) //nolint:errcheck
require.NoError(t, files.WriteTo(file, tmpOutput))
// ensure contents of the initial input file and the output file are identical.
fo, err := os.Open(tmpOutput)
require.NoError(t, err)
bz2, err := ioutil.ReadAll(fo)
require.NoError(t, err)
require.NoError(t, fo.Close())
require.Equal(t, inputContents, bz2)
}
func TestRoundtripUnixFS_Filestore(t *testing.T) {
//stm: @CLIENT_DATA_IMPORT_001
ctx := context.Background()
a := &API{
Imports: &imports.Manager{},
}
inputPath, inputContents := genInputFile(t)
defer os.Remove(inputPath) //nolint:errcheck
dst := newTmpFile(t)
defer os.Remove(dst) //nolint:errcheck
root, err := a.createUnixFSFilestore(ctx, inputPath, dst)
require.NoError(t, err)
require.NotEqual(t, cid.Undef, root)
// convert the CARv2 to a normal file again and ensure the contents match
fs, err := stores.ReadOnlyFilestore(dst)
require.NoError(t, err)
defer fs.Close() //nolint:errcheck
dags := merkledag.NewDAGService(blockservice.New(fs, offline.Exchange(fs)))
nd, err := dags.Get(ctx, root)
require.NoError(t, err)
file, err := unixfile.NewUnixfsFile(ctx, dags, nd)
require.NoError(t, err)
tmpOutput := newTmpFile(t)
defer os.Remove(tmpOutput) //nolint:errcheck
require.NoError(t, files.WriteTo(file, tmpOutput))
// ensure contents of the initial input file and the output file are identical.
fo, err := os.Open(tmpOutput)
require.NoError(t, err)
bz2, err := ioutil.ReadAll(fo)
require.NoError(t, err)
require.NoError(t, fo.Close())
require.Equal(t, inputContents, bz2)
}
func newTmpFile(t *testing.T) string {
f, err := os.CreateTemp("", "")
require.NoError(t, err)
require.NoError(t, f.Close())
return f.Name()
}
func genInputFile(t *testing.T) (filepath string, contents []byte) {
s := strings.Repeat("abcde", 100)
tmp, err := os.CreateTemp("", "")
require.NoError(t, err)
_, err = io.Copy(tmp, strings.NewReader(s))
require.NoError(t, err)
require.NoError(t, tmp.Close())
return tmp.Name(), []byte(s)
}