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

Merge branch 'feat/testnet2' into feat/event-states

Browse Source
This commit is contained in:
Łukasz Magiera
2020-01-13 15:52:43 +01:00
parent f6d41ee77d a896b30039
commit 34abc3d023
144 changed files with 2298 additions and 7110 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/builder.go
+20 -14
View File
@@ -5,6 +5,7 @@ import (
"errors"
"time"
sectorbuilder "github.com/filecoin-project/go-sectorbuilder"
blockstore "github.com/ipfs/go-ipfs-blockstore"
ci "github.com/libp2p/go-libp2p-core/crypto"
"github.com/libp2p/go-libp2p-core/host"
@@ -18,10 +19,13 @@ import (
"go.uber.org/fx"
"golang.org/x/xerrors"
"github.com/filecoin-project/go-fil-markets/retrievalmarket"
"github.com/filecoin-project/go-fil-markets/retrievalmarket/discovery"
"github.com/filecoin-project/go-fil-markets/storagemarket"
deals "github.com/filecoin-project/go-fil-markets/storagemarket/impl"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/chain"
"github.com/filecoin-project/lotus/chain/blocksync"
"github.com/filecoin-project/lotus/chain/deals"
"github.com/filecoin-project/lotus/chain/gen"
"github.com/filecoin-project/lotus/chain/market"
"github.com/filecoin-project/lotus/chain/messagepool"
@@ -30,7 +34,7 @@ import (
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/chain/wallet"
"github.com/filecoin-project/lotus/lib/sectorbuilder"
"github.com/filecoin-project/lotus/markets/storageadapter"
"github.com/filecoin-project/lotus/miner"
"github.com/filecoin-project/lotus/node/config"
"github.com/filecoin-project/lotus/node/hello"
@@ -43,8 +47,6 @@ import (
"github.com/filecoin-project/lotus/node/repo"
"github.com/filecoin-project/lotus/paych"
"github.com/filecoin-project/lotus/peermgr"
"github.com/filecoin-project/lotus/retrieval"
"github.com/filecoin-project/lotus/retrieval/discovery"
"github.com/filecoin-project/lotus/storage"
"github.com/filecoin-project/lotus/storage/sectorblocks"
)
@@ -220,14 +222,15 @@ func Online() Option {
Override(RunPeerMgrKey, modules.RunPeerMgr),
Override(HandleIncomingBlocksKey, modules.HandleIncomingBlocks),
Override(new(*discovery.Local), discovery.NewLocal),
Override(new(discovery.PeerResolver), modules.RetrievalResolver),
Override(new(*discovery.Local), modules.NewLocalDiscovery),
Override(new(retrievalmarket.PeerResolver), modules.RetrievalResolver),
Override(new(*retrieval.Client), retrieval.NewClient),
Override(new(retrievalmarket.RetrievalClient), modules.RetrievalClient),
Override(new(dtypes.ClientDealStore), modules.NewClientDealStore),
Override(new(dtypes.ClientDataTransfer), modules.NewClientDAGServiceDataTransfer),
Override(new(*deals.ClientRequestValidator), deals.NewClientRequestValidator),
Override(new(*deals.Client), deals.NewClient),
Override(new(*deals.ClientRequestValidator), modules.NewClientRequestValidator),
Override(new(storagemarket.StorageClient), modules.StorageClient),
Override(new(storagemarket.StorageClientNode), storageadapter.NewClientNodeAdapter),
Override(RegisterClientValidatorKey, modules.RegisterClientValidator),
Override(RunDealClientKey, modules.RunDealClient),
@@ -238,18 +241,20 @@ func Online() Option {
// Storage miner
ApplyIf(func(s *Settings) bool { return s.nodeType == repo.StorageMiner },
Override(new(*sectorbuilder.SectorBuilder), modules.SectorBuilder),
Override(new(storage.SectorBuilder), modules.SectorBuilder),
Override(new(*sectorblocks.SectorBlocks), sectorblocks.NewSectorBlocks),
Override(new(storage.TicketFn), modules.SealTicketGen),
Override(new(*storage.Miner), modules.StorageMiner),
Override(new(dtypes.StagingBlockstore), modules.StagingBlockstore),
Override(new(dtypes.StagingDAG), modules.StagingDAG),
Override(new(*retrieval.Miner), retrieval.NewMiner),
Override(new(dtypes.StagingGraphsync), modules.StagingGraphsync),
Override(new(retrievalmarket.RetrievalProvider), modules.RetrievalProvider),
Override(new(dtypes.ProviderDealStore), modules.NewProviderDealStore),
Override(new(dtypes.ProviderDataTransfer), modules.NewProviderDAGServiceDataTransfer),
Override(new(*deals.ProviderRequestValidator), deals.NewProviderRequestValidator),
Override(new(*deals.Provider), deals.NewProvider),
Override(new(*deals.ProviderRequestValidator), modules.NewProviderRequestValidator),
Override(new(storagemarket.StorageProvider), modules.StorageProvider),
Override(new(storagemarket.StorageProviderNode), storageadapter.NewProviderNodeAdapter),
Override(RegisterProviderValidatorKey, modules.RegisterProviderValidator),
Override(HandleRetrievalKey, modules.HandleRetrieval),
Override(GetParamsKey, modules.GetParams),
@@ -372,6 +377,7 @@ func Repo(r repo.Repo) Option {
Override(new(dtypes.ClientFilestore), modules.ClientFstore),
Override(new(dtypes.ClientBlockstore), modules.ClientBlockstore),
Override(new(dtypes.ClientDAG), modules.ClientDAG),
Override(new(dtypes.ClientGraphsync), modules.ClientGraphsync),
Override(new(ci.PrivKey), lp2p.PrivKey),
Override(new(ci.PubKey), ci.PrivKey.GetPublic),
node/fxlog.go
+1 -1
View File
@@ -1,7 +1,7 @@
package node
import (
logging "github.com/ipfs/go-log"
logging "github.com/ipfs/go-log/v2"
"go.uber.org/fx"
)
node/hello/hello.go
+1 -1
View File
@@ -6,7 +6,7 @@ import (
"github.com/ipfs/go-cid"
cbor "github.com/ipfs/go-ipld-cbor"
logging "github.com/ipfs/go-log"
logging "github.com/ipfs/go-log/v2"
"github.com/libp2p/go-libp2p-core/host"
inet "github.com/libp2p/go-libp2p-core/network"
"github.com/libp2p/go-libp2p-core/peer"
node/impl/client/client.go
+81 -34
View File
@@ -1,6 +1,7 @@
package client
import (
"bytes"
"context"
"errors"
"io"
@@ -17,22 +18,24 @@ import (
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/libp2p/go-libp2p-core/peer"
"go.uber.org/fx"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/go-fil-markets/retrievalmarket"
"github.com/filecoin-project/go-fil-markets/shared/tokenamount"
"github.com/filecoin-project/go-fil-markets/storagemarket"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/deals"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/markets/utils"
"github.com/filecoin-project/lotus/node/impl/full"
"github.com/filecoin-project/lotus/node/impl/paych"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/retrieval"
"github.com/filecoin-project/lotus/retrieval/discovery"
)
type API struct {
@@ -43,9 +46,9 @@ type API struct {
full.WalletAPI
paych.PaychAPI
DealClient *deals.Client
RetDiscovery discovery.PeerResolver
Retrieval *retrieval.Client
SMDealClient storagemarket.StorageClient
RetDiscovery retrievalmarket.PeerResolver
Retrieval retrievalmarket.RetrievalClient
Chain *store.ChainStore
LocalDAG dtypes.ClientDAG
@@ -71,28 +74,26 @@ func (a *API) ClientStartDeal(ctx context.Context, data cid.Cid, addr address.Ad
if err != nil {
return nil, xerrors.Errorf("failed getting miner worker: %w", err)
}
providerInfo := utils.NewStorageProviderInfo(miner, mw, 0, pid)
result, err := a.SMDealClient.ProposeStorageDeal(
ctx,
addr,
&providerInfo,
data,
storagemarket.Epoch(math.MaxUint64),
storagemarket.Epoch(blocksDuration),
utils.ToSharedTokenAmount(epochPrice),
tokenamount.Empty)
proposal := deals.ClientDealProposal{
Data: data,
PricePerEpoch: epochPrice,
ProposalExpiration: math.MaxUint64, // TODO: set something reasonable
Duration: blocksDuration,
Client: addr,
ProviderAddress: miner,
MinerWorker: mw,
MinerID: pid,
}
c, err := a.DealClient.Start(ctx, proposal)
if err != nil {
return nil, xerrors.Errorf("failed to start deal: %w", err)
}
return &c, nil
return &result.ProposalCid, nil
}
func (a *API) ClientListDeals(ctx context.Context) ([]api.DealInfo, error) {
deals, err := a.DealClient.List()
deals, err := a.SMDealClient.ListInProgressDeals(ctx)
if err != nil {
return nil, err
}
@@ -107,7 +108,7 @@ func (a *API) ClientListDeals(ctx context.Context) ([]api.DealInfo, error) {
PieceRef: v.Proposal.PieceRef,
Size: v.Proposal.PieceSize,
PricePerEpoch: v.Proposal.StoragePricePerEpoch,
PricePerEpoch: utils.FromSharedTokenAmount(v.Proposal.StoragePricePerEpoch),
Duration: v.Proposal.Duration,
}
}
@@ -116,17 +117,18 @@ func (a *API) ClientListDeals(ctx context.Context) ([]api.DealInfo, error) {
}
func (a *API) ClientGetDealInfo(ctx context.Context, d cid.Cid) (*api.DealInfo, error) {
v, err := a.DealClient.GetDeal(d)
v, err := a.SMDealClient.GetInProgressDeal(ctx, d)
if err != nil {
return nil, err
}
return &api.DealInfo{
ProposalCid: v.ProposalCid,
State: v.State,
Provider: v.Proposal.Provider,
PieceRef: v.Proposal.PieceRef,
Size: v.Proposal.PieceSize,
PricePerEpoch: v.Proposal.StoragePricePerEpoch,
PricePerEpoch: utils.FromSharedTokenAmount(v.Proposal.StoragePricePerEpoch),
Duration: v.Proposal.Duration,
}, nil
}
@@ -153,7 +155,18 @@ func (a *API) ClientFindData(ctx context.Context, root cid.Cid) ([]api.QueryOffe
out := make([]api.QueryOffer, len(peers))
for k, p := range peers {
out[k] = a.Retrieval.Query(ctx, p, root)
queryResponse, err := a.Retrieval.Query(ctx, p, root.Bytes(), retrievalmarket.QueryParams{})
if err != nil {
out[k] = api.QueryOffer{Err: err.Error(), Miner: p.Address, MinerPeerID: p.ID}
} else {
out[k] = api.QueryOffer{
Root: root,
Size: queryResponse.Size,
MinPrice: utils.FromSharedTokenAmount(queryResponse.PieceRetrievalPrice()),
Miner: p.Address, // TODO: check
MinerPeerID: p.ID,
}
}
}
return out, nil
@@ -263,20 +276,54 @@ func (a *API) ClientRetrieve(ctx context.Context, order api.RetrievalOrder, path
order.MinerPeerID = pid
}
outFile, err := os.OpenFile(path, os.O_TRUNC|os.O_CREATE|os.O_WRONLY, 0777)
if err != nil {
return err
retrievalResult := make(chan error, 1)
unsubscribe := a.Retrieval.SubscribeToEvents(func(event retrievalmarket.ClientEvent, state retrievalmarket.ClientDealState) {
if bytes.Equal(state.PieceCID, order.Root.Bytes()) {
switch event {
case retrievalmarket.ClientEventError:
retrievalResult <- xerrors.New("Retrieval Error")
case retrievalmarket.ClientEventComplete:
retrievalResult <- nil
}
}
})
a.Retrieval.Retrieve(
ctx,
order.Root.Bytes(),
retrievalmarket.NewParamsV0(types.BigDiv(order.Total, types.NewInt(order.Size)).Int, 0, 0),
utils.ToSharedTokenAmount(order.Total),
order.MinerPeerID,
order.Client,
order.Miner)
select {
case <-ctx.Done():
return xerrors.New("Retrieval Timed Out")
case err := <-retrievalResult:
if err != nil {
return xerrors.Errorf("RetrieveUnixfs: %w", err)
}
}
err = a.Retrieval.RetrieveUnixfs(ctx, order.Root, order.Size, order.Total, order.MinerPeerID, order.Client, order.Miner, outFile)
if err != nil {
_ = outFile.Close()
return xerrors.Errorf("RetrieveUnixfs: %w", err)
}
unsubscribe()
return outFile.Close()
nd, err := a.LocalDAG.Get(ctx, order.Root)
if err != nil {
return xerrors.Errorf("ClientRetrieve: %w", err)
}
file, err := unixfile.NewUnixfsFile(ctx, a.LocalDAG, nd)
if err != nil {
return xerrors.Errorf("ClientRetrieve: %w", err)
}
return files.WriteTo(file, path)
}
func (a *API) ClientQueryAsk(ctx context.Context, p peer.ID, miner address.Address) (*types.SignedStorageAsk, error) {
return a.DealClient.QueryAsk(ctx, p, miner)
info := utils.NewStorageProviderInfo(miner, address.Undef, 0, p)
signedAsk, err := a.SMDealClient.GetAsk(ctx, info)
if err != nil {
return nil, err
}
return utils.FromSignedStorageAsk(signedAsk)
}
node/impl/full.go
+1 -1
View File
@@ -1,7 +1,7 @@
package impl
import (
logging "github.com/ipfs/go-log"
logging "github.com/ipfs/go-log/v2"
"github.com/filecoin-project/lotus/node/impl/client"
"github.com/filecoin-project/lotus/node/impl/market"
node/impl/full/chain.go
+9
View File
@@ -299,3 +299,12 @@ func (a *ChainAPI) ChainGetNode(ctx context.Context, p string) (interface{}, err
return node, nil
}
func (a *ChainAPI) ChainGetMessage(ctx context.Context, mc cid.Cid) (*types.Message, error) {
cm, err := a.Chain.GetCMessage(mc)
if err != nil {
return nil, err
}
return cm.VMMessage(), nil
}
node/impl/full/mpool.go
+1 -1
View File
@@ -78,7 +78,7 @@ func (a *MpoolAPI) MpoolPending(ctx context.Context, ts *types.TipSet) ([]*types
}
}
func (a *MpoolAPI) MpoolPush(ctx context.Context, smsg *types.SignedMessage) error {
func (a *MpoolAPI) MpoolPush(ctx context.Context, smsg *types.SignedMessage) (cid.Cid, error) {
return a.Mpool.Push(smsg)
}
node/impl/full/state.go
+49
View File
@@ -351,3 +351,52 @@ func (a *StateAPI) StateChangedActors(ctx context.Context, old cid.Cid, new cid.
func (a *StateAPI) StateMinerSectorCount(ctx context.Context, addr address.Address, ts *types.TipSet) (api.MinerSectors, error) {
return stmgr.SectorSetSizes(ctx, a.StateManager, addr, ts)
}
func (a *StateAPI) StateListMessages(ctx context.Context, match *types.Message, ts *types.TipSet, toheight uint64) ([]cid.Cid, error) {
if ts == nil {
ts = a.Chain.GetHeaviestTipSet()
}
if match.To == address.Undef && match.From == address.Undef {
return nil, xerrors.Errorf("must specify at least To or From in message filter")
}
matchFunc := func(msg *types.Message) bool {
if match.From != address.Undef && match.From != msg.From {
return false
}
if match.To != address.Undef && match.To != msg.To {
return false
}
return true
}
var out []cid.Cid
for ts.Height() >= toheight {
msgs, err := a.Chain.MessagesForTipset(ts)
if err != nil {
return nil, xerrors.Errorf("failed to get messages for tipset (%s): %w", ts.Key(), err)
}
for _, msg := range msgs {
if matchFunc(msg.VMMessage()) {
out = append(out, msg.Cid())
}
}
if ts.Height() == 0 {
break
}
next, err := a.Chain.LoadTipSet(ts.Parents())
if err != nil {
return nil, xerrors.Errorf("loading next tipset: %w", err)
}
ts = next
}
return out, nil
}
node/impl/full/sync.go
+8
View File
@@ -6,6 +6,8 @@ import (
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/chain"
"github.com/filecoin-project/lotus/chain/types"
cid "github.com/ipfs/go-cid"
"github.com/prometheus/common/log"
pubsub "github.com/libp2p/go-libp2p-pubsub"
"go.uber.org/fx"
@@ -80,3 +82,9 @@ func (a *SyncAPI) SyncSubmitBlock(ctx context.Context, blk *types.BlockMsg) erro
func (a *SyncAPI) SyncIncomingBlocks(ctx context.Context) (<-chan *types.BlockHeader, error) {
return a.Syncer.IncomingBlocks(ctx)
}
func (a *SyncAPI) SyncMarkBad(ctx context.Context, bcid cid.Cid) error {
log.Warnf("Marking block %s as bad", bcid)
a.Syncer.MarkBad(bcid)
return nil
}
node/impl/paych/paych.go
+2 -2
View File
@@ -129,7 +129,7 @@ func (a *PaychAPI) PaychClose(ctx context.Context, addr address.Address) (cid.Ci
return cid.Undef, err
}
if err := a.MpoolPush(ctx, smsg); err != nil {
if _, err := a.MpoolPush(ctx, smsg); err != nil {
return cid.Undef, err
}
@@ -244,7 +244,7 @@ func (a *PaychAPI) PaychVoucherSubmit(ctx context.Context, ch address.Address, s
return cid.Undef, err
}
if err := a.MpoolPush(ctx, smsg); err != nil {
if _, err := a.MpoolPush(ctx, smsg); err != nil {
return cid.Undef, err
}
node/impl/storminer.go
+4 -3
View File
@@ -3,18 +3,19 @@ package impl
import (
"context"
"encoding/json"
"github.com/filecoin-project/lotus/api/apistruct"
"io"
"mime"
"net/http"
"os"
"github.com/filecoin-project/lotus/api/apistruct"
"github.com/gorilla/mux"
files "github.com/ipfs/go-ipfs-files"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/go-sectorbuilder"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/lib/sectorbuilder"
"github.com/filecoin-project/lotus/lib/tarutil"
"github.com/filecoin-project/lotus/miner"
"github.com/filecoin-project/lotus/storage"
@@ -25,7 +26,7 @@ type StorageMinerAPI struct {
CommonAPI
SectorBuilderConfig *sectorbuilder.Config
SectorBuilder *sectorbuilder.SectorBuilder
SectorBuilder storage.SectorBuilder
SectorBlocks *sectorblocks.SectorBlocks
Miner *storage.Miner
node/modules/client.go
+48 -6
View File
@@ -5,10 +5,23 @@ import (
"path/filepath"
"reflect"
"github.com/filecoin-project/lotus/lib/statestore"
"github.com/filecoin-project/lotus/markets/retrievaladapter"
"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"
"github.com/filecoin-project/go-fil-markets/storagemarket"
"github.com/filecoin-project/go-statestore"
"github.com/filecoin-project/lotus/node/modules/helpers"
"github.com/filecoin-project/lotus/paych"
"github.com/ipfs/go-bitswap"
"github.com/ipfs/go-bitswap/network"
graphsync "github.com/ipfs/go-graphsync/impl"
"github.com/ipfs/go-graphsync/ipldbridge"
gsnet "github.com/ipfs/go-graphsync/network"
"github.com/ipfs/go-graphsync/storeutil"
"github.com/libp2p/go-libp2p-core/host"
"github.com/libp2p/go-libp2p-core/routing"
@@ -20,8 +33,10 @@ import (
"github.com/ipfs/go-merkledag"
"go.uber.org/fx"
"github.com/filecoin-project/lotus/chain/deals"
"github.com/filecoin-project/lotus/datatransfer"
"github.com/filecoin-project/go-data-transfer/impl/graphsync"
deals "github.com/filecoin-project/go-fil-markets/storagemarket/impl"
storageimpl "github.com/filecoin-project/go-fil-markets/storagemarket/impl"
payapi "github.com/filecoin-project/lotus/node/impl/paych"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/repo"
)
@@ -49,15 +64,15 @@ func ClientBlockstore(fstore dtypes.ClientFilestore) dtypes.ClientBlockstore {
// request validator with the data transfer module as the validator for
// StorageDataTransferVoucher types
func RegisterClientValidator(crv *deals.ClientRequestValidator, dtm dtypes.ClientDataTransfer) {
if err := dtm.RegisterVoucherType(reflect.TypeOf(deals.StorageDataTransferVoucher{}), crv); err != nil {
if err := dtm.RegisterVoucherType(reflect.TypeOf(&deals.StorageDataTransferVoucher{}), crv); err != nil {
panic(err)
}
}
// NewClientDAGServiceDataTransfer returns a data transfer manager that just
// uses the clients's Client DAG service for transfers
func NewClientDAGServiceDataTransfer(dag dtypes.ClientDAG) dtypes.ClientDataTransfer {
return datatransfer.NewDAGServiceDataTransfer(dag)
func NewClientDAGServiceDataTransfer(h host.Host, gs dtypes.ClientGraphsync) dtypes.ClientDataTransfer {
return graphsyncimpl.NewGraphSyncDataTransfer(h, gs)
}
// NewClientDealStore creates a statestore for the client to store its deals
@@ -65,6 +80,7 @@ func NewClientDealStore(ds dtypes.MetadataDS) dtypes.ClientDealStore {
return statestore.New(namespace.Wrap(ds, datastore.NewKey("/deals/client")))
}
// ClientDAG is a DAGService for the ClientBlockstore
func ClientDAG(mctx helpers.MetricsCtx, lc fx.Lifecycle, ibs dtypes.ClientBlockstore, rt routing.Routing, h host.Host) dtypes.ClientDAG {
bitswapNetwork := network.NewFromIpfsHost(h, rt)
exch := bitswap.New(helpers.LifecycleCtx(mctx, lc), bitswapNetwork, ibs)
@@ -80,3 +96,29 @@ func ClientDAG(mctx helpers.MetricsCtx, lc fx.Lifecycle, ibs dtypes.ClientBlocks
return dag
}
// ClientGraphsync creates a graphsync instance which reads and writes blocks
// to the ClientBlockstore
func ClientGraphsync(mctx helpers.MetricsCtx, lc fx.Lifecycle, ibs dtypes.ClientBlockstore, h host.Host) dtypes.ClientGraphsync {
graphsyncNetwork := gsnet.NewFromLibp2pHost(h)
ipldBridge := ipldbridge.NewIPLDBridge()
loader := storeutil.LoaderForBlockstore(ibs)
storer := storeutil.StorerForBlockstore(ibs)
gs := graphsync.New(helpers.LifecycleCtx(mctx, lc), graphsyncNetwork, ipldBridge, loader, storer)
return gs
}
func NewClientRequestValidator(deals dtypes.ClientDealStore) *storageimpl.ClientRequestValidator {
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)
}
// 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 {
adapter := retrievaladapter.NewRetrievalClientNode(pmgr, payapi)
return retrievalimpl.NewClient(h, bs, adapter)
}
node/modules/core.go
+1 -1
View File
@@ -13,7 +13,7 @@ import (
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/repo"
"github.com/gbrlsnchs/jwt/v3"
logging "github.com/ipfs/go-log"
logging "github.com/ipfs/go-log/v2"
"github.com/libp2p/go-libp2p-core/peerstore"
record "github.com/libp2p/go-libp2p-record"
"golang.org/x/xerrors"
node/modules/dtypes/storage.go
+6 -2
View File
@@ -4,12 +4,13 @@ import (
bserv "github.com/ipfs/go-blockservice"
"github.com/ipfs/go-datastore"
"github.com/ipfs/go-filestore"
"github.com/ipfs/go-graphsync"
blockstore "github.com/ipfs/go-ipfs-blockstore"
exchange "github.com/ipfs/go-ipfs-exchange-interface"
ipld "github.com/ipfs/go-ipld-format"
"github.com/filecoin-project/lotus/datatransfer"
"github.com/filecoin-project/lotus/lib/statestore"
"github.com/filecoin-project/go-data-transfer"
"github.com/filecoin-project/go-statestore"
)
// MetadataDS stores metadata
@@ -26,6 +27,7 @@ type ChainBlockService bserv.BlockService
type ClientFilestore *filestore.Filestore
type ClientBlockstore blockstore.Blockstore
type ClientDAG ipld.DAGService
type ClientGraphsync graphsync.GraphExchange
type ClientDealStore *statestore.StateStore
// ClientDataTransfer is a data transfer manager for the client
@@ -37,3 +39,5 @@ type ProviderDealStore *statestore.StateStore
type ProviderDataTransfer datatransfer.Manager
type StagingDAG ipld.DAGService
type StagingBlockstore blockstore.Blockstore
type StagingGraphsync graphsync.GraphExchange
node/modules/lp2p/libp2p.go
+1 -1
View File
@@ -8,7 +8,7 @@ import (
"github.com/filecoin-project/lotus/chain/types"
"golang.org/x/xerrors"
logging "github.com/ipfs/go-log"
logging "github.com/ipfs/go-log/v2"
"github.com/libp2p/go-libp2p"
connmgr "github.com/libp2p/go-libp2p-connmgr"
"github.com/libp2p/go-libp2p-core/crypto"
node/modules/services.go
+10 -4
View File
@@ -8,15 +8,17 @@ import (
pubsub "github.com/libp2p/go-libp2p-pubsub"
"go.uber.org/fx"
"github.com/filecoin-project/go-fil-markets/retrievalmarket"
"github.com/filecoin-project/go-fil-markets/retrievalmarket/discovery"
"github.com/filecoin-project/go-fil-markets/storagemarket"
"github.com/filecoin-project/lotus/chain"
"github.com/filecoin-project/lotus/chain/blocksync"
"github.com/filecoin-project/lotus/chain/deals"
"github.com/filecoin-project/lotus/chain/messagepool"
"github.com/filecoin-project/lotus/chain/sub"
"github.com/filecoin-project/lotus/node/hello"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/modules/helpers"
"github.com/filecoin-project/lotus/peermgr"
"github.com/filecoin-project/lotus/retrieval/discovery"
)
func RunHello(mctx helpers.MetricsCtx, lc fx.Lifecycle, h host.Host, svc *hello.Service) {
@@ -65,7 +67,7 @@ func HandleIncomingMessages(mctx helpers.MetricsCtx, lc fx.Lifecycle, pubsub *pu
go sub.HandleIncomingMessages(ctx, mpool, msgsub)
}
func RunDealClient(mctx helpers.MetricsCtx, lc fx.Lifecycle, c *deals.Client) {
func RunDealClient(mctx helpers.MetricsCtx, lc fx.Lifecycle, c storagemarket.StorageClient) {
ctx := helpers.LifecycleCtx(mctx, lc)
lc.Append(fx.Hook{
@@ -80,6 +82,10 @@ func RunDealClient(mctx helpers.MetricsCtx, lc fx.Lifecycle, c *deals.Client) {
})
}
func RetrievalResolver(l *discovery.Local) discovery.PeerResolver {
func NewLocalDiscovery(ds dtypes.MetadataDS) *discovery.Local {
return discovery.NewLocal(ds)
}
func RetrievalResolver(l *discovery.Local) retrievalmarket.PeerResolver {
return discovery.Multi(l)
}
node/modules/storageminer.go
+61 -20
View File
@@ -11,6 +11,10 @@ import (
"github.com/ipfs/go-blockservice"
"github.com/ipfs/go-datastore"
"github.com/ipfs/go-datastore/namespace"
graphsync "github.com/ipfs/go-graphsync/impl"
"github.com/ipfs/go-graphsync/ipldbridge"
gsnet "github.com/ipfs/go-graphsync/network"
"github.com/ipfs/go-graphsync/storeutil"
blockstore "github.com/ipfs/go-ipfs-blockstore"
"github.com/ipfs/go-merkledag"
"github.com/libp2p/go-libp2p-core/host"
@@ -20,19 +24,25 @@ import (
"golang.org/x/xerrors"
"github.com/filecoin-project/go-address"
dtgraphsync "github.com/filecoin-project/go-data-transfer/impl/graphsync"
"github.com/filecoin-project/go-fil-markets/retrievalmarket"
retrievalimpl "github.com/filecoin-project/go-fil-markets/retrievalmarket/impl"
"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"
"github.com/filecoin-project/go-sectorbuilder"
"github.com/filecoin-project/go-statestore"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/deals"
"github.com/filecoin-project/lotus/chain/gen"
"github.com/filecoin-project/lotus/datatransfer"
"github.com/filecoin-project/lotus/lib/sectorbuilder"
"github.com/filecoin-project/lotus/lib/statestore"
"github.com/filecoin-project/lotus/miner"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/modules/helpers"
"github.com/filecoin-project/lotus/node/repo"
"github.com/filecoin-project/lotus/retrieval"
"github.com/filecoin-project/lotus/markets/retrievaladapter"
"github.com/filecoin-project/lotus/storage"
"github.com/filecoin-project/lotus/storage/sectorblocks"
)
func minerAddrFromDS(ds dtypes.MetadataDS) (address.Address, error) {
@@ -45,7 +55,7 @@ func minerAddrFromDS(ds dtypes.MetadataDS) (address.Address, error) {
}
func GetParams(sbc *sectorbuilder.Config) error {
if err := paramfetch.GetParams(sbc.SectorSize); err != nil {
if err := paramfetch.GetParams(build.ParametersJson, sbc.SectorSize); err != nil {
return xerrors.Errorf("fetching proof parameters: %w", err)
}
@@ -88,7 +98,7 @@ func SectorBuilderConfig(storagePath string, threads uint, noprecommit, nocommit
}
}
func StorageMiner(mctx helpers.MetricsCtx, lc fx.Lifecycle, api api.FullNode, h host.Host, ds dtypes.MetadataDS, sb *sectorbuilder.SectorBuilder, tktFn storage.TicketFn) (*storage.Miner, error) {
func StorageMiner(mctx helpers.MetricsCtx, lc fx.Lifecycle, api api.FullNode, h host.Host, ds dtypes.MetadataDS, sb storage.SectorBuilder, tktFn storage.TicketFn) (*storage.Miner, error) {
maddr, err := minerAddrFromDS(ds)
if err != nil {
return nil, err
@@ -111,24 +121,21 @@ func StorageMiner(mctx helpers.MetricsCtx, lc fx.Lifecycle, api api.FullNode, h
return sm, nil
}
func HandleRetrieval(host host.Host, lc fx.Lifecycle, m *retrieval.Miner) {
func HandleRetrieval(host host.Host, lc fx.Lifecycle, m retrievalmarket.RetrievalProvider) {
lc.Append(fx.Hook{
OnStart: func(context.Context) error {
host.SetStreamHandler(retrieval.QueryProtocolID, m.HandleQueryStream)
host.SetStreamHandler(retrieval.ProtocolID, m.HandleDealStream)
m.Start(host)
return nil
},
})
}
func HandleDeals(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host, h *deals.Provider) {
func HandleDeals(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host, h storagemarket.StorageProvider) {
ctx := helpers.LifecycleCtx(mctx, lc)
lc.Append(fx.Hook{
OnStart: func(context.Context) error {
h.Run(ctx)
host.SetStreamHandler(deals.DealProtocolID, h.HandleStream)
host.SetStreamHandler(deals.AskProtocolID, h.HandleAskStream)
h.Run(ctx, host)
return nil
},
OnStop: func(context.Context) error {
@@ -142,15 +149,15 @@ func HandleDeals(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host, h *de
// request validator with the data transfer module as the validator for
// StorageDataTransferVoucher types
func RegisterProviderValidator(mrv *deals.ProviderRequestValidator, dtm dtypes.ProviderDataTransfer) {
if err := dtm.RegisterVoucherType(reflect.TypeOf(deals.StorageDataTransferVoucher{}), mrv); err != nil {
if err := dtm.RegisterVoucherType(reflect.TypeOf(&deals.StorageDataTransferVoucher{}), mrv); err != nil {
panic(err)
}
}
// NewProviderDAGServiceDataTransfer returns a data transfer manager that just
// uses the provider's Staging DAG service for transfers
func NewProviderDAGServiceDataTransfer(dag dtypes.StagingDAG) dtypes.ProviderDataTransfer {
return datatransfer.NewDAGServiceDataTransfer(dag)
func NewProviderDAGServiceDataTransfer(h host.Host, gs dtypes.StagingGraphsync) dtypes.ProviderDataTransfer {
return dtgraphsync.NewGraphSyncDataTransfer(h, gs)
}
// NewProviderDealStore creates a statestore for the client to store its deals
@@ -158,7 +165,9 @@ func NewProviderDealStore(ds dtypes.MetadataDS) dtypes.ProviderDealStore {
return statestore.New(namespace.Wrap(ds, datastore.NewKey("/deals/client")))
}
func StagingDAG(mctx helpers.MetricsCtx, lc fx.Lifecycle, r repo.LockedRepo, rt routing.Routing, h host.Host) (dtypes.StagingDAG, error) {
// 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) {
stagingds, err := r.Datastore("/staging")
if err != nil {
return nil, err
@@ -167,8 +176,14 @@ func StagingDAG(mctx helpers.MetricsCtx, lc fx.Lifecycle, r repo.LockedRepo, rt
bs := blockstore.NewBlockstore(stagingds)
ibs := blockstore.NewIdStore(bs)
return ibs, nil
}
// StagingDAG is a DAGService for the StagingBlockstore
func StagingDAG(mctx helpers.MetricsCtx, lc fx.Lifecycle, ibs dtypes.StagingBlockstore, rt routing.Routing, h host.Host) (dtypes.StagingDAG, error) {
bitswapNetwork := network.NewFromIpfsHost(h, rt)
exch := bitswap.New(helpers.LifecycleCtx(mctx, lc), bitswapNetwork, bs)
exch := bitswap.New(helpers.LifecycleCtx(mctx, lc), bitswapNetwork, ibs)
bsvc := blockservice.New(ibs, exch)
dag := merkledag.NewDAGService(bsvc)
@@ -182,6 +197,18 @@ func StagingDAG(mctx helpers.MetricsCtx, lc fx.Lifecycle, r repo.LockedRepo, rt
return dag, nil
}
// StagingGraphsync creates a graphsync instance which reads and writes blocks
// to the StagingBlockstore
func StagingGraphsync(mctx helpers.MetricsCtx, lc fx.Lifecycle, ibs dtypes.StagingBlockstore, h host.Host) dtypes.StagingGraphsync {
graphsyncNetwork := gsnet.NewFromLibp2pHost(h)
ipldBridge := ipldbridge.NewIPLDBridge()
loader := storeutil.LoaderForBlockstore(ibs)
storer := storeutil.StorerForBlockstore(ibs)
gs := graphsync.New(helpers.LifecycleCtx(mctx, lc), graphsyncNetwork, ipldBridge, loader, storer)
return gs
}
func SetupBlockProducer(lc fx.Lifecycle, ds dtypes.MetadataDS, api api.FullNode, epp gen.ElectionPoStProver) (*miner.Miner, error) {
minerAddr, err := minerAddrFromDS(ds)
if err != nil {
@@ -206,7 +233,7 @@ func SetupBlockProducer(lc fx.Lifecycle, ds dtypes.MetadataDS, api api.FullNode,
}
func SectorBuilder(cfg *sectorbuilder.Config, ds dtypes.MetadataDS) (*sectorbuilder.SectorBuilder, error) {
sb, err := sectorbuilder.New(cfg, ds)
sb, err := sectorbuilder.New(cfg, namespace.Wrap(ds, datastore.NewKey("/sectorbuilder")))
if err != nil {
return nil, err
}
@@ -237,3 +264,17 @@ func SealTicketGen(api api.FullNode) storage.TicketFn {
}, nil
}
}
func NewProviderRequestValidator(deals dtypes.ProviderDealStore) *storageimpl.ProviderRequestValidator {
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)
}
// 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)
}
node/modules/testing/genesis.go
+1 -1
View File
@@ -13,7 +13,7 @@ import (
"github.com/ipfs/go-car"
"github.com/ipfs/go-cid"
offline "github.com/ipfs/go-ipfs-exchange-offline"
logging "github.com/ipfs/go-log"
logging "github.com/ipfs/go-log/v2"
"github.com/ipfs/go-merkledag"
"github.com/libp2p/go-libp2p-core/peer"
"github.com/mitchellh/go-homedir"
node/node_test.go
+8 -9
View File
@@ -9,18 +9,18 @@ import (
"path/filepath"
"testing"
"github.com/filecoin-project/go-address"
sectorbuilder "github.com/filecoin-project/go-sectorbuilder"
"github.com/filecoin-project/lotus/build"
"github.com/libp2p/go-libp2p-core/crypto"
"github.com/ipfs/go-datastore"
badger "github.com/ipfs/go-ds-badger"
logging "github.com/ipfs/go-log"
"github.com/ipfs/go-datastore/namespace"
badger "github.com/ipfs/go-ds-badger2"
logging "github.com/ipfs/go-log/v2"
"github.com/libp2p/go-libp2p-core/crypto"
"github.com/libp2p/go-libp2p-core/peer"
mocknet "github.com/libp2p/go-libp2p/p2p/net/mock"
"github.com/stretchr/testify/require"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/api/client"
"github.com/filecoin-project/lotus/api/test"
@@ -30,7 +30,6 @@ import (
"github.com/filecoin-project/lotus/cmd/lotus-seed/seed"
"github.com/filecoin-project/lotus/genesis"
"github.com/filecoin-project/lotus/lib/jsonrpc"
"github.com/filecoin-project/lotus/lib/sectorbuilder"
"github.com/filecoin-project/lotus/miner"
"github.com/filecoin-project/lotus/node"
"github.com/filecoin-project/lotus/node/impl"
@@ -230,12 +229,12 @@ func builder(t *testing.T, nFull int, storage []int) ([]test.TestNode, []test.Te
WorkerThreads: 2,
Miner: genMiner,
Dir: psd,
}, mds)
}, namespace.Wrap(mds, datastore.NewKey("/sectorbuilder")))
if err != nil {
t.Fatal(err)
}
if err := sma.SectorBuilder.ImportFrom(osb, false); err != nil {
if err := sma.SectorBuilder.(*sectorbuilder.SectorBuilder).ImportFrom(osb, false); err != nil {
t.Fatal(err)
}
node/repo/fsrepo.go
+2 -2
View File
@@ -12,9 +12,9 @@ import (
"github.com/ipfs/go-datastore"
"github.com/ipfs/go-datastore/namespace"
badger "github.com/ipfs/go-ds-badger"
badger "github.com/ipfs/go-ds-badger2"
fslock "github.com/ipfs/go-fs-lock"
logging "github.com/ipfs/go-log"
logging "github.com/ipfs/go-log/v2"
"github.com/mitchellh/go-homedir"
"github.com/multiformats/go-base32"
"github.com/multiformats/go-multiaddr"