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

Merge branch 'master' into chore/snake_context_through_blockstore_init

Browse Source
This commit is contained in:
Raúl Kripalani
2021-01-25 19:31:41 +00:00
parent 47141042b6 519b5e191e
commit ce3af308ed
103 changed files with 3323 additions and 689 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/config/def.go
+2
View File
@@ -69,6 +69,7 @@ type SealingConfig struct {
type MinerFeeConfig struct {
MaxPreCommitGasFee types.FIL
MaxCommitGasFee types.FIL
MaxTerminateGasFee types.FIL
MaxWindowPoStGasFee types.FIL
MaxPublishDealsFee types.FIL
MaxMarketBalanceAddFee types.FIL
@@ -211,6 +212,7 @@ func DefaultStorageMiner() *StorageMiner {
Fees: MinerFeeConfig{
MaxPreCommitGasFee: types.MustParseFIL("0.025"),
MaxCommitGasFee: types.MustParseFIL("0.05"),
MaxTerminateGasFee: types.MustParseFIL("0.5"),
MaxWindowPoStGasFee: types.MustParseFIL("5"),
MaxPublishDealsFee: types.MustParseFIL("0.05"),
MaxMarketBalanceAddFee: types.MustParseFIL("0.007"),
node/impl/full/gas.go
+6 -1
View File
@@ -93,13 +93,15 @@ type gasMeta struct {
limit int64
}
// finds 55th percntile instead of median to put negative pressure on gas price
func medianGasPremium(prices []gasMeta, blocks int) abi.TokenAmount {
sort.Slice(prices, func(i, j int) bool {
// sort desc by price
return prices[i].price.GreaterThan(prices[j].price)
})
at := build.BlockGasTarget * int64(blocks) / 2
at := build.BlockGasTarget * int64(blocks) / 2 // 50th
at += build.BlockGasTarget * int64(blocks) / (2 * 20) // move 5% further
prev1, prev2 := big.Zero(), big.Zero()
for _, price := range prices {
prev1, prev2 = price.price, prev1
@@ -227,6 +229,9 @@ func gasEstimateGasLimit(
pending, ts := mpool.PendingFor(fromA)
priorMsgs := make([]types.ChainMsg, 0, len(pending))
for _, m := range pending {
if m.Message.Nonce == msg.Nonce {
break
}
priorMsgs = append(priorMsgs, m)
}
node/impl/full/state.go
+9 -2
View File
@@ -55,6 +55,7 @@ type StateModuleAPI interface {
StateMinerPower(context.Context, address.Address, types.TipSetKey) (*api.MinerPower, error)
StateNetworkVersion(ctx context.Context, key types.TipSetKey) (network.Version, error)
StateSectorGetInfo(ctx context.Context, maddr address.Address, n abi.SectorNumber, tsk types.TipSetKey) (*miner.SectorOnChainInfo, error)
StateSearchMsg(ctx context.Context, msg cid.Cid) (*api.MsgLookup, error)
StateVerifiedClientStatus(ctx context.Context, addr address.Address, tsk types.TipSetKey) (*abi.StoragePower, error)
StateWaitMsg(ctx context.Context, msg cid.Cid, confidence uint64) (*api.MsgLookup, error)
}
@@ -589,8 +590,14 @@ func stateWaitMsgLimited(ctx context.Context, smgr *stmgr.StateManager, cstore *
}, nil
}
func (a *StateAPI) StateSearchMsg(ctx context.Context, msg cid.Cid) (*api.MsgLookup, error) {
ts, recpt, found, err := a.StateManager.SearchForMessage(ctx, msg)
func (m *StateModule) StateSearchMsg(ctx context.Context, msg cid.Cid) (*api.MsgLookup, error) {
return stateSearchMsgLimited(ctx, m.StateManager, msg, stmgr.LookbackNoLimit)
}
func (a *StateAPI) StateSearchMsgLimited(ctx context.Context, msg cid.Cid, lookbackLimit abi.ChainEpoch) (*api.MsgLookup, error) {
return stateSearchMsgLimited(ctx, a.StateManager, msg, lookbackLimit)
}
func stateSearchMsgLimited(ctx context.Context, smgr *stmgr.StateManager, msg cid.Cid, lookbackLimit abi.ChainEpoch) (*api.MsgLookup, error) {
ts, recpt, found, err := smgr.SearchForMessage(ctx, msg, lookbackLimit)
if err != nil {
return nil, err
}
node/impl/market/market.go
+30 -2
View File
@@ -3,21 +3,49 @@ package market
import (
"context"
"github.com/ipfs/go-cid"
"go.uber.org/fx"
"github.com/ipfs/go-cid"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/lotus/chain/actors"
marketactor "github.com/filecoin-project/lotus/chain/actors/builtin/market"
"github.com/filecoin-project/lotus/chain/market"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/node/impl/full"
)
type MarketAPI struct {
fx.In
full.MpoolAPI
FMgr *market.FundManager
}
func (a *MarketAPI) MarketAddBalance(ctx context.Context, wallet, addr address.Address, amt types.BigInt) (cid.Cid, error) {
params, err := actors.SerializeParams(&addr)
if err != nil {
return cid.Undef, err
}
smsg, aerr := a.MpoolPushMessage(ctx, &types.Message{
To: marketactor.Address,
From: wallet,
Value: amt,
Method: marketactor.Methods.AddBalance,
Params: params,
}, nil)
if aerr != nil {
return cid.Undef, aerr
}
return smsg.Cid(), nil
}
func (a *MarketAPI) MarketGetReserved(ctx context.Context, addr address.Address) (types.BigInt, error) {
return a.FMgr.GetReserved(addr), nil
}
func (a *MarketAPI) MarketReserveFunds(ctx context.Context, wallet address.Address, addr address.Address, amt types.BigInt) (cid.Cid, error) {
return a.FMgr.Reserve(ctx, wallet, addr, amt)
}
node/impl/storminer.go
+12
View File
@@ -328,6 +328,18 @@ func (sm *StorageMinerAPI) SectorRemove(ctx context.Context, id abi.SectorNumber
return sm.Miner.RemoveSector(ctx, id)
}
func (sm *StorageMinerAPI) SectorTerminate(ctx context.Context, id abi.SectorNumber) error {
return sm.Miner.TerminateSector(ctx, id)
}
func (sm *StorageMinerAPI) SectorTerminateFlush(ctx context.Context) (*cid.Cid, error) {
return sm.Miner.TerminateFlush(ctx)
}
func (sm *StorageMinerAPI) SectorTerminatePending(ctx context.Context) ([]abi.SectorID, error) {
return sm.Miner.TerminatePending(ctx)
}
func (sm *StorageMinerAPI) SectorMarkForUpgrade(ctx context.Context, id abi.SectorNumber) error {
return sm.Miner.MarkForUpgrade(id)
}
node/modules/client.go
+13 -3
View File
@@ -120,7 +120,11 @@ func RegisterClientValidator(crv dtypes.ClientRequestValidator, dtm dtypes.Clien
// uses the clients's Client DAG service for transfers
func NewClientGraphsyncDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.Graphsync, ds dtypes.MetadataDS, r repo.LockedRepo) (dtypes.ClientDataTransfer, error) {
sc := storedcounter.New(ds, datastore.NewKey("/datatransfer/client/counter"))
net := dtnet.NewFromLibp2pHost(h)
// go-data-transfer protocol retries:
// 1s, 5s, 25s, 2m5s, 5m x 11 ~= 1 hour
dtRetryParams := dtnet.RetryParameters(time.Second, 5*time.Minute, 15, 5)
net := dtnet.NewFromLibp2pHost(h, dtRetryParams)
dtDs := namespace.Wrap(ds, datastore.NewKey("/datatransfer/client/transfers"))
transport := dtgstransport.NewTransport(h.ID(), gs)
@@ -129,7 +133,9 @@ func NewClientGraphsyncDataTransfer(lc fx.Lifecycle, h host.Host, gs dtypes.Grap
return nil, err
}
dt, err := dtimpl.NewDataTransfer(dtDs, filepath.Join(r.Path(), "data-transfer"), net, transport, sc)
// data-transfer push channel restart configuration
dtRestartConfig := dtimpl.PushChannelRestartConfig(time.Minute, 10, 1024, 10*time.Minute, 3)
dt, err := dtimpl.NewDataTransfer(dtDs, filepath.Join(r.Path(), "data-transfer"), net, transport, sc, dtRestartConfig)
if err != nil {
return nil, err
}
@@ -153,7 +159,11 @@ func NewClientDatastore(ds dtypes.MetadataDS) dtypes.ClientDatastore {
}
func StorageClient(lc fx.Lifecycle, h host.Host, ibs dtypes.ClientBlockstore, mds dtypes.ClientMultiDstore, r repo.LockedRepo, dataTransfer dtypes.ClientDataTransfer, discovery *discoveryimpl.Local, deals dtypes.ClientDatastore, scn storagemarket.StorageClientNode, j journal.Journal) (storagemarket.StorageClient, error) {
net := smnet.NewFromLibp2pHost(h)
// go-fil-markets protocol retries:
// 1s, 5s, 25s, 2m5s, 5m x 11 ~= 1 hour
marketsRetryParams := smnet.RetryParameters(time.Second, 5*time.Minute, 15, 5)
net := smnet.NewFromLibp2pHost(h, marketsRetryParams)
c, err := storageimpl.NewClient(net, ibs, mds, dataTransfer, discovery, deals, scn, storageimpl.DealPollingInterval(time.Second))
if err != nil {
return nil, err
node/modules/core.go
+55 -28
View File
@@ -7,6 +7,7 @@ import (
"io"
"io/ioutil"
"os"
"path/filepath"
"time"
"github.com/gbrlsnchs/jwt/v3"
@@ -14,6 +15,7 @@ import (
"github.com/libp2p/go-libp2p-core/peer"
"github.com/libp2p/go-libp2p-core/peerstore"
record "github.com/libp2p/go-libp2p-record"
"github.com/raulk/go-watchdog"
"go.uber.org/fx"
"golang.org/x/xerrors"
@@ -28,7 +30,6 @@ import (
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/repo"
"github.com/filecoin-project/lotus/system"
"github.com/raulk/go-watchdog"
)
const (
@@ -69,45 +70,71 @@ func MemoryConstraints() system.MemoryConstraints {
// MemoryWatchdog starts the memory watchdog, applying the computed resource
// constraints.
func MemoryWatchdog(lc fx.Lifecycle, constraints system.MemoryConstraints) {
func MemoryWatchdog(lr repo.LockedRepo, lc fx.Lifecycle, constraints system.MemoryConstraints) {
if os.Getenv(EnvWatchdogDisabled) == "1" {
log.Infof("memory watchdog is disabled via %s", EnvWatchdogDisabled)
return
}
cfg := watchdog.MemConfig{
Resolution: 5 * time.Second,
Policy: &watchdog.WatermarkPolicy{
Watermarks: []float64{0.50, 0.60, 0.70, 0.85, 0.90, 0.925, 0.95},
EmergencyWatermark: 0.95,
},
Logger: logWatchdog,
// configure heap profile capture so that one is captured per episode where
// utilization climbs over 90% of the limit. A maximum of 10 heapdumps
// will be captured during life of this process.
watchdog.HeapProfileDir = filepath.Join(lr.Path(), "heapprof")
watchdog.HeapProfileMaxCaptures = 10
watchdog.HeapProfileThreshold = 0.9
watchdog.Logger = logWatchdog
policy := watchdog.NewWatermarkPolicy(0.50, 0.60, 0.70, 0.85, 0.90, 0.925, 0.95)
// Try to initialize a watchdog in the following order of precedence:
// 1. If a max heap limit has been provided, initialize a heap-driven watchdog.
// 2. Else, try to initialize a cgroup-driven watchdog.
// 3. Else, try to initialize a system-driven watchdog.
// 4. Else, log a warning that the system is flying solo, and return.
addStopHook := func(stopFn func()) {
lc.Append(fx.Hook{
OnStop: func(ctx context.Context) error {
stopFn()
return nil
},
})
}
// if user has set max heap limit, apply it. Otherwise, fall back to total
// system memory constraint.
// 1. If user has set max heap limit, apply it.
if maxHeap := constraints.MaxHeapMem; maxHeap != 0 {
log.Infof("memory watchdog will apply max heap constraint: %d bytes", maxHeap)
cfg.Limit = maxHeap
cfg.Scope = watchdog.ScopeHeap
} else {
log.Infof("max heap size not provided; memory watchdog will apply total system memory constraint: %d bytes", constraints.TotalSystemMem)
cfg.Limit = constraints.TotalSystemMem
cfg.Scope = watchdog.ScopeSystem
const minGOGC = 10
err, stopFn := watchdog.HeapDriven(maxHeap, minGOGC, policy)
if err == nil {
log.Infof("initialized heap-driven watchdog; max heap: %d bytes", maxHeap)
addStopHook(stopFn)
return
}
log.Warnf("failed to initialize heap-driven watchdog; err: %s", err)
log.Warnf("trying a cgroup-driven watchdog")
}
err, stop := watchdog.Memory(cfg)
if err != nil {
log.Warnf("failed to instantiate memory watchdog: %s", err)
// 2. cgroup-driven watchdog.
err, stopFn := watchdog.CgroupDriven(5*time.Second, policy)
if err == nil {
log.Infof("initialized cgroup-driven watchdog")
addStopHook(stopFn)
return
}
log.Warnf("failed to initialize cgroup-driven watchdog; err: %s", err)
log.Warnf("trying a system-driven watchdog")
// 3. system-driven watchdog.
err, stopFn = watchdog.SystemDriven(0, 5*time.Second, policy) // 0 calculates the limit automatically.
if err == nil {
log.Infof("initialized system-driven watchdog")
addStopHook(stopFn)
return
}
lc.Append(fx.Hook{
OnStop: func(ctx context.Context) error {
stop()
return nil
},
})
// 4. log the failure
log.Warnf("failed to initialize system-driven watchdog; err: %s", err)
log.Warnf("system running without a memory watchdog")
}
type JwtPayload struct {
@@ -166,7 +193,7 @@ func BuiltinBootstrap() (dtypes.BootstrapPeers, error) {
func DrandBootstrap(ds dtypes.DrandSchedule) (dtypes.DrandBootstrap, error) {
// TODO: retry resolving, don't fail if at least one resolve succeeds
res := []peer.AddrInfo{}
var res []peer.AddrInfo
for _, d := range ds {
addrs, err := addrutil.ParseAddresses(context.TODO(), d.Config.Relays)
if err != nil {
node/node_test.go
+14
View File
@@ -164,6 +164,20 @@ func TestWindowedPost(t *testing.T) {
test.TestWindowPost(t, builder.MockSbBuilder, 2*time.Millisecond, 10)
}
func TestTerminate(t *testing.T) {
if os.Getenv("LOTUS_TEST_WINDOW_POST") != "1" {
t.Skip("this takes a few minutes, set LOTUS_TEST_WINDOW_POST=1 to run")
}
logging.SetLogLevel("miner", "ERROR")
logging.SetLogLevel("chainstore", "ERROR")
logging.SetLogLevel("chain", "ERROR")
logging.SetLogLevel("sub", "ERROR")
logging.SetLogLevel("storageminer", "ERROR")
test.TestTerminate(t, builder.MockSbBuilder, 2*time.Millisecond)
}
func TestCCUpgrade(t *testing.T) {
logging.SetLogLevel("miner", "ERROR")
logging.SetLogLevel("chainstore", "ERROR")