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

initial resolution of conflicts

Browse Source
This commit is contained in:
Anton Evangelatov
2021-06-08 13:01:22 +02:00
parent de646a5b93 631fdaa217
commit 9f3ec82cc1
66 changed files with 4144 additions and 1486 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/impl/client/client.go
+78
View File
@@ -6,6 +6,7 @@ import (
"fmt"
"io"
"os"
"sort"
"time"
"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
@@ -836,6 +837,83 @@ func (a *API) clientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
return
}
func (a *API) ClientListRetrievals(ctx context.Context) ([]api.RetrievalInfo, error) {
deals, err := a.Retrieval.ListDeals()
if err != nil {
return nil, err
}
dataTransfersByID, err := a.transfersByID(ctx)
if err != nil {
return nil, err
}
out := make([]api.RetrievalInfo, 0, len(deals))
for _, v := range deals {
// Find the data transfer associated with this deal
var transferCh *api.DataTransferChannel
if v.ChannelID != nil {
if ch, ok := dataTransfersByID[*v.ChannelID]; ok {
transferCh = &ch
}
}
out = append(out, a.newRetrievalInfoWithTransfer(transferCh, v))
}
sort.Slice(out, func(a, b int) bool {
return out[a].ID < out[b].ID
})
return out, nil
}
func (a *API) ClientGetRetrievalUpdates(ctx context.Context) (<-chan api.RetrievalInfo, error) {
updates := make(chan api.RetrievalInfo)
unsub := a.Retrieval.SubscribeToEvents(func(_ rm.ClientEvent, deal rm.ClientDealState) {
updates <- a.newRetrievalInfo(ctx, deal)
})
go func() {
defer unsub()
<-ctx.Done()
}()
return updates, nil
}
func (a *API) newRetrievalInfoWithTransfer(ch *api.DataTransferChannel, deal rm.ClientDealState) api.RetrievalInfo {
return api.RetrievalInfo{
PayloadCID: deal.PayloadCID,
ID: deal.ID,
PieceCID: deal.PieceCID,
PricePerByte: deal.PricePerByte,
UnsealPrice: deal.UnsealPrice,
Status: deal.Status,
Message: deal.Message,
Provider: deal.Sender,
BytesReceived: deal.TotalReceived,
BytesPaidFor: deal.BytesPaidFor,
TotalPaid: deal.FundsSpent,
TransferChannelID: deal.ChannelID,
DataTransfer: ch,
}
}
func (a *API) newRetrievalInfo(ctx context.Context, v rm.ClientDealState) api.RetrievalInfo {
// Find the data transfer associated with this deal
var transferCh *api.DataTransferChannel
if v.ChannelID != nil {
state, err := a.DataTransfer.ChannelState(ctx, *v.ChannelID)
// Note: If there was an error just ignore it, as the data transfer may
// be not found if it's no longer active
if err == nil {
ch := api.NewDataTransferChannel(a.Host.ID(), state)
ch.Stages = state.Stages()
transferCh = &ch
}
}
return a.newRetrievalInfoWithTransfer(transferCh, v)
}
type multiStoreRetrievalStore struct {
storeID multistore.StoreID
store *multistore.Store
node/impl/common/common.go
+1 -1
View File
@@ -156,7 +156,7 @@ func (a *CommonAPI) NetFindPeer(ctx context.Context, p peer.ID) (peer.AddrInfo,
}
func (a *CommonAPI) NetAutoNatStatus(ctx context.Context) (i api.NatInfo, err error) {
autonat := a.RawHost.(*basichost.BasicHost).AutoNat
autonat := a.RawHost.(*basichost.BasicHost).GetAutoNat()
if autonat == nil {
return api.NatInfo{
node/impl/remoteworker.go
+10
View File
@@ -38,6 +38,16 @@ func connectRemoteWorker(ctx context.Context, fa api.Common, url string) (*remot
return nil, xerrors.Errorf("creating jsonrpc client: %w", err)
}
wver, err := wapi.Version(ctx)
if err != nil {
closer()
return nil, err
}
if !wver.EqMajorMinor(api.WorkerAPIVersion0) {
return nil, xerrors.Errorf("unsupported worker api version: %s (expected %s)", wver, api.WorkerAPIVersion0)
}
return &remoteWorker{wapi, closer}, nil
}
node/modules/dtypes/miner.go
+4 -2
View File
@@ -74,10 +74,12 @@ type ConsiderUnverifiedStorageDealsConfigFunc func() (bool, error)
// disable or enable unverified storage deal acceptance.
type SetConsiderUnverifiedStorageDealsConfigFunc func(bool) error
// SetSealingDelay sets how long a sector waits for more deals before sealing begins.
// SetSealingConfigFunc is a function which is used to
// sets the sealing config.
type SetSealingConfigFunc func(sealiface.Config) error
// GetSealingDelay returns how long a sector waits for more deals before sealing begins.
// GetSealingConfigFunc is a function which is used to
// get the sealing config.
type GetSealingConfigFunc func() (sealiface.Config, error)
// SetExpectedSealDurationFunc is a function which is used to set how long sealing is expected to take.
node/modules/storageminer.go
+2 -3
View File
@@ -647,8 +647,7 @@ func RetrievalProvider(
pieceStore dtypes.ProviderPieceStore,
mds dtypes.StagingMultiDstore,
dt dtypes.ProviderDataTransfer,
onlineOk dtypes.ConsiderOnlineRetrievalDealsConfigFunc,
offlineOk dtypes.ConsiderOfflineRetrievalDealsConfigFunc,
pieceProvider sectorstorage.PieceProvider,
userFilter dtypes.RetrievalDealFilter,
) (retrievalmarket.RetrievalProvider, error) {
opt := retrievalimpl.DealDeciderOpt(retrievalimpl.DealDecider(userFilter))
@@ -664,7 +663,7 @@ func LocalStorage(mctx helpers.MetricsCtx, lc fx.Lifecycle, ls stores.LocalStora
}
func RemoteStorage(lstor *stores.Local, si stores.SectorIndex, sa sectorstorage.StorageAuth, sc sectorstorage.SealerConfig) *stores.Remote {
return stores.NewRemote(lstor, si, http.Header(sa), sc.ParallelFetchLimit)
return stores.NewRemote(lstor, si, http.Header(sa), sc.ParallelFetchLimit, &stores.DefaultPartialFileHandler{})
}
func SectorStorage(mctx helpers.MetricsCtx, lc fx.Lifecycle, lstor *stores.Local, stor *stores.Remote, ls stores.LocalStorage, si stores.SectorIndex, sc sectorstorage.SealerConfig, sa sectorstorage.StorageAuth, ds dtypes.MetadataDS) (*sectorstorage.Manager, error) {
node/node_test.go
+6
View File
@@ -58,6 +58,12 @@ func TestAPIDealFlow(t *testing.T) {
t.Run("TestPublishDealsBatching", func(t *testing.T) {
test.TestPublishDealsBatching(t, builder.MockSbBuilder, blockTime, dealStartEpoch)
})
t.Run("TestOfflineDealFlow", func(t *testing.T) {
test.TestOfflineDealFlow(t, builder.MockSbBuilder, blockTime, dealStartEpoch, false)
})
t.Run("TestOfflineDealFlowFastRetrieval", func(t *testing.T) {
test.TestOfflineDealFlow(t, builder.MockSbBuilder, blockTime, dealStartEpoch, true)
})
}
func TestBatchDealInput(t *testing.T) {
node/test/builder.go
+14 -19
View File
@@ -522,18 +522,15 @@ func mockSbBuilderOpts(t *testing.T, fullOpts []test.FullNodeOpts, storage []tes
fulls[i] = fullRpc(t, fulls[i])
}
mgr := mock.NewMockSectorMgr(nil)
fulls[i].Stb = storageBuilder(fulls[i], mn, node.Options(
node.Override(new(sectorstorage.SectorManager), func() (sectorstorage.SectorManager, error) {
return mgr, nil
}),
node.Override(new(sectorstorage.PieceProvider), func() (sectorstorage.PieceProvider, error) {
return mgr, nil
}),
node.Override(new(sectorstorage.Unsealer), func() (sectorstorage.Unsealer, error) {
return mgr, nil
node.Override(new(*mock.SectorMgr), func() (*mock.SectorMgr, error) {
return mock.NewMockSectorMgr(nil), nil
}),
node.Override(new(sectorstorage.SectorManager), node.From(new(*mock.SectorMgr))),
node.Override(new(sectorstorage.Unsealer), node.From(new(*mock.SectorMgr))),
node.Override(new(sectorstorage.PieceProvider), node.From(new(*mock.SectorMgr))),
node.Override(new(ffiwrapper.Verifier), mock.MockVerifier),
node.Override(new(ffiwrapper.Prover), mock.MockProver),
node.Unset(new(*sectorstorage.Manager)),
@@ -568,22 +565,20 @@ func mockSbBuilderOpts(t *testing.T, fullOpts []test.FullNodeOpts, storage []tes
Number: sector.SectorID,
}
}
mgr := mock.NewMockSectorMgr(sectors)
opts := def.Opts
if opts == nil {
opts = node.Options()
}
storers[i] = CreateTestStorageNode(ctx, t, genms[i].Worker, maddrs[i], pidKeys[i], f, mn, node.Options(
node.Override(new(sectorstorage.SectorManager), func() (sectorstorage.SectorManager, error) {
return mgr, nil
}),
node.Override(new(sectorstorage.PieceProvider), func() (sectorstorage.PieceProvider, error) {
return mgr, nil
}),
node.Override(new(sectorstorage.Unsealer), func() (sectorstorage.Unsealer, error) {
return mgr, nil
node.Override(new(*mock.SectorMgr), func() (*mock.SectorMgr, error) {
return mock.NewMockSectorMgr(sectors), nil
}),
node.Override(new(sectorstorage.SectorManager), node.From(new(*mock.SectorMgr))),
node.Override(new(sectorstorage.Unsealer), node.From(new(*mock.SectorMgr))),
node.Override(new(sectorstorage.PieceProvider), node.From(new(*mock.SectorMgr))),
node.Override(new(ffiwrapper.Verifier), mock.MockVerifier),
node.Override(new(ffiwrapper.Prover), mock.MockProver),
node.Unset(new(*sectorstorage.Manager)),