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

Merge branch 'next' into inmem-journal

Browse Source
This commit is contained in:
Raúl Kripalani
2020-08-11 12:41:11 +01:00
parent 5074ce5a38 0d1174cb81
commit 92598237c3
261 changed files with 16505 additions and 5690 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/modules/chain.go
+9 -4
View File
@@ -8,7 +8,6 @@ import (
"github.com/ipfs/go-bitswap/network"
"github.com/ipfs/go-blockservice"
"github.com/ipfs/go-datastore"
blockstore "github.com/ipfs/go-ipfs-blockstore"
"github.com/ipld/go-car"
"github.com/libp2p/go-libp2p-core/host"
"github.com/libp2p/go-libp2p-core/routing"
@@ -17,16 +16,18 @@ import (
"golang.org/x/xerrors"
"github.com/filecoin-project/sector-storage/ffiwrapper"
"github.com/filecoin-project/specs-actors/actors/runtime"
"github.com/filecoin-project/lotus/chain"
"github.com/filecoin-project/lotus/chain/beacon"
"github.com/filecoin-project/lotus/chain/blocksync"
"github.com/filecoin-project/lotus/chain/gen/slashfilter"
"github.com/filecoin-project/lotus/chain/messagepool"
"github.com/filecoin-project/lotus/chain/stmgr"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/chain/vm"
"github.com/filecoin-project/lotus/journal"
"github.com/filecoin-project/lotus/lib/blockstore"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/modules/helpers"
"github.com/filecoin-project/lotus/node/repo"
@@ -73,7 +74,7 @@ func ChainBlockstore(lc fx.Lifecycle, mctx helpers.MetricsCtx, r repo.LockedRepo
return nil, err
}
return blockstore.NewIdStore(cbs), nil
return cbs, nil
}
func ChainGCBlockstore(bs dtypes.ChainBlockstore, gcl dtypes.ChainGCLocker) dtypes.ChainGCBlockstore {
@@ -84,7 +85,7 @@ func ChainBlockservice(bs dtypes.ChainBlockstore, rem dtypes.ChainExchange) dtyp
return blockservice.New(bs, rem)
}
func ChainStore(lc fx.Lifecycle, bs dtypes.ChainBlockstore, ds dtypes.MetadataDS, syscalls runtime.Syscalls, journal journal.Journal) *store.ChainStore {
func ChainStore(lc fx.Lifecycle, bs dtypes.ChainBlockstore, ds dtypes.MetadataDS, syscalls vm.SyscallBuilder, journal journal.Journal) *store.ChainStore {
chain := store.NewChainStore(bs, ds, syscalls, journal)
if err := chain.Load(); err != nil {
@@ -168,3 +169,7 @@ func NewSyncer(lc fx.Lifecycle, sm *stmgr.StateManager, bsync *blocksync.BlockSy
})
return syncer, nil
}
func NewSlashFilter(ds dtypes.MetadataDS) *slashfilter.SlashFilter {
return slashfilter.New(ds)
}
node/modules/client.go
+41 -15
View File
@@ -4,11 +4,9 @@ import (
"context"
"time"
"github.com/filecoin-project/lotus/lib/bufbstore"
"github.com/filecoin-project/go-multistore"
"golang.org/x/xerrors"
blockstore "github.com/ipfs/go-ipfs-blockstore"
"github.com/libp2p/go-libp2p-core/host"
"go.uber.org/fx"
dtimpl "github.com/filecoin-project/go-data-transfer/impl"
@@ -20,19 +18,24 @@ import (
rmnet "github.com/filecoin-project/go-fil-markets/retrievalmarket/network"
"github.com/filecoin-project/go-fil-markets/storagemarket"
storageimpl "github.com/filecoin-project/go-fil-markets/storagemarket/impl"
"github.com/filecoin-project/go-fil-markets/storagemarket/impl/funds"
"github.com/filecoin-project/go-fil-markets/storagemarket/impl/requestvalidation"
smnet "github.com/filecoin-project/go-fil-markets/storagemarket/network"
"github.com/filecoin-project/go-statestore"
"github.com/filecoin-project/go-storedcounter"
"github.com/ipfs/go-datastore"
"github.com/ipfs/go-datastore/namespace"
"github.com/libp2p/go-libp2p-core/host"
"github.com/filecoin-project/lotus/lib/blockstore"
marketevents "github.com/filecoin-project/lotus/markets/loggers"
"github.com/filecoin-project/lotus/markets/retrievaladapter"
"github.com/filecoin-project/lotus/node/impl/full"
payapi "github.com/filecoin-project/lotus/node/impl/paych"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/repo"
"github.com/filecoin-project/lotus/node/repo/importmgr"
"github.com/filecoin-project/lotus/node/repo/retrievalstoremgr"
"github.com/filecoin-project/lotus/paychmgr"
)
@@ -42,7 +45,7 @@ func ClientMultiDatastore(lc fx.Lifecycle, r repo.LockedRepo) (dtypes.ClientMult
return nil, xerrors.Errorf("getting datastore out of reop: %w", err)
}
mds, err := importmgr.NewMultiDstore(ds)
mds, err := multistore.NewMultiDstore(ds)
if err != nil {
return nil, err
}
@@ -61,12 +64,8 @@ func ClientImportMgr(mds dtypes.ClientMultiDstore, ds dtypes.MetadataDS) dtypes.
}
func ClientBlockstore(imgr dtypes.ClientImportMgr) dtypes.ClientBlockstore {
// TODO: This isn't.. the best
// - If it's easy to pass per-retrieval blockstores with markets we don't need this
// - If it's not easy, we need to store this in a separate datastore on disk
defaultWrite := blockstore.NewBlockstore(datastore.NewMapDatastore())
return blockstore.NewIdStore(bufbstore.NewTieredBstore(imgr.Blockstore, defaultWrite))
// in most cases this is now unused in normal operations -- however, it's important to preserve for the IPFS use case
return blockstore.WrapIDStore(imgr.Blockstore)
}
// RegisterClientValidator is an initialization hook that registers the client
@@ -116,14 +115,21 @@ func NewClientRequestValidator(deals dtypes.ClientDealStore) dtypes.ClientReques
return requestvalidation.NewUnifiedRequestValidator(nil, deals)
}
func StorageClient(lc fx.Lifecycle, h host.Host, ibs dtypes.ClientBlockstore, r repo.LockedRepo, dataTransfer dtypes.ClientDataTransfer, discovery *discovery.Local, deals dtypes.ClientDatastore, scn storagemarket.StorageClientNode) (storagemarket.StorageClient, error) {
type ClientDealFunds funds.DealFunds
func NewClientDealFunds(ds dtypes.MetadataDS) (ClientDealFunds, error) {
return funds.NewDealFunds(ds, datastore.NewKey("/marketfunds/client"))
}
func StorageClient(lc fx.Lifecycle, h host.Host, ibs dtypes.ClientBlockstore, mds dtypes.ClientMultiDstore, r repo.LockedRepo, dataTransfer dtypes.ClientDataTransfer, discovery *discovery.Local, deals dtypes.ClientDatastore, scn storagemarket.StorageClientNode, dealFunds ClientDealFunds) (storagemarket.StorageClient, error) {
net := smnet.NewFromLibp2pHost(h)
c, err := storageimpl.NewClient(net, ibs, dataTransfer, discovery, deals, scn, storageimpl.DealPollingInterval(time.Second))
c, err := storageimpl.NewClient(net, ibs, mds, dataTransfer, discovery, deals, scn, dealFunds, storageimpl.DealPollingInterval(time.Second))
if err != nil {
return nil, err
}
lc.Append(fx.Hook{
OnStart: func(ctx context.Context) error {
c.SubscribeToEvents(marketevents.StorageClientLogger)
return c.Start(ctx)
},
OnStop: func(context.Context) error {
@@ -135,9 +141,29 @@ func StorageClient(lc fx.Lifecycle, h host.Host, ibs dtypes.ClientBlockstore, r
}
// RetrievalClient creates a new retrieval client attached to the client blockstore
func RetrievalClient(h host.Host, bs dtypes.ClientBlockstore, dt dtypes.ClientDataTransfer, pmgr *paychmgr.Manager, payapi payapi.PaychAPI, resolver retrievalmarket.PeerResolver, ds dtypes.MetadataDS, chainapi full.ChainAPI) (retrievalmarket.RetrievalClient, error) {
adapter := retrievaladapter.NewRetrievalClientNode(pmgr, payapi, chainapi)
func RetrievalClient(lc fx.Lifecycle, h host.Host, mds dtypes.ClientMultiDstore, dt dtypes.ClientDataTransfer, pmgr *paychmgr.Manager, payAPI payapi.PaychAPI, resolver retrievalmarket.PeerResolver, ds dtypes.MetadataDS, chainAPI full.ChainAPI, stateAPI full.StateAPI) (retrievalmarket.RetrievalClient, error) {
adapter := retrievaladapter.NewRetrievalClientNode(pmgr, payAPI, chainAPI, stateAPI)
network := rmnet.NewFromLibp2pHost(h)
sc := storedcounter.New(ds, datastore.NewKey("/retr"))
return retrievalimpl.NewClient(network, bs, dt, adapter, resolver, namespace.Wrap(ds, datastore.NewKey("/retrievals/client")), sc)
client, err := retrievalimpl.NewClient(network, mds, dt, adapter, resolver, namespace.Wrap(ds, datastore.NewKey("/retrievals/client")), sc)
if err != nil {
return nil, err
}
lc.Append(fx.Hook{
OnStart: func(ctx context.Context) error {
client.SubscribeToEvents(marketevents.RetrievalClientLogger)
return nil
},
})
return client, nil
}
// ClientRetrievalStoreManager is the default version of the RetrievalStoreManager that runs on multistore
func ClientRetrievalStoreManager(imgr dtypes.ClientImportMgr) dtypes.ClientRetrievalStoreManager {
return retrievalstoremgr.NewMultiStoreRetrievalStoreManager(imgr)
}
// ClientBlockstoreRetrievalStoreManager is the default version of the RetrievalStoreManager that runs on multistore
func ClientBlockstoreRetrievalStoreManager(bs dtypes.ClientBlockstore) dtypes.ClientRetrievalStoreManager {
return retrievalstoremgr.NewBlockstoreRetrievalStoreManager(bs)
}
node/modules/core.go
+2 -2
View File
@@ -90,6 +90,6 @@ func BuiltinBootstrap() (dtypes.BootstrapPeers, error) {
return build.BuiltinBootstrap()
}
func DrandBootstrap() (dtypes.DrandBootstrap, error) {
return build.DrandBootstrap()
func DrandBootstrap(d dtypes.DrandConfig) (dtypes.DrandBootstrap, error) {
return addrutil.ParseAddresses(context.TODO(), d.Relays)
}
node/modules/dtypes/beacon.go
+1
View File
@@ -2,5 +2,6 @@ package dtypes
type DrandConfig struct {
Servers []string
Relays []string
ChainInfoJSON string
}
node/modules/dtypes/miner.go
+9 -3
View File
@@ -1,10 +1,14 @@
package dtypes
import (
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/specs-actors/actors/abi"
"github.com/ipfs/go-cid"
"context"
"time"
"github.com/ipfs/go-cid"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/go-fil-markets/storagemarket"
"github.com/filecoin-project/specs-actors/actors/abi"
)
type MinerAddress address.Address
@@ -64,3 +68,5 @@ type SetExpectedSealDurationFunc func(time.Duration) error
// GetExpectedSealDurationFunc is a function which reads from miner
// too determine how long sealing is expected to take
type GetExpectedSealDurationFunc func() (time.Duration, error)
type DealFilter func(ctx context.Context, deal storagemarket.MinerDeal) (bool, string, error)
node/modules/dtypes/scorekeeper.go
+4 -3
View File
@@ -4,20 +4,21 @@ import (
"sync"
peer "github.com/libp2p/go-libp2p-core/peer"
pubsub "github.com/libp2p/go-libp2p-pubsub"
)
type ScoreKeeper struct {
lk sync.Mutex
scores map[peer.ID]float64
scores map[peer.ID]*pubsub.PeerScoreSnapshot
}
func (sk *ScoreKeeper) Update(scores map[peer.ID]float64) {
func (sk *ScoreKeeper) Update(scores map[peer.ID]*pubsub.PeerScoreSnapshot) {
sk.lk.Lock()
sk.scores = scores
sk.lk.Unlock()
}
func (sk *ScoreKeeper) Get() map[peer.ID]float64 {
func (sk *ScoreKeeper) Get() map[peer.ID]*pubsub.PeerScoreSnapshot {
sk.lk.Lock()
defer sk.lk.Unlock()
return sk.scores
node/modules/dtypes/storage.go
+7 -2
View File
@@ -4,16 +4,19 @@ import (
bserv "github.com/ipfs/go-blockservice"
"github.com/ipfs/go-datastore"
"github.com/ipfs/go-graphsync"
blockstore "github.com/ipfs/go-ipfs-blockstore"
exchange "github.com/ipfs/go-ipfs-exchange-interface"
format "github.com/ipfs/go-ipld-format"
"github.com/filecoin-project/go-fil-markets/storagemarket/impl/requestvalidation"
"github.com/filecoin-project/go-multistore"
datatransfer "github.com/filecoin-project/go-data-transfer"
"github.com/filecoin-project/go-fil-markets/piecestore"
"github.com/filecoin-project/go-statestore"
"github.com/filecoin-project/lotus/lib/blockstore"
"github.com/filecoin-project/lotus/node/repo/importmgr"
"github.com/filecoin-project/lotus/node/repo/retrievalstoremgr"
)
// MetadataDS stores metadata
@@ -27,12 +30,13 @@ type ChainGCBlockstore blockstore.GCBlockstore
type ChainExchange exchange.Interface
type ChainBlockService bserv.BlockService
type ClientMultiDstore *importmgr.MultiStore
type ClientMultiDstore *multistore.MultiStore
type ClientImportMgr *importmgr.Mgr
type ClientBlockstore blockstore.Blockstore
type ClientDealStore *statestore.StateStore
type ClientRequestValidator *requestvalidation.UnifiedRequestValidator
type ClientDatastore datastore.Batching
type ClientRetrievalStoreManager retrievalstoremgr.RetrievalStoreManager
type Graphsync graphsync.GraphExchange
@@ -49,3 +53,4 @@ type ProviderDataTransfer datatransfer.Manager
type StagingDAG format.DAGService
type StagingBlockstore blockstore.Blockstore
type StagingGraphsync graphsync.GraphExchange
type StagingMultiDstore *multistore.MultiStore
node/modules/ipfsclient.go
+4 -10
View File
@@ -4,10 +4,9 @@ import (
"go.uber.org/fx"
"golang.org/x/xerrors"
blockstore "github.com/ipfs/go-ipfs-blockstore"
"github.com/multiformats/go-multiaddr"
"github.com/filecoin-project/lotus/lib/bufbstore"
"github.com/filecoin-project/lotus/lib/blockstore"
"github.com/filecoin-project/lotus/lib/ipfsbstore"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/modules/helpers"
@@ -17,10 +16,10 @@ import (
// If ipfsMaddr is empty, a local IPFS node is assumed considering IPFS_PATH configuration.
// If ipfsMaddr is not empty, it will connect to the remote IPFS node with the provided multiaddress.
// The flag useForRetrieval indicates if the IPFS node will also be used for storing retrieving deals.
func IpfsClientBlockstore(ipfsMaddr string, useForRetrieval bool) func(helpers.MetricsCtx, fx.Lifecycle, dtypes.ClientImportMgr) (dtypes.ClientBlockstore, error) {
func IpfsClientBlockstore(ipfsMaddr string) func(helpers.MetricsCtx, fx.Lifecycle, dtypes.ClientImportMgr) (dtypes.ClientBlockstore, error) {
return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, localStore dtypes.ClientImportMgr) (dtypes.ClientBlockstore, error) {
var err error
var ipfsbs *ipfsbstore.IpfsBstore
var ipfsbs blockstore.Blockstore
if ipfsMaddr != "" {
var ma multiaddr.Multiaddr
ma, err = multiaddr.NewMultiaddr(ipfsMaddr)
@@ -34,11 +33,6 @@ func IpfsClientBlockstore(ipfsMaddr string, useForRetrieval bool) func(helpers.M
if err != nil {
return nil, xerrors.Errorf("constructing ipfs blockstore: %w", err)
}
var ws blockstore.Blockstore
ws = ipfsbs
if !useForRetrieval {
ws = blockstore.NewIdStore(localStore.Blockstore)
}
return bufbstore.NewTieredBstore(ipfsbs, ws), nil
return blockstore.WrapIDStore(ipfsbs), nil
}
}
node/modules/lp2p/pubsub.go
+27 -6
View File
@@ -301,7 +301,7 @@ func GossipSub(in GossipIn) (service *pubsub.PubSub, err error) {
return nil, err
}
trw := newTracerWrapper(tr)
trw := newTracerWrapper(tr, build.BlocksTopic(in.Nn))
options = append(options, pubsub.WithEventTracer(trw))
}
@@ -317,12 +317,27 @@ func HashMsgId(m *pubsub_pb.Message) string {
return string(hash[:])
}
func newTracerWrapper(tr pubsub.EventTracer) pubsub.EventTracer {
return &tracerWrapper{tr: tr}
func newTracerWrapper(tr pubsub.EventTracer, topics ...string) pubsub.EventTracer {
topicsMap := make(map[string]struct{})
for _, topic := range topics {
topicsMap[topic] = struct{}{}
}
return &tracerWrapper{tr: tr, topics: topicsMap}
}
type tracerWrapper struct {
tr pubsub.EventTracer
tr pubsub.EventTracer
topics map[string]struct{}
}
func (trw *tracerWrapper) traceMessage(topics []string) bool {
for _, topic := range topics {
_, ok := trw.topics[topic]
if ok {
return true
}
}
return false
}
func (trw *tracerWrapper) Trace(evt *pubsub_pb.TraceEvent) {
@@ -330,12 +345,18 @@ func (trw *tracerWrapper) Trace(evt *pubsub_pb.TraceEvent) {
// JOIN/LEAVE/GRAFT/PRUNE/PUBLISH/DELIVER. This significantly reduces bandwidth usage and still
// collects enough data to recover the state of the mesh and compute message delivery latency
// distributions.
// Furthermore, we only trace message publication and deliveries for specified topics
// (here just the blocks topic).
// TODO: hook all events into local metrics for inspection through the dashboard
switch evt.GetType() {
case pubsub_pb.TraceEvent_PUBLISH_MESSAGE:
trw.tr.Trace(evt)
if trw.traceMessage(evt.GetPublishMessage().Topics) {
trw.tr.Trace(evt)
}
case pubsub_pb.TraceEvent_DELIVER_MESSAGE:
trw.tr.Trace(evt)
if trw.traceMessage(evt.GetDeliverMessage().Topics) {
trw.tr.Trace(evt)
}
case pubsub_pb.TraceEvent_JOIN:
trw.tr.Trace(evt)
case pubsub_pb.TraceEvent_LEAVE:
node/modules/services.go
+19 -4
View File
@@ -41,7 +41,13 @@ func RunHello(mctx helpers.MetricsCtx, lc fx.Lifecycle, h host.Host, svc *hello.
pic := evt.(event.EvtPeerIdentificationCompleted)
go func() {
if err := svc.SayHello(helpers.LifecycleCtx(mctx, lc), pic.Peer); err != nil {
log.Warnw("failed to say hello", "error", err, "peer", pic.Peer)
protos, _ := h.Peerstore().GetProtocols(pic.Peer)
agent, _ := h.Peerstore().Get(pic.Peer, "AgentVersion")
if protosContains(protos, hello.ProtocolID) {
log.Warnw("failed to say hello", "error", err, "peer", pic.Peer, "supported", protos, "agent", agent)
} else {
log.Debugw("failed to say hello", "error", err, "peer", pic.Peer, "supported", protos, "agent", agent)
}
return
}
}()
@@ -50,6 +56,15 @@ func RunHello(mctx helpers.MetricsCtx, lc fx.Lifecycle, h host.Host, svc *hello.
return nil
}
func protosContains(protos []string, search string) bool {
for _, p := range protos {
if p == search {
return true
}
}
return false
}
func RunPeerMgr(mctx helpers.MetricsCtx, lc fx.Lifecycle, pmgr *peermgr.PeerMgr) {
go pmgr.Run(helpers.LifecycleCtx(mctx, lc))
}
@@ -58,7 +73,7 @@ func RunBlockSync(h host.Host, svc *blocksync.BlockSyncService) {
h.SetStreamHandler(blocksync.BlockSyncProtocolID, svc.HandleStream)
}
func HandleIncomingBlocks(mctx helpers.MetricsCtx, lc fx.Lifecycle, ps *pubsub.PubSub, s *chain.Syncer, chain *store.ChainStore, stmgr *stmgr.StateManager, h host.Host, nn dtypes.NetworkName) {
func HandleIncomingBlocks(mctx helpers.MetricsCtx, lc fx.Lifecycle, ps *pubsub.PubSub, s *chain.Syncer, bserv dtypes.ChainBlockService, chain *store.ChainStore, stmgr *stmgr.StateManager, h host.Host, nn dtypes.NetworkName) {
ctx := helpers.LifecycleCtx(mctx, lc)
blocksub, err := ps.Subscribe(build.BlocksTopic(nn))
@@ -77,7 +92,7 @@ func HandleIncomingBlocks(mctx helpers.MetricsCtx, lc fx.Lifecycle, ps *pubsub.P
panic(err)
}
go sub.HandleIncomingBlocks(ctx, blocksub, s, h.ConnManager())
go sub.HandleIncomingBlocks(ctx, blocksub, s, bserv, h.ConnManager())
}
func HandleIncomingMessages(mctx helpers.MetricsCtx, lc fx.Lifecycle, ps *pubsub.PubSub, mpool *messagepool.MessagePool, nn dtypes.NetworkName) {
@@ -114,7 +129,7 @@ type RandomBeaconParams struct {
}
func BuiltinDrandConfig() dtypes.DrandConfig {
return build.DrandConfig
return build.DrandConfig()
}
func RandomBeacon(p RandomBeaconParams, _ dtypes.AfterGenesisSet) (beacon.RandomBeacon, error) {
node/modules/storageminer.go
+119 -76
View File
@@ -7,6 +7,10 @@ import (
"net/http"
"time"
"go.uber.org/fx"
"go.uber.org/multierr"
"golang.org/x/xerrors"
"github.com/ipfs/go-bitswap"
"github.com/ipfs/go-bitswap/network"
"github.com/ipfs/go-blockservice"
@@ -16,13 +20,9 @@ import (
graphsync "github.com/ipfs/go-graphsync/impl"
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"
"github.com/libp2p/go-libp2p-core/routing"
"go.uber.org/fx"
"go.uber.org/multierr"
"golang.org/x/xerrors"
"github.com/filecoin-project/go-address"
dtimpl "github.com/filecoin-project/go-data-transfer/impl"
@@ -35,15 +35,16 @@ import (
rmnet "github.com/filecoin-project/go-fil-markets/retrievalmarket/network"
"github.com/filecoin-project/go-fil-markets/storagemarket"
storageimpl "github.com/filecoin-project/go-fil-markets/storagemarket/impl"
"github.com/filecoin-project/go-fil-markets/storagemarket/impl/funds"
"github.com/filecoin-project/go-fil-markets/storagemarket/impl/requestvalidation"
"github.com/filecoin-project/go-fil-markets/storagemarket/impl/storedask"
smnet "github.com/filecoin-project/go-fil-markets/storagemarket/network"
"github.com/filecoin-project/go-jsonrpc/auth"
"github.com/filecoin-project/go-multistore"
paramfetch "github.com/filecoin-project/go-paramfetch"
"github.com/filecoin-project/go-statestore"
"github.com/filecoin-project/go-storedcounter"
"github.com/filecoin-project/lotus/journal"
"github.com/filecoin-project/lotus/node/config"
sectorstorage "github.com/filecoin-project/sector-storage"
"github.com/filecoin-project/sector-storage/ffiwrapper"
"github.com/filecoin-project/sector-storage/stores"
@@ -53,9 +54,13 @@ import (
lapi "github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/gen"
"github.com/filecoin-project/lotus/chain/gen/slashfilter"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/lib/blockstore"
marketevents "github.com/filecoin-project/lotus/markets/loggers"
"github.com/filecoin-project/lotus/markets/retrievaladapter"
"github.com/filecoin-project/lotus/miner"
"github.com/filecoin-project/lotus/node/config"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/modules/helpers"
"github.com/filecoin-project/lotus/node/repo"
@@ -207,6 +212,7 @@ func StorageMiner(params StorageMinerParams) (*storage.Miner, error) {
func HandleRetrieval(host host.Host, lc fx.Lifecycle, m retrievalmarket.RetrievalProvider) {
lc.Append(fx.Hook{
OnStart: func(context.Context) error {
m.SubscribeToEvents(marketevents.RetrievalProviderLogger)
return m.Start()
},
OnStop: func(context.Context) error {
@@ -220,6 +226,7 @@ func HandleDeals(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host, h sto
lc.Append(fx.Hook{
OnStart: func(context.Context) error {
h.SubscribeToEvents(marketevents.StorageProviderLogger)
return h.Start(ctx)
},
OnStop: func(context.Context) error {
@@ -272,6 +279,26 @@ func NewProviderPieceStore(ds dtypes.MetadataDS) dtypes.ProviderPieceStore {
return piecestore.NewPieceStore(namespace.Wrap(ds, datastore.NewKey("/storagemarket")))
}
func StagingMultiDatastore(lc fx.Lifecycle, r repo.LockedRepo) (dtypes.StagingMultiDstore, error) {
ds, err := r.Datastore("/staging")
if err != nil {
return nil, xerrors.Errorf("getting datastore out of reop: %w", err)
}
mds, err := multistore.NewMultiDstore(ds)
if err != nil {
return nil, err
}
lc.Append(fx.Hook{
OnStop: func(ctx context.Context) error {
return mds.Close()
},
})
return mds, nil
}
// 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) {
@@ -280,10 +307,7 @@ func StagingBlockstore(r repo.LockedRepo) (dtypes.StagingBlockstore, error) {
return nil, err
}
bs := blockstore.NewBlockstore(stagingds)
ibs := blockstore.NewIdStore(bs)
return ibs, nil
return blockstore.NewBlockstore(stagingds), nil
}
// StagingDAG is a DAGService for the StagingBlockstore
@@ -316,13 +340,13 @@ func StagingGraphsync(mctx helpers.MetricsCtx, lc fx.Lifecycle, ibs dtypes.Stagi
return gs
}
func SetupBlockProducer(lc fx.Lifecycle, ds dtypes.MetadataDS, api lapi.FullNode, epp gen.WinningPoStProver) (*miner.Miner, error) {
func SetupBlockProducer(lc fx.Lifecycle, ds dtypes.MetadataDS, api lapi.FullNode, epp gen.WinningPoStProver, sf *slashfilter.SlashFilter) (*miner.Miner, error) {
minerAddr, err := minerAddrFromDS(ds)
if err != nil {
return nil, err
}
m := miner.NewMiner(api, epp, minerAddr)
m := miner.NewMiner(api, epp, minerAddr, sf)
lc.Append(fx.Hook{
OnStart: func(ctx context.Context) error {
@@ -356,94 +380,113 @@ func NewStorageAsk(ctx helpers.MetricsCtx, fapi lapi.FullNode, ds dtypes.Metadat
}
// Hacky way to set max piece size to the sector size
a := storedAsk.GetAsk().Ask
err = storedAsk.SetAsk(a.Price, a.Expiry-a.Timestamp, storagemarket.MaxPieceSize(abi.PaddedPieceSize(mi.SectorSize)))
err = storedAsk.SetAsk(a.Price, a.VerifiedPrice, a.Expiry-a.Timestamp, storagemarket.MaxPieceSize(abi.PaddedPieceSize(mi.SectorSize)))
if err != nil {
return storedAsk, err
}
return storedAsk, nil
}
type ProviderDealFunds funds.DealFunds
func NewProviderDealFunds(ds dtypes.MetadataDS) (ProviderDealFunds, error) {
return funds.NewDealFunds(ds, datastore.NewKey("/marketfunds/provider"))
}
func BasicDealFilter(user dtypes.DealFilter) func(onlineOk dtypes.ConsiderOnlineStorageDealsConfigFunc,
offlineOk dtypes.ConsiderOfflineStorageDealsConfigFunc,
blocklistFunc dtypes.StorageDealPieceCidBlocklistConfigFunc,
expectedSealTimeFunc dtypes.GetExpectedSealDurationFunc,
spn storagemarket.StorageProviderNode) dtypes.DealFilter {
return func(onlineOk dtypes.ConsiderOnlineStorageDealsConfigFunc,
offlineOk dtypes.ConsiderOfflineStorageDealsConfigFunc,
blocklistFunc dtypes.StorageDealPieceCidBlocklistConfigFunc,
expectedSealTimeFunc dtypes.GetExpectedSealDurationFunc,
spn storagemarket.StorageProviderNode) dtypes.DealFilter {
return func(ctx context.Context, deal storagemarket.MinerDeal) (bool, string, error) {
b, err := onlineOk()
if err != nil {
return false, "miner error", err
}
if deal.Ref != nil && deal.Ref.TransferType != storagemarket.TTManual && !b {
log.Warnf("online storage deal consideration disabled; rejecting storage deal proposal from client: %s", deal.Client.String())
return false, "miner is not considering online storage deals", nil
}
b, err = offlineOk()
if err != nil {
return false, "miner error", err
}
if deal.Ref != nil && deal.Ref.TransferType == storagemarket.TTManual && !b {
log.Warnf("offline storage deal consideration disabled; rejecting storage deal proposal from client: %s", deal.Client.String())
return false, "miner is not accepting offline storage deals", nil
}
blocklist, err := blocklistFunc()
if err != nil {
return false, "miner error", err
}
for idx := range blocklist {
if deal.Proposal.PieceCID.Equals(blocklist[idx]) {
log.Warnf("piece CID in proposal %s is blocklisted; rejecting storage deal proposal from client: %s", deal.Proposal.PieceCID, deal.Client.String())
return false, fmt.Sprintf("miner has blocklisted piece CID %s", deal.Proposal.PieceCID), nil
}
}
sealDuration, err := expectedSealTimeFunc()
if err != nil {
return false, "miner error", err
}
sealEpochs := sealDuration / (time.Duration(build.BlockDelaySecs) * time.Second)
_, ht, err := spn.GetChainHead(ctx)
if err != nil {
return false, "failed to get chain head", err
}
earliest := abi.ChainEpoch(sealEpochs) + ht
if deal.Proposal.StartEpoch < earliest {
log.Warnw("proposed deal would start before sealing can be completed; rejecting storage deal proposal from client", "piece_cid", deal.Proposal.PieceCID, "client", deal.Client.String(), "seal_duration", sealDuration, "earliest", earliest, "curepoch", ht)
return false, fmt.Sprintf("cannot seal a sector before %s", deal.Proposal.StartEpoch), nil
}
if user != nil {
return user(ctx, deal)
}
return true, "", nil
}
}
}
func StorageProvider(minerAddress dtypes.MinerAddress,
ffiConfig *ffiwrapper.Config,
storedAsk *storedask.StoredAsk,
h host.Host, ds dtypes.MetadataDS,
ibs dtypes.StagingBlockstore,
mds dtypes.StagingMultiDstore,
r repo.LockedRepo,
pieceStore dtypes.ProviderPieceStore,
dataTransfer dtypes.ProviderDataTransfer,
spn storagemarket.StorageProviderNode,
onlineOk dtypes.ConsiderOnlineStorageDealsConfigFunc,
offlineOk dtypes.ConsiderOfflineStorageDealsConfigFunc,
blocklistFunc dtypes.StorageDealPieceCidBlocklistConfigFunc,
expectedSealTimeFunc dtypes.GetExpectedSealDurationFunc) (storagemarket.StorageProvider, error) {
df dtypes.DealFilter,
funds ProviderDealFunds,
) (storagemarket.StorageProvider, error) {
net := smnet.NewFromLibp2pHost(h)
store, err := piecefilestore.NewLocalFileStore(piecefilestore.OsPath(r.Path()))
if err != nil {
return nil, err
}
opt := storageimpl.CustomDealDecisionLogic(func(ctx context.Context, deal storagemarket.MinerDeal) (bool, string, error) {
b, err := onlineOk()
if err != nil {
return false, "miner error", err
}
opt := storageimpl.CustomDealDecisionLogic(storageimpl.DealDeciderFunc(df))
if deal.Ref != nil && deal.Ref.TransferType != storagemarket.TTManual && !b {
log.Warnf("online storage deal consideration disabled; rejecting storage deal proposal from client: %s", deal.Client.String())
return false, "miner is not considering online storage deals", nil
}
b, err = offlineOk()
if err != nil {
return false, "miner error", err
}
if deal.Ref != nil && deal.Ref.TransferType == storagemarket.TTManual && !b {
log.Warnf("offline storage deal consideration disabled; rejecting storage deal proposal from client: %s", deal.Client.String())
return false, "miner is not accepting offline storage deals", nil
}
blocklist, err := blocklistFunc()
if err != nil {
return false, "miner error", err
}
for idx := range blocklist {
if deal.Proposal.PieceCID.Equals(blocklist[idx]) {
log.Warnf("piece CID in proposal %s is blocklisted; rejecting storage deal proposal from client: %s", deal.Proposal.PieceCID, deal.Client.String())
return false, fmt.Sprintf("miner has blocklisted piece CID %s", deal.Proposal.PieceCID), nil
}
}
sealDuration, err := expectedSealTimeFunc()
if err != nil {
return false, "miner error", err
}
sealEpochs := sealDuration / (time.Duration(build.BlockDelaySecs) * time.Second)
_, ht, err := spn.GetChainHead(ctx)
if err != nil {
return false, "failed to get chain head", err
}
earliest := abi.ChainEpoch(sealEpochs) + ht
if deal.Proposal.StartEpoch < earliest {
log.Warnf("proposed deal would start before sealing can be completed; rejecting storage deal proposal from client: %s", deal.Proposal.PieceCID, deal.Client.String())
return false, fmt.Sprintf("cannot seal a sector before %s", deal.Proposal.StartEpoch), nil
}
return true, "", nil
})
p, err := storageimpl.NewProvider(net, namespace.Wrap(ds, datastore.NewKey("/deals/provider")), ibs, store, pieceStore, dataTransfer, spn, address.Address(minerAddress), ffiConfig.SealProofType, storedAsk, opt)
if err != nil {
return p, err
}
return p, nil
return storageimpl.NewProvider(net, namespace.Wrap(ds, datastore.NewKey("/deals/provider")), store, mds, pieceStore, dataTransfer, spn, address.Address(minerAddress), ffiConfig.SealProofType, storedAsk, funds, opt)
}
// RetrievalProvider creates a new retrieval provider attached to the provider blockstore
func RetrievalProvider(h host.Host, miner *storage.Miner, sealer sectorstorage.SectorManager, full lapi.FullNode, ds dtypes.MetadataDS, pieceStore dtypes.ProviderPieceStore, ibs dtypes.StagingBlockstore, dt dtypes.ProviderDataTransfer, onlineOk dtypes.ConsiderOnlineRetrievalDealsConfigFunc, offlineOk dtypes.ConsiderOfflineRetrievalDealsConfigFunc) (retrievalmarket.RetrievalProvider, error) {
func RetrievalProvider(h host.Host, miner *storage.Miner, sealer sectorstorage.SectorManager, full lapi.FullNode, ds dtypes.MetadataDS, pieceStore dtypes.ProviderPieceStore, mds dtypes.StagingMultiDstore, dt dtypes.ProviderDataTransfer, onlineOk dtypes.ConsiderOnlineRetrievalDealsConfigFunc, offlineOk dtypes.ConsiderOfflineRetrievalDealsConfigFunc) (retrievalmarket.RetrievalProvider, error) {
adapter := retrievaladapter.NewRetrievalProviderNode(miner, sealer, full)
maddr, err := minerAddrFromDS(ds)
@@ -476,7 +519,7 @@ func RetrievalProvider(h host.Host, miner *storage.Miner, sealer sectorstorage.S
return true, "", nil
})
return retrievalimpl.NewProvider(maddr, adapter, netwk, pieceStore, ibs, dt, namespace.Wrap(ds, datastore.NewKey("/retrievals/provider")), opt)
return retrievalimpl.NewProvider(maddr, adapter, netwk, pieceStore, mds, dt, namespace.Wrap(ds, datastore.NewKey("/retrievals/provider")), opt)
}
func SectorStorage(mctx helpers.MetricsCtx, lc fx.Lifecycle, ls stores.LocalStorage, si stores.SectorIndex, cfg *ffiwrapper.Config, sc sectorstorage.SealerConfig, urls sectorstorage.URLs, sa sectorstorage.StorageAuth) (*sectorstorage.Manager, error) {
node/modules/testing/genesis.go
+5 -6
View File
@@ -17,12 +17,11 @@ import (
"github.com/mitchellh/go-homedir"
"golang.org/x/xerrors"
"github.com/filecoin-project/specs-actors/actors/runtime"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/gen"
genesis2 "github.com/filecoin-project/lotus/chain/gen/genesis"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/chain/vm"
"github.com/filecoin-project/lotus/genesis"
"github.com/filecoin-project/lotus/node/modules"
"github.com/filecoin-project/lotus/node/modules/dtypes"
@@ -30,8 +29,8 @@ import (
var glog = logging.Logger("genesis")
func MakeGenesisMem(out io.Writer, template genesis.Template) func(bs dtypes.ChainBlockstore, syscalls runtime.Syscalls) modules.Genesis {
return func(bs dtypes.ChainBlockstore, syscalls runtime.Syscalls) modules.Genesis {
func MakeGenesisMem(out io.Writer, template genesis.Template) func(bs dtypes.ChainBlockstore, syscalls vm.SyscallBuilder) modules.Genesis {
return func(bs dtypes.ChainBlockstore, syscalls vm.SyscallBuilder) modules.Genesis {
return func() (*types.BlockHeader, error) {
glog.Warn("Generating new random genesis block, note that this SHOULD NOT happen unless you are setting up new network")
b, err := genesis2.MakeGenesisBlock(context.TODO(), bs, syscalls, template)
@@ -51,8 +50,8 @@ func MakeGenesisMem(out io.Writer, template genesis.Template) func(bs dtypes.Cha
}
}
func MakeGenesis(outFile, genesisTemplate string) func(bs dtypes.ChainBlockstore, syscalls runtime.Syscalls) modules.Genesis {
return func(bs dtypes.ChainBlockstore, syscalls runtime.Syscalls) modules.Genesis {
func MakeGenesis(outFile, genesisTemplate string) func(bs dtypes.ChainBlockstore, syscalls vm.SyscallBuilder) modules.Genesis {
return func(bs dtypes.ChainBlockstore, syscalls vm.SyscallBuilder) modules.Genesis {
return func() (*types.BlockHeader, error) {
glog.Warn("Generating new random genesis block, note that this SHOULD NOT happen unless you are setting up new network")
genesisTemplate, err := homedir.Expand(genesisTemplate)