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

Merge branch 'master' into nonsense/split-market-miner-processes

Browse Source
This commit is contained in:
Anton Evangelatov
2021-05-31 11:26:11 +02:00
parent 46a368d2b9 be3a8bd320
commit d195a12565
192 changed files with 5357 additions and 1492 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/impl/client/client.go
+131 -20
View File
@@ -6,6 +6,7 @@ import (
"fmt"
"io"
"os"
"time"
"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
@@ -31,10 +32,12 @@ import (
"github.com/ipld/go-ipld-prime/traversal/selector/builder"
"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"
cborutil "github.com/filecoin-project/go-cbor-util"
"github.com/filecoin-project/go-commp-utils/ffiwrapper"
"github.com/filecoin-project/go-commp-utils/writer"
datatransfer "github.com/filecoin-project/go-data-transfer"
@@ -43,8 +46,10 @@ import (
rm "github.com/filecoin-project/go-fil-markets/retrievalmarket"
"github.com/filecoin-project/go-fil-markets/shared"
"github.com/filecoin-project/go-fil-markets/storagemarket"
"github.com/filecoin-project/go-fil-markets/storagemarket/network"
"github.com/filecoin-project/go-multistore"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/specs-actors/v3/actors/builtin/market"
marketevents "github.com/filecoin-project/lotus/markets/loggers"
@@ -91,7 +96,13 @@ func calcDealExpiration(minDuration uint64, md *dline.Info, startEpoch abi.Chain
minExp := startEpoch + abi.ChainEpoch(minDuration)
// Align on miners ProvingPeriodBoundary
return minExp + md.WPoStProvingPeriod - (minExp % md.WPoStProvingPeriod) + (md.PeriodStart % md.WPoStProvingPeriod) - 1
exp := minExp + md.WPoStProvingPeriod - (minExp % md.WPoStProvingPeriod) + (md.PeriodStart % md.WPoStProvingPeriod) - 1
// Should only be possible for miners created around genesis
for exp < minExp {
exp += md.WPoStProvingPeriod
}
return exp
}
func (a *API) imgr() *importmgr.Mgr {
@@ -99,8 +110,23 @@ func (a *API) imgr() *importmgr.Mgr {
}
func (a *API) ClientStartDeal(ctx context.Context, params *api.StartDealParams) (*cid.Cid, error) {
return a.dealStarter(ctx, params, false)
}
func (a *API) ClientStatelessDeal(ctx context.Context, params *api.StartDealParams) (*cid.Cid, error) {
return a.dealStarter(ctx, params, true)
}
func (a *API) dealStarter(ctx context.Context, params *api.StartDealParams, isStateless bool) (*cid.Cid, error) {
var storeID *multistore.StoreID
if params.Data.TransferType == storagemarket.TTGraphsync {
if isStateless {
if params.Data.TransferType != storagemarket.TTManual {
return nil, xerrors.Errorf("invalid transfer type %s for stateless storage deal", params.Data.TransferType)
}
if !params.EpochPrice.IsZero() {
return nil, xerrors.New("stateless storage deals can only be initiated with storage price of 0")
}
} else if params.Data.TransferType == storagemarket.TTGraphsync {
importIDs := a.imgr().List()
for _, importID := range importIDs {
info, err := a.imgr().Info(importID)
@@ -148,8 +174,6 @@ func (a *API) ClientStartDeal(ctx context.Context, params *api.StartDealParams)
return nil, xerrors.New("data doesn't fit in a sector")
}
providerInfo := utils.NewStorageProviderInfo(params.Miner, mi.Worker, mi.SectorSize, *mi.PeerId, mi.Multiaddrs)
dealStart := params.DealStartEpoch
if dealStart <= 0 { // unset, or explicitly 'epoch undefined'
ts, err := a.ChainHead(ctx)
@@ -171,25 +195,112 @@ func (a *API) ClientStartDeal(ctx context.Context, params *api.StartDealParams)
return nil, xerrors.Errorf("failed to get seal proof type: %w", err)
}
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,
StoreID: storeID,
})
// regular flow
if !isStateless {
providerInfo := utils.NewStorageProviderInfo(params.Miner, mi.Worker, mi.SectorSize, *mi.PeerId, mi.Multiaddrs)
if err != nil {
return nil, xerrors.Errorf("failed to start deal: %w", err)
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,
StoreID: storeID,
})
if err != nil {
return nil, xerrors.Errorf("failed to start deal: %w", err)
}
return &result.ProposalCid, nil
}
return &result.ProposalCid, nil
//
// stateless flow from here to the end
//
dealProposal := &market.DealProposal{
PieceCID: *params.Data.PieceCid,
PieceSize: params.Data.PieceSize.Padded(),
Client: walletKey,
Provider: params.Miner,
Label: params.Data.Root.Encode(multibase.MustNewEncoder('u')),
StartEpoch: dealStart,
EndEpoch: calcDealExpiration(params.MinBlocksDuration, md, dealStart),
StoragePricePerEpoch: big.Zero(),
ProviderCollateral: params.ProviderCollateral,
ClientCollateral: big.Zero(),
VerifiedDeal: params.VerifiedDeal,
}
if dealProposal.ProviderCollateral.IsZero() {
networkCollateral, err := a.StateDealProviderCollateralBounds(ctx, params.Data.PieceSize.Padded(), params.VerifiedDeal, types.EmptyTSK)
if err != nil {
return nil, xerrors.Errorf("failed to determine minimum provider collateral: %w", err)
}
dealProposal.ProviderCollateral = networkCollateral.Min
}
dealProposalSerialized, err := cborutil.Dump(dealProposal)
if err != nil {
return nil, xerrors.Errorf("failed to serialize deal proposal: %w", err)
}
dealProposalSig, err := a.WalletSign(ctx, walletKey, dealProposalSerialized)
if err != nil {
return nil, xerrors.Errorf("failed to sign proposal : %w", err)
}
dealProposalSigned := &market.ClientDealProposal{
Proposal: *dealProposal,
ClientSignature: *dealProposalSig,
}
dStream, err := network.NewFromLibp2pHost(a.Host,
// params duplicated from .../node/modules/client.go
// https://github.com/filecoin-project/lotus/pull/5961#discussion_r629768011
network.RetryParameters(time.Second, 5*time.Minute, 15, 5),
).NewDealStream(ctx, *mi.PeerId)
if err != nil {
return nil, xerrors.Errorf("opening dealstream to %s/%s failed: %w", params.Miner, *mi.PeerId, err)
}
if err = dStream.WriteDealProposal(network.Proposal{
FastRetrieval: true,
DealProposal: dealProposalSigned,
Piece: &storagemarket.DataRef{
TransferType: storagemarket.TTManual,
Root: params.Data.Root,
PieceCid: params.Data.PieceCid,
PieceSize: params.Data.PieceSize,
},
}); err != nil {
return nil, xerrors.Errorf("sending deal proposal failed: %w", err)
}
resp, _, err := dStream.ReadDealResponse()
if err != nil {
return nil, xerrors.Errorf("reading proposal response failed: %w", err)
}
dealProposalIpld, err := cborutil.AsIpld(dealProposalSigned)
if err != nil {
return nil, xerrors.Errorf("serializing proposal node failed: %w", err)
}
if !dealProposalIpld.Cid().Equals(resp.Response.Proposal) {
return nil, xerrors.Errorf("provider returned proposal cid %s but we expected %s", resp.Response.Proposal, dealProposalIpld.Cid())
}
if resp.Response.State != storagemarket.StorageDealWaitingForData {
return nil, xerrors.Errorf("provider returned unexpected state %d for proposal %s, with message: %s", resp.Response.State, resp.Response.Proposal, resp.Response.Message)
}
return &resp.Response.Proposal, nil
}
func (a *API) ClientListDeals(ctx context.Context) ([]api.DealInfo, error) {
node/modules/client.go
+10 -15
View File
@@ -134,26 +134,21 @@ func NewClientGraphsyncDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.Grap
// data-transfer push / pull channel restart configuration:
dtRestartConfig := dtimpl.ChannelRestartConfig(channelmonitor.Config{
// For now only monitor push channels (for storage deals)
MonitorPushChannels: true,
// TODO: Enable pull channel monitoring (for retrievals) when the
// following issue has been fixed:
// https://github.com/filecoin-project/go-data-transfer/issues/172
MonitorPullChannels: false,
// Wait up to 30s for the other side to respond to an Open channel message
AcceptTimeout: 30 * time.Second,
// Send a restart message if the data rate falls below 1024 bytes / minute
Interval: time.Minute,
MinBytesTransferred: 1024,
// Perform check 10 times / minute
ChecksPerInterval: 10,
// Wait up to 2m for the other side to respond to an Open channel message
AcceptTimeout: 2 * time.Minute,
// When an error occurs, wait a little while until all related errors
// have fired before sending a restart message
RestartDebounce: 10 * time.Second,
// After sending a restart, wait for at least 1 minute before sending another
RestartBackoff: time.Minute,
// After trying to restart 3 times, give up and fail the transfer
MaxConsecutiveRestarts: 3,
// Wait up to 30s for the other side to send a Complete message once all
// After sending a restart message, the time to wait for the peer to
// respond with an ack of the restart
RestartAckTimeout: 30 * time.Second,
// Wait up to 10m for the other side to send a Complete message once all
// data has been sent / received
CompleteTimeout: 30 * time.Second,
CompleteTimeout: 10 * time.Minute,
})
dt, err := dtimpl.NewDataTransfer(dtDs, filepath.Join(r.Path(), "data-transfer"), net, transport, dtRestartConfig)
if err != nil {
node/test/builder.go
+4
View File
@@ -12,6 +12,8 @@ import (
"testing"
"time"
"github.com/filecoin-project/go-state-types/network"
"github.com/gorilla/mux"
"golang.org/x/xerrors"
@@ -277,6 +279,7 @@ func mockBuilderOpts(t *testing.T, fullOpts []test.FullNodeOpts, storage []test.
genms = append(genms, *genm)
}
templ := &genesis.Template{
NetworkVersion: network.Version0,
Accounts: genaccs,
Miners: genms,
NetworkName: "test",
@@ -440,6 +443,7 @@ func mockSbBuilderOpts(t *testing.T, fullOpts []test.FullNodeOpts, storage []tes
genms = append(genms, *genm)
}
templ := &genesis.Template{
NetworkVersion: network.Version0,
Accounts: genaccs,
Miners: genms,
NetworkName: "test",