Add Resource Manager to P2P Host (#4815)

sophoah · web-flow · commit c84173a1a0aa · 2024-12-19T17:19:51.000+07:00
Add Resource Manager to P2P Host
diff --git a/cmd/bootnode/main.go b/cmd/bootnode/main.go
@@ -114,6 +114,9 @@ func main() {
 	maxConnPerIP := flag.Int("max_conn_per_ip", 10, "max connections number for same ip")
 	forceReachabilityPublic := flag.Bool("force_public", false, "forcing the local node to believe it is reachable externally")
 	connMgrHighWaterMark := flag.Int("cmg_high_watermark", 900, "connection manager trims excess connections when they pass the high watermark")
+	resourceManagerEnabled := flag.Bool("resmgr-enabled", true, "enable p2p resource manager")
+	resourceManagerMemoryLimitBytes := flag.Uint64("resmgr-memory-limit-bytes", 0, "memory limit for p2p resource manager")
+	resourceManagerFileDescriptorsLimit := flag.Uint64("resmgr-file-descriptor-limit", 0, "file descriptor limit for p2p resource manager")
 	noTransportSecurity := flag.Bool("no_transport_security", false, "disable TLS encrypted transport")
 	muxer := flag.String("muxer", "mplex, yamux", "protocol muxer to mux per-protocol streams (mplex, yamux)")
 	userAgent := flag.String("user_agent", defUserAgent, "explicitly set the user-agent, so we can differentiate from other Go libp2p users")
@@ -149,20 +152,23 @@ func main() {
 	selfPeer := p2p.Peer{IP: *ip, Port: *port}
 
 	host, err := p2p.NewHost(p2p.HostConfig{
-		Self:                     &selfPeer,
-		BLSKey:                   privKey,
-		BootNodes:                nil, // Boot nodes have no boot nodes :) Will be connected when other nodes joined
-		TrustedNodes:             nil,
-		DataStoreFile:            &dataStorePath,
-		MaxConnPerIP:             *maxConnPerIP,
-		ForceReachabilityPublic:  *forceReachabilityPublic,
-		ConnManagerHighWatermark: *connMgrHighWaterMark,
-		NoTransportSecurity:      *noTransportSecurity,
-		NAT:                      true,
-		UserAgent:                *userAgent,
-		DialTimeout:              time.Minute,
-		Muxer:                    *muxer,
-		NoRelay:                  *noRelay,
+		Self:                            &selfPeer,
+		BLSKey:                          privKey,
+		BootNodes:                       nil, // Boot nodes have no boot nodes :) Will be connected when other nodes joined
+		TrustedNodes:                    nil,
+		DataStoreFile:                   &dataStorePath,
+		MaxConnPerIP:                    *maxConnPerIP,
+		ForceReachabilityPublic:         *forceReachabilityPublic,
+		ConnManagerHighWatermark:        *connMgrHighWaterMark,
+		ResourceMgrEnabled:              *resourceManagerEnabled,
+		ResourceMgrMemoryLimitBytes:     *resourceManagerMemoryLimitBytes,
+		ResourceMgrFileDescriptorsLimit: *resourceManagerFileDescriptorsLimit,
+		NoTransportSecurity:             *noTransportSecurity,
+		NAT:                             true,
+		UserAgent:                       *userAgent,
+		DialTimeout:                     time.Minute,
+		Muxer:                           *muxer,
+		NoRelay:                         *noRelay,
 	})
 	if err != nil {
 		utils.FatalErrMsg(err, "cannot initialize network")
diff --git a/cmd/config/config_migrations.go b/cmd/config/config_migrations.go
@@ -466,6 +466,20 @@ func init() {
 		return confTree
 	}
 
+	migrations["2.6.4"] = func(confTree *toml.Tree) *toml.Tree {
+		if confTree.Get("P2P.ResourceMgrEnabled") == nil {
+			confTree.Set("P2P.ResourceMgrEnabled", defaultConfig.P2P.ResourceMgrEnabled)
+		}
+		if confTree.Get("P2P.ResourceMgrMemoryLimitBytes") == nil {
+			confTree.Set("P2P.ResourceMgrMemoryLimitBytes", defaultConfig.P2P.ResourceMgrMemoryLimitBytes)
+		}
+		if confTree.Get("Sync.ResourceMgrFileDescriptorsLimit") == nil {
+			confTree.Set("Sync.ResourceMgrFileDescriptorsLimit", defaultConfig.P2P.ResourceMgrFileDescriptorsLimit)
+		}
+		confTree.Set("Version", "2.6.5")
+		return confTree
+	}
+
 	// check that the latest version here is the same as in default.go
 	largestKey := getNextVersion(migrations)
 	if largestKey != tomlConfigVersion {
diff --git a/cmd/config/default.go b/cmd/config/default.go
@@ -9,7 +9,7 @@ import (
 	nodeconfig "github.com/harmony-one/harmony/internal/configs/node"
 )
 
-const tomlConfigVersion = "2.6.4"
+const tomlConfigVersion = "2.6.5"
 
 const (
 	defNetworkType = nodeconfig.Mainnet
@@ -30,22 +30,25 @@ var defaultConfig = harmonyconfig.HarmonyConfig{
 	Network:  GetDefaultNetworkConfig(defNetworkType),
 	Localnet: GetDefaultLocalnetConfig(),
 	P2P: harmonyconfig.P2pConfig{
-		Port:                     nodeconfig.DefaultP2PPort,
-		IP:                       nodeconfig.DefaultPublicListenIP,
-		KeyFile:                  "./.hmykey",
-		DiscConcurrency:          nodeconfig.DefaultP2PConcurrency,
-		MaxConnsPerIP:            nodeconfig.DefaultMaxConnPerIP,
-		DisablePrivateIPScan:     false,
-		MaxPeers:                 nodeconfig.DefaultMaxPeers,
-		ConnManagerLowWatermark:  nodeconfig.DefaultConnManagerLowWatermark,
-		ConnManagerHighWatermark: nodeconfig.DefaultConnManagerHighWatermark,
-		WaitForEachPeerToConnect: nodeconfig.DefaultWaitForEachPeerToConnect,
-		NoTransportSecurity:      nodeconfig.DefaultNoTransportSecurity,
-		NAT:                      nodeconfig.DefaultNAT,
-		UserAgent:                nodeconfig.DefaultUserAgent,
-		DialTimeout:              nodeconfig.DefaultDialTimeout,
-		Muxer:                    nodeconfig.DefaultMuxer,
-		NoRelay:                  nodeconfig.DefaultNoRelay,
+		Port:                            nodeconfig.DefaultP2PPort,
+		IP:                              nodeconfig.DefaultPublicListenIP,
+		KeyFile:                         "./.hmykey",
+		DiscConcurrency:                 nodeconfig.DefaultP2PConcurrency,
+		MaxConnsPerIP:                   nodeconfig.DefaultMaxConnPerIP,
+		DisablePrivateIPScan:            false,
+		MaxPeers:                        nodeconfig.DefaultMaxPeers,
+		ConnManagerLowWatermark:         nodeconfig.DefaultConnManagerLowWatermark,
+		ConnManagerHighWatermark:        nodeconfig.DefaultConnManagerHighWatermark,
+		ResourceMgrEnabled:              nodeconfig.DefaultResourceMgrEnabled,
+		ResourceMgrMemoryLimitBytes:     nodeconfig.DefaultResourceMgrMemoryLimitBytes,
+		ResourceMgrFileDescriptorsLimit: nodeconfig.DefaultResourceMgrFileDescriptorsLimit,
+		WaitForEachPeerToConnect:        nodeconfig.DefaultWaitForEachPeerToConnect,
+		NoTransportSecurity:             nodeconfig.DefaultNoTransportSecurity,
+		NAT:                             nodeconfig.DefaultNAT,
+		UserAgent:                       nodeconfig.DefaultUserAgent,
+		DialTimeout:                     nodeconfig.DefaultDialTimeout,
+		Muxer:                           nodeconfig.DefaultMuxer,
+		NoRelay:                         nodeconfig.DefaultNoRelay,
 	},
 	HTTP: harmonyconfig.HttpConfig{
 		Enabled:        true,
diff --git a/cmd/config/flags.go b/cmd/config/flags.go
@@ -66,6 +66,9 @@ var (
 		maxPeersFlag,
 		connManagerLowWatermarkFlag,
 		connManagerHighWatermarkFlag,
+		resourceManagerEnabledFlag,
+		resourceManagerMemoryLimitBytesFlag,
+		resourceManagerFileDescriptorsLimitFlag,
 		noTransportSecurityFlag,
 		natFlag,
 		userAgentFlag,
@@ -653,6 +656,21 @@ var (
 		Usage:    "highest number of connections that'll be maintained in connection manager. Set both high and low watermarks to zero to disable connection manager",
 		DefValue: defaultConfig.P2P.ConnManagerHighWatermark,
 	}
+	resourceManagerEnabledFlag = cli.BoolFlag{
+		Name:     "p2p.resmgr-enabled",
+		Usage:    "enable p2p resource manager",
+		DefValue: defaultConfig.P2P.ResourceMgrEnabled,
+	}
+	resourceManagerMemoryLimitBytesFlag = cli.Uint64Flag{
+		Name:     "p2p.resmgr-memory-limit-bytes",
+		Usage:    "memory limit for p2p resource manager",
+		DefValue: defaultConfig.P2P.ResourceMgrMemoryLimitBytes,
+	}
+	resourceManagerFileDescriptorsLimitFlag = cli.Uint64Flag{
+		Name:     "p2p.resmgr-file-descriptor-limit",
+		Usage:    "file descriptor limit for p2p resource manager",
+		DefValue: defaultConfig.P2P.ResourceMgrFileDescriptorsLimit,
+	}
 	waitForEachPeerToConnectFlag = cli.BoolFlag{
 		Name:     "p2p.wait-for-connections",
 		Usage:    "node waits for each single peer to connect and it doesn't add them to peers list after timeout",
@@ -730,7 +748,15 @@ func applyP2PFlags(cmd *cobra.Command, config *harmonyconfig.HarmonyConfig) {
 	if cli.IsFlagChanged(cmd, connManagerHighWatermarkFlag) {
 		config.P2P.ConnManagerHighWatermark = cli.GetIntFlagValue(cmd, connManagerHighWatermarkFlag)
 	}
-
+	if cli.IsFlagChanged(cmd, resourceManagerEnabledFlag) {
+		config.P2P.ResourceMgrEnabled = cli.GetBoolFlagValue(cmd, resourceManagerEnabledFlag)
+	}
+	if cli.IsFlagChanged(cmd, resourceManagerMemoryLimitBytesFlag) {
+		config.P2P.ResourceMgrMemoryLimitBytes = cli.GetUint64FlagValue(cmd, resourceManagerMemoryLimitBytesFlag)
+	}
+	if cli.IsFlagChanged(cmd, resourceManagerFileDescriptorsLimitFlag) {
+		config.P2P.ResourceMgrFileDescriptorsLimit = cli.GetUint64FlagValue(cmd, resourceManagerFileDescriptorsLimitFlag)
+	}
 	if cli.IsFlagChanged(cmd, p2pDisablePrivateIPScanFlag) {
 		config.P2P.DisablePrivateIPScan = cli.GetBoolFlagValue(cmd, p2pDisablePrivateIPScanFlag)
 	}
diff --git a/cmd/harmony/main.go b/cmd/harmony/main.go
@@ -599,25 +599,28 @@ func createGlobalConfig(hc harmonyconfig.HarmonyConfig) (*nodeconfig.ConfigType,
 	}
 
 	myHost, err = p2p.NewHost(p2p.HostConfig{
-		Self:                     &selfPeer,
-		BLSKey:                   nodeConfig.P2PPriKey,
-		BootNodes:                hc.Network.BootNodes,
-		TrustedNodes:             hc.Network.TrustedNodes,
-		DataStoreFile:            hc.P2P.DHTDataStore,
-		DiscConcurrency:          hc.P2P.DiscConcurrency,
-		MaxConnPerIP:             hc.P2P.MaxConnsPerIP,
-		DisablePrivateIPScan:     hc.P2P.DisablePrivateIPScan,
-		MaxPeers:                 hc.P2P.MaxPeers,
-		ConnManagerLowWatermark:  hc.P2P.ConnManagerLowWatermark,
-		ConnManagerHighWatermark: hc.P2P.ConnManagerHighWatermark,
-		WaitForEachPeerToConnect: hc.P2P.WaitForEachPeerToConnect,
-		ForceReachabilityPublic:  forceReachabilityPublic,
-		NoTransportSecurity:      hc.P2P.NoTransportSecurity,
-		NAT:                      hc.P2P.NAT,
-		UserAgent:                hc.P2P.UserAgent,
-		DialTimeout:              hc.P2P.DialTimeout,
-		Muxer:                    hc.P2P.Muxer,
-		NoRelay:                  hc.P2P.NoRelay,
+		Self:                            &selfPeer,
+		BLSKey:                          nodeConfig.P2PPriKey,
+		BootNodes:                       hc.Network.BootNodes,
+		TrustedNodes:                    hc.Network.TrustedNodes,
+		DataStoreFile:                   hc.P2P.DHTDataStore,
+		DiscConcurrency:                 hc.P2P.DiscConcurrency,
+		MaxConnPerIP:                    hc.P2P.MaxConnsPerIP,
+		DisablePrivateIPScan:            hc.P2P.DisablePrivateIPScan,
+		MaxPeers:                        hc.P2P.MaxPeers,
+		ConnManagerLowWatermark:         hc.P2P.ConnManagerLowWatermark,
+		ConnManagerHighWatermark:        hc.P2P.ConnManagerHighWatermark,
+		ResourceMgrEnabled:              hc.P2P.ResourceMgrEnabled,
+		ResourceMgrMemoryLimitBytes:     hc.P2P.ResourceMgrMemoryLimitBytes,
+		ResourceMgrFileDescriptorsLimit: hc.P2P.ResourceMgrFileDescriptorsLimit,
+		WaitForEachPeerToConnect:        hc.P2P.WaitForEachPeerToConnect,
+		ForceReachabilityPublic:         forceReachabilityPublic,
+		NoTransportSecurity:             hc.P2P.NoTransportSecurity,
+		NAT:                             hc.P2P.NAT,
+		UserAgent:                       hc.P2P.UserAgent,
+		DialTimeout:                     hc.P2P.DialTimeout,
+		Muxer:                           hc.P2P.Muxer,
+		NoRelay:                         hc.P2P.NoRelay,
 	})
 	if err != nil {
 		return nil, errors.Wrap(err, "cannot create P2P network host")
diff --git a/internal/configs/harmony/harmony.go b/internal/configs/harmony/harmony.go
@@ -128,6 +128,11 @@ type P2pConfig struct {
 	// using Connection Manager will be an optional feature.
 	ConnManagerLowWatermark  int
 	ConnManagerHighWatermark int
+	// P2P Resource Manager configs
+	ResourceMgrEnabled              bool
+	ResourceMgrMemoryLimitBytes     uint64
+	ResourceMgrFileDescriptorsLimit uint64
+	// Node waits for each peer to connect before gets connect to next peer
 	WaitForEachPeerToConnect bool
 	// to disable p2p security (tls and noise)
 	NoTransportSecurity bool
diff --git a/internal/configs/node/network.go b/internal/configs/node/network.go
@@ -77,6 +77,12 @@ const (
 	// When the peer count exceeds the 'high watermark', as many peers will be pruned (and
 	// their connections terminated) until 'low watermark' peers remain.
 	DefaultConnManagerHighWatermark = 192
+	// DefaultResourceMgrEnabled enables P2P host Resource Manager
+	DefaultResourceMgrEnabled = false
+	// DefaultResourceMgrMemoryLimitBytes sets the default memory limit for p2p host Resource Manager(0=unlimited)
+	DefaultResourceMgrMemoryLimitBytes = 0
+	// DefaultResourceMgrFileDescriptorsLimit sets the default limit for file descriptors in P2P host Resource Manager (0=unlimited)
+	DefaultResourceMgrFileDescriptorsLimit = 0
 	// DefaultWaitForEachPeerToConnect sets the sync configs to connect to neighbor peers one by one and waits for each peer to connect.
 	DefaultWaitForEachPeerToConnect = false
 	// DefaultNoTransportSecurity
diff --git a/p2p/host.go b/p2p/host.go
@@ -93,25 +93,28 @@ const (
 
 // HostConfig is the config structure to create a new host
 type HostConfig struct {
-	Self                     *Peer
-	BLSKey                   libp2p_crypto.PrivKey
-	BootNodes                []string
-	TrustedNodes             []string
-	DataStoreFile            *string
-	DiscConcurrency          int
-	MaxConnPerIP             int
-	DisablePrivateIPScan     bool
-	MaxPeers                 int64
-	ConnManagerLowWatermark  int
-	ConnManagerHighWatermark int
-	WaitForEachPeerToConnect bool
-	ForceReachabilityPublic  bool
-	NoTransportSecurity      bool
-	NAT                      bool
-	UserAgent                string
-	DialTimeout              time.Duration
-	Muxer                    string
-	NoRelay                  bool
+	Self                            *Peer
+	BLSKey                          libp2p_crypto.PrivKey
+	BootNodes                       []string
+	TrustedNodes                    []string
+	DataStoreFile                   *string
+	DiscConcurrency                 int
+	MaxConnPerIP                    int
+	DisablePrivateIPScan            bool
+	MaxPeers                        int64
+	ConnManagerLowWatermark         int
+	ConnManagerHighWatermark        int
+	ResourceMgrEnabled              bool
+	ResourceMgrMemoryLimitBytes     uint64
+	ResourceMgrFileDescriptorsLimit uint64
+	WaitForEachPeerToConnect        bool
+	ForceReachabilityPublic         bool
+	NoTransportSecurity             bool
+	NAT                             bool
+	UserAgent                       string
+	DialTimeout                     time.Duration
+	Muxer                           string
+	NoRelay                         bool
 }
 
 func init() {
@@ -162,6 +165,12 @@ func NewHost(cfg HostConfig) (Host, error) {
 		return nil, fmt.Errorf("failed to open connection manager: %w", err)
 	}
 
+	rmgr, err := makeResourceMgr(cfg.ResourceMgrEnabled, cfg.ResourceMgrMemoryLimitBytes, cfg.ResourceMgrFileDescriptorsLimit, cfg.ConnManagerHighWatermark)
+	if err != nil {
+		cancel()
+		return nil, fmt.Errorf("failed to open resource manager: %w", err)
+	}
+
 	// relay
 	var relay libp2p_config.Option
 	if cfg.NoRelay {
@@ -196,6 +205,8 @@ func NewHost(cfg HostConfig) (Host, error) {
 		*/
 		// Connection manager
 		connMngr,
+		// resource manager
+		libp2p.ResourceManager(rmgr),
 		// NAT manager
 		libp2p.NATManager(nat),
 		// Band width Reporter
diff --git a/p2p/resource_manager.go b/p2p/resource_manager.go
