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

resolve merge conflicts with master

Browse Source
This commit is contained in:
Anton Evangelatov
2022-03-18 12:28:22 +01:00
parent 7cdf6dc680 a94e47c6be
commit 0d6493ec3f
194 changed files with 3519 additions and 980 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/builder.go
+6 -1
View File
@@ -33,6 +33,7 @@ import (
"github.com/filecoin-project/lotus/extern/sector-storage/stores"
"github.com/filecoin-project/lotus/journal"
"github.com/filecoin-project/lotus/journal/alerting"
"github.com/filecoin-project/lotus/lib/lotuslog"
"github.com/filecoin-project/lotus/lib/peermgr"
_ "github.com/filecoin-project/lotus/lib/sigs/bls"
_ "github.com/filecoin-project/lotus/lib/sigs/secp"
@@ -221,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),
)
@@ -253,6 +254,9 @@ func Base() Option {
// Config sets up constructors based on the provided Config
func ConfigCommon(cfg *config.Common, enableLibp2pNode bool) Option {
// setup logging early
lotuslog.SetLevelsFromConfig(cfg.Logging.SubsystemLevels)
return Options(
func(s *Settings) error { s.Config = true; return nil },
Override(new(dtypes.APIEndpoint), func() (dtypes.APIEndpoint, error) {
@@ -282,6 +286,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.go
+5
View File
@@ -47,6 +47,11 @@ func defCommon() Common {
ListenAddress: "/ip4/127.0.0.1/tcp/1234/http",
Timeout: Duration(30 * time.Second),
},
Logging: Logging{
SubsystemLevels: map[string]string{
"example-subsystem": "INFO",
},
},
Libp2p: Libp2p{
ListenAddresses: []string{
"/ip4/0.0.0.0/tcp/0",
node/config/def_test.go
+21
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
@@ -32,7 +34,26 @@ func TestDefaultFullNodeRoundtrip(t *testing.T) {
require.True(t, reflect.DeepEqual(c, c2))
}
func TestDefaultFullNodeCommentRoundtrip(t *testing.T) {
c := DefaultFullNode()
var s string
{
c, err := ConfigComment(DefaultFullNode())
require.NoError(t, err)
s = string(c)
}
c2, err := FromReader(strings.NewReader(s), DefaultFullNode())
require.NoError(t, err)
fmt.Println(s)
require.True(t, reflect.DeepEqual(c, c2))
}
func TestDefaultMinerRoundtrip(t *testing.T) {
//stm: @OTHER_IMPLEMENTATION_001
c := DefaultStorageMiner()
var s string
node/config/doc_gen.go
+20
View File
@@ -127,6 +127,12 @@ of automatically performing on-chain operations.`,
Comment: ``,
},
{
Name: "Logging",
Type: "Logging",
Comment: ``,
},
{
Name: "Libp2p",
Type: "Libp2p",
@@ -484,6 +490,14 @@ count towards this limit.`,
closed by the connection manager.`,
},
},
"Logging": []DocField{
{
Name: "SubsystemLevels",
Type: "map[string]string",
Comment: `SubsystemLevels specify per-subsystem log levels`,
},
},
"MinerAddressConfig": []DocField{
{
Name: "PreCommitControl",
@@ -736,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/doc_util.go
+1 -1
View File
@@ -16,7 +16,7 @@ func findDoc(root interface{}, section, name string) *DocField {
return findDocSect("Common", section, name)
}
func findDocSect(root string, section, name string) *DocField {
func findDocSect(root, section, name string) *DocField {
path := strings.Split(section, ".")
docSection := Doc[root]
node/config/load.go
+1 -1
View File
@@ -69,7 +69,7 @@ func ConfigUpdate(cfgCur, cfgDef interface{}, comment bool) ([]byte, error) {
}
if comment {
// create a map of default lines so we can comment those out later
// create a map of default lines, so we can comment those out later
defLines := strings.Split(defStr, "\n")
defaults := map[string]struct{}{}
for i := range defLines {
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
+14 -4
View File
@@ -13,10 +13,11 @@ import (
// Common is common config between full node and miner
type Common struct {
API API
Backup Backup
Libp2p Libp2p
Pubsub Pubsub
API API
Backup Backup
Logging Logging
Libp2p Libp2p
Pubsub Pubsub
}
// FullNode is a full node config
@@ -39,6 +40,12 @@ type Backup struct {
DisableMetadataLog bool
}
// Logging is the logging system config
type Logging struct {
// SubsystemLevels specify per-subsystem log levels
SubsystemLevels map[string]string
}
// StorageMiner is a miner config
type StorageMiner struct {
Common
@@ -243,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/client.go
+8 -21
View File
@@ -47,7 +47,6 @@ import (
"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"
@@ -1263,28 +1262,12 @@ func (a *API) ClientQueryAsk(ctx context.Context, p peer.ID, miner address.Addre
}
func (a *API) ClientCalcCommP(ctx context.Context, inpath string) (*api.CommPRet, error) {
// Hard-code the sector type to 32GiBV1_1, because:
// - ffiwrapper.GeneratePieceCIDFromFile requires a RegisteredSealProof
// - commP itself is sector-size independent, with rather low probability of that changing
// ( note how the final rust call is identical for every RegSP type )
// https://github.com/filecoin-project/rust-filecoin-proofs-api/blob/v5.0.0/src/seal.rs#L1040-L1050
//
// IF/WHEN this changes in the future we will have to be able to calculate
// "old style" commP, and thus will need to introduce a version switch or similar
arbitraryProofType := abi.RegisteredSealProof_StackedDrg64GiBV1_1
rdr, err := os.Open(inpath)
if err != nil {
return nil, err
}
defer rdr.Close() //nolint:errcheck
stat, err := rdr.Stat()
if err != nil {
return nil, err
}
// check that the data is a car file; if it's not, retrieval won't work
_, err = car.ReadHeader(bufio.NewReader(rdr))
if err != nil {
@@ -1295,16 +1278,20 @@ func (a *API) ClientCalcCommP(ctx context.Context, inpath string) (*api.CommPRet
return nil, xerrors.Errorf("seek to start: %w", err)
}
pieceReader, pieceSize := padreader.New(rdr, uint64(stat.Size()))
commP, err := ffiwrapper.GeneratePieceCIDFromFile(arbitraryProofType, pieceReader, pieceSize)
w := &writer.Writer{}
_, err = io.CopyBuffer(w, rdr, make([]byte, writer.CommPBuf))
if err != nil {
return nil, xerrors.Errorf("copy into commp writer: %w", err)
}
commp, err := w.Sum()
if err != nil {
return nil, xerrors.Errorf("computing commP failed: %w", err)
}
return &api.CommPRet{
Root: commP,
Size: pieceSize,
Root: commp.PieceCID,
Size: commp.PieceSize.Unpadded(),
}, nil
}
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/impl/storminer.go
+4 -4
View File
@@ -676,18 +676,18 @@ func (tc *transferConverter) convertTransfer(channelID datatransfer.ChannelID, h
}
var channelIDPtr *datatransfer.ChannelID
if !hasChannelID {
diagnostics = append(diagnostics, fmt.Sprintf("No data transfer channel id for GraphSync request ID %d", requestID))
diagnostics = append(diagnostics, fmt.Sprintf("No data transfer channel id for GraphSync request ID %s", requestID))
} else {
channelIDPtr = &channelID
if isCurrentChannelRequest && !hasState {
diagnostics = append(diagnostics, fmt.Sprintf("No current request state for data transfer channel id %s", channelID))
} else if !isCurrentChannelRequest && hasState {
diagnostics = append(diagnostics, fmt.Sprintf("Graphsync request %d is a previous request on data transfer channel id %s that was restarted, but it is still running", requestID, channelID))
diagnostics = append(diagnostics, fmt.Sprintf("Graphsync request %s is a previous request on data transfer channel id %s that was restarted, but it is still running", requestID, channelID))
}
}
diagnostics = append(diagnostics, tc.gsDiagnostics[requestID]...)
transfer := &api.GraphSyncDataTransfer{
RequestID: requestID,
RequestID: &requestID,
RequestState: stateString,
IsCurrentChannelRequest: isCurrentChannelRequest,
ChannelID: channelIDPtr,
@@ -717,7 +717,7 @@ func (tc *transferConverter) collectRemainingTransfers() {
channelID := channelID
cs := api.NewDataTransferChannel(channelState.SelfPeer(), channelState)
transfer := &api.GraphSyncDataTransfer{
RequestID: graphsync.RequestID(-1),
RequestID: nil,
RequestState: "graphsync state unknown",
IsCurrentChannelRequest: false,
ChannelID: &channelID,
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
@@ -918,6 +918,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,
@@ -957,6 +958,7 @@ func ToSealingConfig(dealmakingCfg config.DealmakingConfig, sealingCfg config.Se
MakeNewSectorForDeals: dealmakingCfg.MakeNewSectorForDeals,
CommittedCapacitySectorLifetime: time.Duration(sealingCfg.CommittedCapacitySectorLifetime),
WaitDealsDelay: time.Duration(sealingCfg.WaitDealsDelay),
MakeCCSectorsAvailable: sealingCfg.MakeCCSectorsAvailable,
AlwaysKeepUnsealedCopy: sealingCfg.AlwaysKeepUnsealedCopy,
FinalizeEarly: sealingCfg.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.