instantiate resource manager in DI

node/builder.go

@@ -15,6 +15,7 @@ import (
 	logging "github.com/ipfs/go-log/v2"
 	ci "github.com/libp2p/go-libp2p-core/crypto"
 	"github.com/libp2p/go-libp2p-core/host"
+	"github.com/libp2p/go-libp2p-core/network"
 	"github.com/libp2p/go-libp2p-core/peer"
 	"github.com/libp2p/go-libp2p-core/peerstore"
 	"github.com/libp2p/go-libp2p-core/routing"
@@ -68,6 +69,7 @@ var (
 	BandwidthReporterKey = special{11} // Libp2p option
 	ConnGaterKey         = special{12} // libp2p option
 	DAGStoreKey          = special{13} // constructor returns multiple values
+	ResourceManagerKey   = special{14} // Libp2p option
 )
 
 type invoke int
@@ -215,6 +217,10 @@ var LibP2P = Options(
 	Override(ConnectionManagerKey, lp2p.ConnectionManager(50, 200, 20*time.Second, nil)),
 	Override(new(*conngater.BasicConnectionGater), lp2p.ConnGater),
 	Override(ConnGaterKey, lp2p.ConnGaterOption),
+
+	// Services (resource management)
+	Override(new(network.ResourceManager), lp2p.ResourceManager),
+	Override(ResourceManagerKey, lp2p.ResourceManagerOption),
 )
 
 func IsType(t repo.RepoType) func(s *Settings) bool {

node/modules/lp2p/rcmgr.go

@@ -0,0 +1,66 @@
+package lp2p
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+
+	"go.uber.org/fx"
+
+	"github.com/libp2p/go-libp2p"
+	"github.com/libp2p/go-libp2p-core/network"
+	rcmgr "github.com/libp2p/go-libp2p-resource-manager"
+
+	"github.com/filecoin-project/lotus/node/repo"
+)
+
+func ResourceManager(lc fx.Lifecycle, repo repo.LockedRepo) (network.ResourceManager, error) {
+	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()
+		limiter, err = rcmgr.NewDefaultLimiterFromJSON(limitsIn)
+		if err != nil {
+			return nil, fmt.Errorf("error parsing limit file: %w", err)
+		}
+
+	case errors.Is(err, os.ErrNotExist):
+		limiter = rcmgr.NewDefaultLimiter()
+
+	default:
+		return nil, err
+	}
+
+	// TODO: also set appropriate default limits for lotus protocols
+	libp2p.SetDefaultServiceLimits(limiter)
+
+	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)
+	}
+
+	return mgr, nil
+}
+
+func ResourceManagerOption(mgr network.ResourceManager) Libp2pOpts {
+	return Libp2pOpts{
+		Opts: []libp2p.Option{libp2p.ResourceManager(mgr)},
+	}
+}