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

feat(markets): update to support car files

Browse Source 
Upgrades lotus version of go-fil-markets to its head, which supports car files and handles tracking
of pieces, reducing the size of sectorBlocks
This commit is contained in:
hannahhoward
2020-01-28 14:08:13 -08:00
parent eb4b85aea5
commit 71289b58ff
13 changed files with 238 additions and 329 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/builder.go
+1
View File
@@ -256,6 +256,7 @@ func Online() Option {
Override(new(dtypes.ProviderDealStore), modules.NewProviderDealStore),
Override(new(dtypes.ProviderDataTransfer), modules.NewProviderDAGServiceDataTransfer),
Override(new(*deals.ProviderRequestValidator), modules.NewProviderRequestValidator),
Override(new(dtypes.ProviderPieceStore), modules.NewProviderPieceStore),
Override(new(storagemarket.StorageProvider), modules.StorageProvider),
Override(new(storagemarket.StorageProviderNode), storageadapter.NewProviderNodeAdapter),
Override(RegisterProviderValidatorKey, modules.RegisterProviderValidator),
node/impl/client/client.go
+3 -4
View File
@@ -1,7 +1,6 @@
package client
import (
"bytes"
"context"
"errors"
"io"
@@ -155,7 +154,7 @@ func (a *API) ClientFindData(ctx context.Context, root cid.Cid) ([]api.QueryOffe
out := make([]api.QueryOffer, len(peers))
for k, p := range peers {
queryResponse, err := a.Retrieval.Query(ctx, p, root.Bytes(), retrievalmarket.QueryParams{})
queryResponse, err := a.Retrieval.Query(ctx, p, root, retrievalmarket.QueryParams{})
if err != nil {
out[k] = api.QueryOffer{Err: err.Error(), Miner: p.Address, MinerPeerID: p.ID}
} else {
@@ -279,7 +278,7 @@ func (a *API) ClientRetrieve(ctx context.Context, order api.RetrievalOrder, path
retrievalResult := make(chan error, 1)
unsubscribe := a.Retrieval.SubscribeToEvents(func(event retrievalmarket.ClientEvent, state retrievalmarket.ClientDealState) {
if bytes.Equal(state.PieceCID, order.Root.Bytes()) {
if state.PayloadCID.Equals(order.Root) {
switch event {
case retrievalmarket.ClientEventError:
retrievalResult <- xerrors.New("Retrieval Error")
@@ -291,7 +290,7 @@ func (a *API) ClientRetrieve(ctx context.Context, order api.RetrievalOrder, path
a.Retrieval.Retrieve(
ctx,
order.Root.Bytes(),
order.Root,
retrievalmarket.NewParamsV0(types.BigDiv(order.Total, types.NewInt(order.Size)).Int, 0, 0),
utils.ToSharedTokenAmount(order.Total),
order.MinerPeerID,
node/impl/storminer.go
+2 -1
View File
@@ -7,6 +7,7 @@ import (
"mime"
"net/http"
"os"
"strconv"
"github.com/filecoin-project/lotus/api/apistruct"
@@ -210,7 +211,7 @@ func (sm *StorageMinerAPI) SectorsRefs(context.Context) (map[string][]api.Sealed
}
for k, v := range refs {
out[k.String()] = v
out[strconv.FormatUint(k, 10)] = v
}
return out, nil
node/modules/client.go
+12 -5
View File
@@ -5,10 +5,12 @@ import (
"path/filepath"
"reflect"
"github.com/filecoin-project/go-data-transfer/impl/graphsync"
graphsyncimpl "github.com/filecoin-project/go-data-transfer/impl/graphsync"
piecefilestore "github.com/filecoin-project/go-fil-markets/filestore"
"github.com/filecoin-project/go-fil-markets/retrievalmarket"
"github.com/filecoin-project/go-fil-markets/retrievalmarket/discovery"
retrievalimpl "github.com/filecoin-project/go-fil-markets/retrievalmarket/impl"
rmnet "github.com/filecoin-project/go-fil-markets/retrievalmarket/network"
"github.com/filecoin-project/go-fil-markets/storagemarket"
deals "github.com/filecoin-project/go-fil-markets/storagemarket/impl"
storageimpl "github.com/filecoin-project/go-fil-markets/storagemarket/impl"
@@ -109,12 +111,17 @@ func NewClientRequestValidator(deals dtypes.ClientDealStore) *storageimpl.Client
return storageimpl.NewClientRequestValidator(deals)
}
func StorageClient(h host.Host, dag dtypes.ClientDAG, dataTransfer dtypes.ClientDataTransfer, discovery *discovery.Local, deals dtypes.ClientDealStore, scn storagemarket.StorageClientNode) storagemarket.StorageClient {
return storageimpl.NewClient(h, dag, dataTransfer, discovery, deals, scn)
func StorageClient(h host.Host, ibs dtypes.ClientBlockstore, r repo.LockedRepo, dataTransfer dtypes.ClientDataTransfer, discovery *discovery.Local, deals dtypes.ClientDealStore, scn storagemarket.StorageClientNode) (storagemarket.StorageClient, error) {
store, err := piecefilestore.NewLocalFileStore(piecefilestore.OsPath(r.Path()))
if err != nil {
return nil, err
}
return storageimpl.NewClient(h, ibs, store, dataTransfer, discovery, deals, scn), nil
}
// RetrievalClient creates a new retrieval client attached to the client blockstore
func RetrievalClient(h host.Host, bs dtypes.ClientBlockstore, pmgr *paych.Manager, payapi payapi.PaychAPI) retrievalmarket.RetrievalClient {
func RetrievalClient(h host.Host, bs dtypes.ClientBlockstore, pmgr *paych.Manager, payapi payapi.PaychAPI, resolver retrievalmarket.PeerResolver) retrievalmarket.RetrievalClient {
adapter := retrievaladapter.NewRetrievalClientNode(pmgr, payapi)
return retrievalimpl.NewClient(h, bs, adapter)
network := rmnet.NewFromLibp2pHost(h)
return retrievalimpl.NewClient(network, bs, adapter, resolver)
}
node/modules/dtypes/storage.go
+3 -1
View File
@@ -1,6 +1,8 @@
package dtypes
import (
datatransfer "github.com/filecoin-project/go-data-transfer"
"github.com/filecoin-project/go-fil-markets/piecestore"
bserv "github.com/ipfs/go-blockservice"
"github.com/ipfs/go-datastore"
"github.com/ipfs/go-filestore"
@@ -9,7 +11,6 @@ import (
exchange "github.com/ipfs/go-ipfs-exchange-interface"
ipld "github.com/ipfs/go-ipld-format"
"github.com/filecoin-project/go-data-transfer"
"github.com/filecoin-project/go-statestore"
)
@@ -34,6 +35,7 @@ type ClientDealStore *statestore.StateStore
type ClientDataTransfer datatransfer.Manager
type ProviderDealStore *statestore.StateStore
type ProviderPieceStore piecestore.PieceStore
// ProviderDataTransfer is a data transfer manager for the provider
type ProviderDataTransfer datatransfer.Manager
node/modules/storageminer.go
+28 -7
View File
@@ -7,8 +7,11 @@ import (
"github.com/filecoin-project/go-address"
dtgraphsync "github.com/filecoin-project/go-data-transfer/impl/graphsync"
piecefilestore "github.com/filecoin-project/go-fil-markets/filestore"
"github.com/filecoin-project/go-fil-markets/piecestore"
"github.com/filecoin-project/go-fil-markets/retrievalmarket"
retrievalimpl "github.com/filecoin-project/go-fil-markets/retrievalmarket/impl"
rmnet "github.com/filecoin-project/go-fil-markets/retrievalmarket/network"
"github.com/filecoin-project/go-fil-markets/storagemarket"
deals "github.com/filecoin-project/go-fil-markets/storagemarket/impl"
storageimpl "github.com/filecoin-project/go-fil-markets/storagemarket/impl"
@@ -42,7 +45,6 @@ import (
"github.com/filecoin-project/lotus/node/repo"
"github.com/filecoin-project/lotus/storage"
"github.com/filecoin-project/lotus/storage/sealing"
"github.com/filecoin-project/lotus/storage/sectorblocks"
)
func minerAddrFromDS(ds dtypes.MetadataDS) (address.Address, error) {
@@ -124,7 +126,11 @@ func StorageMiner(mctx helpers.MetricsCtx, lc fx.Lifecycle, api api.FullNode, h
func HandleRetrieval(host host.Host, lc fx.Lifecycle, m retrievalmarket.RetrievalProvider) {
lc.Append(fx.Hook{
OnStart: func(context.Context) error {
m.Start(host)
m.Start()
return nil
},
OnStop: func(context.Context) error {
m.Stop()
return nil
},
})
@@ -165,6 +171,12 @@ func NewProviderDealStore(ds dtypes.MetadataDS) dtypes.ProviderDealStore {
return statestore.New(namespace.Wrap(ds, datastore.NewKey("/deals/client")))
}
// NewProviderPieceStore creates a statestore for storing metadata about pieces
// shared by the storage and retrieval providers
func NewProviderPieceStore(ds dtypes.MetadataDS) dtypes.ProviderPieceStore {
return piecestore.NewPieceStore(ds)
}
// StagingBlockstore creates a blockstore for staging blocks for a miner
// in a storage deal, prior to sealing
func StagingBlockstore(r repo.LockedRepo) (dtypes.StagingBlockstore, error) {
@@ -269,12 +281,21 @@ func NewProviderRequestValidator(deals dtypes.ProviderDealStore) *storageimpl.Pr
return storageimpl.NewProviderRequestValidator(deals)
}
func StorageProvider(ds dtypes.MetadataDS, dag dtypes.StagingDAG, dataTransfer dtypes.ProviderDataTransfer, spn storagemarket.StorageProviderNode) (storagemarket.StorageProvider, error) {
return storageimpl.NewProvider(ds, dag, dataTransfer, spn)
func StorageProvider(ds dtypes.MetadataDS, ibs dtypes.StagingBlockstore, r repo.LockedRepo, pieceStore dtypes.ProviderPieceStore, dataTransfer dtypes.ProviderDataTransfer, spn storagemarket.StorageProviderNode) (storagemarket.StorageProvider, error) {
store, err := piecefilestore.NewLocalFileStore(piecefilestore.OsPath(r.Path()))
if err != nil {
return nil, err
}
return storageimpl.NewProvider(ds, ibs, store, pieceStore, dataTransfer, spn)
}
// RetrievalProvider creates a new retrieval provider attached to the provider blockstore
func RetrievalProvider(sblks *sectorblocks.SectorBlocks, full api.FullNode) retrievalmarket.RetrievalProvider {
adapter := retrievaladapter.NewRetrievalProviderNode(sblks, full)
return retrievalimpl.NewProvider(adapter)
func RetrievalProvider(h host.Host, miner *storage.Miner, sb sectorbuilder.Interface, full api.FullNode, ds dtypes.MetadataDS, pieceStore dtypes.ProviderPieceStore, ibs dtypes.StagingBlockstore) (retrievalmarket.RetrievalProvider, error) {
adapter := retrievaladapter.NewRetrievalProviderNode(miner, sb, full)
address, err := minerAddrFromDS(ds)
if err != nil {
return nil, err
}
network := rmnet.NewFromLibp2pHost(h)
return retrievalimpl.NewProvider(address, adapter, network, pieceStore, ibs), nil
}