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

Merge remote-tracking branch 'origin/master' into feat/post-worker

Browse Source
This commit is contained in:
Łukasz Magiera
2022-03-18 10:54:44 +01:00
parent 6401363ef1 18987f92ee
commit 6ddbe41376
169 changed files with 2319 additions and 871 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/builder.go
+2 -1
View File
@@ -222,7 +222,7 @@ var LibP2P = Options(
Override(ConnGaterKey, lp2p.ConnGaterOption),
// Services (resource management)
Override(new(network.ResourceManager), lp2p.ResourceManager),
Override(new(network.ResourceManager), lp2p.ResourceManager(200)),
Override(ResourceManagerKey, lp2p.ResourceManagerOption),
)
@@ -282,6 +282,7 @@ func ConfigCommon(cfg *config.Common, enableLibp2pNode bool) Option {
cfg.Libp2p.ConnMgrHigh,
time.Duration(cfg.Libp2p.ConnMgrGrace),
cfg.Libp2p.ProtectedPeers)),
Override(new(network.ResourceManager), lp2p.ResourceManager(cfg.Libp2p.ConnMgrHigh)),
Override(new(*pubsub.PubSub), lp2p.GossipSub),
Override(new(*config.Pubsub), &cfg.Pubsub),
node/builder_chain.go
+1 -1
View File
@@ -65,7 +65,7 @@ var ChainNode = Options(
Override(new(ffiwrapper.Verifier), ffiwrapper.ProofVerifier),
Override(new(ffiwrapper.Prover), ffiwrapper.ProofProver),
// Consensus: VM
// Consensus: LegacyVM
Override(new(vm.SyscallBuilder), vm.Syscalls),
// Consensus: Chain storage/access
node/config/def_test.go
+3
View File
@@ -1,3 +1,4 @@
//stm: #unit
package config
import (
@@ -12,6 +13,7 @@ import (
)
func TestDefaultFullNodeRoundtrip(t *testing.T) {
//stm: @OTHER_IMPLEMENTATION_001
c := DefaultFullNode()
var s string
@@ -51,6 +53,7 @@ func TestDefaultFullNodeCommentRoundtrip(t *testing.T) {
}
func TestDefaultMinerRoundtrip(t *testing.T) {
//stm: @OTHER_IMPLEMENTATION_001
c := DefaultStorageMiner()
var s string
node/config/doc_gen.go
+6
View File
@@ -750,6 +750,12 @@ avoid the relatively high cost of unsealing the data later, at the cost of more
Comment: `Run sector finalization before submitting sector proof to the chain`,
},
{
Name: "MakeCCSectorsAvailable",
Type: "bool",
Comment: `After sealing CC sectors, make them available for upgrading with deals`,
},
{
Name: "CollateralFromMinerBalance",
Type: "bool",
node/config/load_test.go
+2
View File
@@ -12,6 +12,7 @@ import (
)
func TestDecodeNothing(t *testing.T) {
//stm: @NODE_CONFIG_LOAD_FILE_002
assert := assert.New(t)
{
@@ -30,6 +31,7 @@ func TestDecodeNothing(t *testing.T) {
}
func TestParitalConfig(t *testing.T) {
//stm: @NODE_CONFIG_LOAD_FILE_003
assert := assert.New(t)
cfgString := `
[API]
node/config/types.go
+3
View File
@@ -250,6 +250,9 @@ type SealingConfig struct {
// Run sector finalization before submitting sector proof to the chain
FinalizeEarly bool
// After sealing CC sectors, make them available for upgrading with deals
MakeCCSectorsAvailable bool
// Whether to use available miner balance for sector collateral instead of sending it with each message
CollateralFromMinerBalance bool
// Minimum available balance to keep in the miner actor before sending it with messages
node/impl/client/import_test.go
+2
View File
@@ -27,6 +27,7 @@ import (
// This test uses a full "dense" CARv2, and not a filestore (positional mapping).
func TestRoundtripUnixFS_Dense(t *testing.T) {
//stm: @CLIENT_DATA_IMPORT_002
ctx := context.Background()
inputPath, inputContents := genInputFile(t)
@@ -75,6 +76,7 @@ func TestRoundtripUnixFS_Dense(t *testing.T) {
}
func TestRoundtripUnixFS_Filestore(t *testing.T) {
//stm: @CLIENT_DATA_IMPORT_001
ctx := context.Background()
a := &API{
Imports: &imports.Manager{},
node/impl/full/gas_test.go
+1
View File
@@ -13,6 +13,7 @@ import (
)
func TestMedian(t *testing.T) {
//stm: @MARKET_GAS_GET_MEDIAN_PREMIUM_001
require.Equal(t, types.NewInt(5), medianGasPremium([]GasMeta{
{big.NewInt(5), build.BlockGasTarget},
}, 1))
node/modules/lp2p/libp2p.go
+1 -1
View File
@@ -10,10 +10,10 @@ import (
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"
"github.com/libp2p/go-libp2p-core/peer"
"github.com/libp2p/go-libp2p-core/peerstore"
"github.com/libp2p/go-libp2p/p2p/net/connmgr"
"go.uber.org/fx"
)
node/modules/lp2p/rcmgr.go
+93 -41
View File
@@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
"math/bits"
"os"
"path/filepath"
@@ -15,6 +16,8 @@ import (
"github.com/libp2p/go-libp2p-core/protocol"
rcmgr "github.com/libp2p/go-libp2p-resource-manager"
logging "github.com/ipfs/go-log/v2"
"github.com/filecoin-project/lotus/metrics"
"github.com/filecoin-project/lotus/node/repo"
@@ -22,55 +25,104 @@ import (
"go.opencensus.io/tag"
)
func ResourceManager(lc fx.Lifecycle, repo repo.LockedRepo) (network.ResourceManager, error) {
var limiter *rcmgr.BasicLimiter
var opts []rcmgr.Option
func ResourceManager(connMgrHi uint) func(lc fx.Lifecycle, repo repo.LockedRepo) (network.ResourceManager, error) {
return func(lc fx.Lifecycle, repo repo.LockedRepo) (network.ResourceManager, error) {
envvar := os.Getenv("LOTUS_RCMGR")
if envvar == "" || envvar == "0" {
// TODO opt-in for now -- flip this to enabled by default once we are comfortable with testing
log.Info("libp2p resource manager is disabled")
return network.NullResourceManager, nil
}
repoPath := repo.Path()
log.Info("libp2p resource manager is enabled")
// enable debug logs for rcmgr
logging.SetLogLevel("rcmgr", "debug")
// create limiter -- parse $repo/limits.json if exists
limitsFile := filepath.Join(repoPath, "limits.json")
limitsIn, err := os.Open(limitsFile)
switch {
case err == nil:
defer limitsIn.Close() //nolint:errcheck
limiter, err = rcmgr.NewDefaultLimiterFromJSON(limitsIn)
// Adjust default limits
// - give it more memory, up to 4G, min of 1G
// - if maxconns are too high, adjust Conn/FD/Stream limits
defaultLimits := rcmgr.DefaultLimits.WithSystemMemory(.125, 1<<30, 4<<30)
maxconns := int(connMgrHi)
if 2*maxconns > defaultLimits.SystemBaseLimit.ConnsInbound {
// adjust conns to 2x to allow for two conns per peer (TCP+QUIC)
defaultLimits.SystemBaseLimit.ConnsInbound = logScale(2 * maxconns)
defaultLimits.SystemBaseLimit.ConnsOutbound = logScale(2 * maxconns)
defaultLimits.SystemBaseLimit.Conns = logScale(4 * maxconns)
defaultLimits.SystemBaseLimit.StreamsInbound = logScale(16 * maxconns)
defaultLimits.SystemBaseLimit.StreamsOutbound = logScale(64 * maxconns)
defaultLimits.SystemBaseLimit.Streams = logScale(64 * maxconns)
if 2*maxconns > defaultLimits.SystemBaseLimit.FD {
defaultLimits.SystemBaseLimit.FD = logScale(2 * maxconns)
}
defaultLimits.ServiceBaseLimit.StreamsInbound = logScale(8 * maxconns)
defaultLimits.ServiceBaseLimit.StreamsOutbound = logScale(32 * maxconns)
defaultLimits.ServiceBaseLimit.Streams = logScale(32 * maxconns)
defaultLimits.ProtocolBaseLimit.StreamsInbound = logScale(8 * maxconns)
defaultLimits.ProtocolBaseLimit.StreamsOutbound = logScale(32 * maxconns)
defaultLimits.ProtocolBaseLimit.Streams = logScale(32 * maxconns)
log.Info("adjusted default resource manager limits")
}
// initialize
var limiter *rcmgr.BasicLimiter
var opts []rcmgr.Option
repoPath := repo.Path()
// create limiter -- parse $repo/limits.json if exists
limitsFile := filepath.Join(repoPath, "limits.json")
limitsIn, err := os.Open(limitsFile)
switch {
case err == nil:
defer limitsIn.Close() //nolint:errcheck
limiter, err = rcmgr.NewLimiterFromJSON(limitsIn, defaultLimits)
if err != nil {
return nil, fmt.Errorf("error parsing limit file: %w", err)
}
case errors.Is(err, os.ErrNotExist):
limiter = rcmgr.NewStaticLimiter(defaultLimits)
default:
return nil, err
}
// TODO: also set appropriate default limits for lotus protocols
libp2p.SetDefaultServiceLimits(limiter)
opts = append(opts, rcmgr.WithMetrics(rcmgrMetrics{}))
if os.Getenv("LOTUS_DEBUG_RCMGR") != "" {
debugPath := filepath.Join(repoPath, "debug")
if err := os.MkdirAll(debugPath, 0755); err != nil {
return nil, fmt.Errorf("error creating debug directory: %w", err)
}
traceFile := filepath.Join(debugPath, "rcmgr.json.gz")
opts = append(opts, rcmgr.WithTrace(traceFile))
}
mgr, err := rcmgr.NewResourceManager(limiter, opts...)
if err != nil {
return nil, fmt.Errorf("error parsing limit file: %w", err)
return nil, fmt.Errorf("error creating resource manager: %w", err)
}
case errors.Is(err, os.ErrNotExist):
limiter = rcmgr.NewDefaultLimiter()
lc.Append(fx.Hook{
OnStop: func(_ context.Context) error {
return mgr.Close()
}})
default:
return nil, err
return mgr, nil
}
}
// TODO: also set appropriate default limits for lotus protocols
libp2p.SetDefaultServiceLimits(limiter)
opts = append(opts, rcmgr.WithMetrics(rcmgrMetrics{}))
if os.Getenv("LOTUS_DEBUG_RCMGR") != "" {
debugPath := filepath.Join(repoPath, "debug")
if err := os.MkdirAll(debugPath, 0755); err != nil {
return nil, fmt.Errorf("error creating debug directory: %w", err)
}
traceFile := filepath.Join(debugPath, "rcmgr.json.gz")
opts = append(opts, rcmgr.WithTrace(traceFile))
}
mgr, err := rcmgr.NewResourceManager(limiter, opts...)
if err != nil {
return nil, fmt.Errorf("error creating resource manager: %w", err)
}
lc.Append(fx.Hook{
OnStop: func(_ context.Context) error {
return mgr.Close()
}})
return mgr, nil
func logScale(val int) int {
bitlen := bits.Len(uint(val))
return 1 << bitlen
}
func ResourceManagerOption(mgr network.ResourceManager) Libp2pOpts {
node/modules/storageminer.go
+2
View File
@@ -897,6 +897,7 @@ func NewSetSealConfigFunc(r repo.LockedRepo) (dtypes.SetSealingConfigFunc, error
MaxSealingSectorsForDeals: cfg.MaxSealingSectorsForDeals,
CommittedCapacitySectorLifetime: config.Duration(cfg.CommittedCapacitySectorLifetime),
WaitDealsDelay: config.Duration(cfg.WaitDealsDelay),
MakeCCSectorsAvailable: cfg.MakeCCSectorsAvailable,
AlwaysKeepUnsealedCopy: cfg.AlwaysKeepUnsealedCopy,
FinalizeEarly: cfg.FinalizeEarly,
@@ -935,6 +936,7 @@ func ToSealingConfig(cfg *config.StorageMiner) sealiface.Config {
MakeNewSectorForDeals: cfg.Dealmaking.MakeNewSectorForDeals,
CommittedCapacitySectorLifetime: time.Duration(cfg.Sealing.CommittedCapacitySectorLifetime),
WaitDealsDelay: time.Duration(cfg.Sealing.WaitDealsDelay),
MakeCCSectorsAvailable: cfg.Sealing.MakeCCSectorsAvailable,
AlwaysKeepUnsealedCopy: cfg.Sealing.AlwaysKeepUnsealedCopy,
FinalizeEarly: cfg.Sealing.FinalizeEarly,
node/repo/fsrepo_test.go
+9 -12
View File
@@ -2,16 +2,11 @@
package repo
import (
"io/ioutil"
"os"
"testing"
)
func genFsRepo(t *testing.T) (*FsRepo, func()) {
path, err := ioutil.TempDir("", "lotus-repo-")
if err != nil {
t.Fatal(err)
}
func genFsRepo(t *testing.T) *FsRepo {
path := t.TempDir()
repo, err := NewFS(path)
if err != nil {
@@ -22,13 +17,15 @@ func genFsRepo(t *testing.T) (*FsRepo, func()) {
if err != ErrRepoExists && err != nil {
t.Fatal(err)
}
return repo, func() {
_ = os.RemoveAll(path)
}
return repo
}
func TestFsBasic(t *testing.T) {
repo, closer := genFsRepo(t)
defer closer()
//stm: @NODE_FS_REPO_LOCK_001,@NODE_FS_REPO_LOCK_002,@NODE_FS_REPO_UNLOCK_001
//stm: @NODE_FS_REPO_SET_API_ENDPOINT_001, @NODE_FS_REPO_GET_API_ENDPOINT_001
//stm: @NODE_FS_REPO_GET_CONFIG_001, @NODE_FS_REPO_SET_CONFIG_001
//stm: @NODE_FS_REPO_LIST_KEYS_001, @NODE_FS_REPO_PUT_KEY_001
//stm: @NODE_FS_REPO_GET_KEY_001, NODE_FS_REPO_DELETE_KEY_001
repo := genFsRepo(t)
basicTest(t, repo)
}
node/repo/memrepo_test.go
+1
View File
@@ -6,6 +6,7 @@ import (
)
func TestMemBasic(t *testing.T) {
//stm: @REPO_MEM_001
repo := NewMemory(nil)
basicTest(t, repo)
}
node/shutdown_test.go
+2
View File
@@ -1,3 +1,4 @@
//stm: #unit
package node
import (
@@ -10,6 +11,7 @@ import (
)
func TestMonitorShutdown(t *testing.T) {
//stm: @NODE_COMMON_SHUTDOWN_001
signalCh := make(chan struct{})
// Three shutdown handlers.