Merge pull request #5456 from oasisprotocol/peternose/bugfix/improve-…

…seed-node-performance

go/p2p: Improve seed node performance

.changelog/5456.bugfix.2.md

@@ -0,0 +1 @@
+go/p2p: Increase incoming connection limit for seed nodes

.changelog/5456.bugfix.md

@@ -0,0 +1,7 @@
+go/p2p: Close connection to seed node after every request
+
+Bootstrap client, which is responsible for peer discovery and advertisement,
+now terminates connection to the seed node after every request. This action
+should free up recourses (e.g. inbound/outbound connections) on both sides
+without affecting performance since discovered peers are cached (see retention
+period) and advertisement is done infrequently (see TTL).

go/p2p/discovery/bootstrap/client.go

@@ -133,6 +133,13 @@ func (c *client) Advertise(ctx context.Context, ns string, _ ...discovery.Option
 
 	pf.RecordSuccess()
 
+	// Close connections after every call because requests to the seed node are infrequent.
+	if err = c.rc.Close(c.seed.ID); err != nil {
+		c.logger.Warn("failed to close connections to seed node",
+			"err", err,
+		)
+	}
+
 	return res.TTL, nil
 }
 
@@ -239,6 +246,13 @@ func (c *client) fetchPeers(ctx context.Context, ns string, limit int) []peer.Ad
 		pf.RecordFailure()
 	}
 
+	// Close connections after every call because requests to the seed node are infrequent.
+	if err := c.rc.Close(c.seed.ID); err != nil {
+		c.logger.Warn("failed to close connections to seed node",
+			"err", err,
+		)
+	}
+
 	return cache.peers
 }
 

go/p2p/host.go

@@ -8,7 +8,9 @@ import (
 	"github.com/libp2p/go-libp2p"
 	"github.com/libp2p/go-libp2p/core"
 	"github.com/libp2p/go-libp2p/core/host"
+	"github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/peer"
+	rcmgr "github.com/libp2p/go-libp2p/p2p/host/resource-manager"
 	"github.com/libp2p/go-libp2p/p2p/net/conngater"
 	"github.com/libp2p/go-libp2p/p2p/net/connmgr"
 	"github.com/multiformats/go-multiaddr"
@@ -36,6 +38,12 @@ type HostConfig struct {
 func NewHost(cfg *HostConfig) (host.Host, *conngater.BasicConnectionGater, error) {
 	id := api.SignerToPrivKey(cfg.Signer)
 
+	// Set up a resource manager so that we can reserve more resources.
+	rm, err := NewResourceManager()
+	if err != nil {
+		return nil, nil, err
+	}
+
 	// Set up a connection manager so we can limit the number of connections.
 	cm, err := NewConnManager(&cfg.ConnManagerConfig)
 	if err != nil {
@@ -52,6 +60,7 @@ func NewHost(cfg *HostConfig) (host.Host, *conngater.BasicConnectionGater, error
 		libp2p.UserAgent(cfg.UserAgent),
 		libp2p.ListenAddrs(cfg.ListenAddr),
 		libp2p.Identity(id),
+		libp2p.ResourceManager(rm),
 		libp2p.ConnectionManager(cm),
 		libp2p.ConnectionGater(cg),
 	)
@@ -197,3 +206,34 @@ func (cfg *ConnGaterConfig) Load() error {
 
 	return nil
 }
+
+// NewResourceManager constructs a new resource manager.
+func NewResourceManager() (network.ResourceManager, error) {
+	// Use the default resource manager for non-seed nodes.
+	if config.GlobalConfig.Mode != config.ModeSeed {
+		return nil, nil
+	}
+
+	// Tweak limits for seed nodes.
+	//
+	// Note: The connection manager will trim connections when the total number of inbound and
+	// outbound connections exceeds the high watermark (default set to 130). Using autoscaling
+	// and configuring the default limit to 128 seems to be a prudent choice.
+	defaultLimits := rcmgr.DefaultLimits
+	defaultLimits.SystemBaseLimit.ConnsInbound = 128
+	defaultLimits.SystemBaseLimit.StreamsInbound = 128 * 16
+	defaultLimits.SystemLimitIncrease.ConnsInbound = 128
+	defaultLimits.SystemLimitIncrease.StreamsInbound = 128 * 16
+
+	// Add limits around included libp2p protocols.
+	libp2p.SetDefaultServiceLimits(&defaultLimits)
+
+	// Scale limits.
+	scaledLimits := defaultLimits.AutoScale()
+
+	// The resource manager expects a limiter, se we create one from our limits.
+	limiter := rcmgr.NewFixedLimiter(scaledLimits)
+
+	// Initialize the resource manager.
+	return rcmgr.NewResourceManager(limiter)
+}

go/p2p/rpc/client.go

@@ -2,6 +2,7 @@ package rpc
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"reflect"
 	"sync"
@@ -13,7 +14,7 @@ import (
 	"github.com/libp2p/go-libp2p/core/protocol"
 
 	"github.com/oasisprotocol/oasis-core/go/common/cbor"
-	"github.com/oasisprotocol/oasis-core/go/common/errors"
+	commonErrors "github.com/oasisprotocol/oasis-core/go/common/errors"
 	"github.com/oasisprotocol/oasis-core/go/common/logging"
 	"github.com/oasisprotocol/oasis-core/go/common/workerpool"
 )
@@ -252,6 +253,9 @@ type Client interface {
 		opts ...CallMultiOption,
 	) ([]interface{}, []PeerFeedback, error)
 
+	// Close closes all connections to the given peer.
+	Close(peerID core.PeerID) error
+
 	// RegisterListener subscribes the listener to the client notification events.
 	// If the listener is already registered this is a noop operation.
 	RegisterListener(l ClientListener)
@@ -273,6 +277,7 @@ type client struct {
 	logger *logging.Logger
 }
 
+// Implements Client.
 func (c *client) Call(
 	ctx context.Context,
 	peer core.PeerID,
@@ -283,6 +288,7 @@ func (c *client) Call(
 	return c.CallOne(ctx, []core.PeerID{peer}, method, body, rsp, opts...)
 }
 
+// Implements Client.
 func (c *client) CallOne(
 	ctx context.Context,
 	peers []core.PeerID,
@@ -345,6 +351,7 @@ func (c *client) CallOne(
 	return pf, err
 }
 
+// Implements Client.
 func (c *client) CallMulti(
 	ctx context.Context,
 	peers []core.PeerID,
@@ -449,7 +456,7 @@ func (c *client) timeCall(
 
 	if err != nil {
 		// If the caller canceled the context we should not degrade the peer.
-		if !errors.Is(err, context.Canceled) {
+		if !commonErrors.Is(err, context.Canceled) {
 			c.recordFailure(peerID, latency)
 		}
 
@@ -519,7 +526,7 @@ func (c *client) call(
 
 	// Decode response.
 	if rawRsp.Error != nil {
-		return errors.FromCode(rawRsp.Error.Module, rawRsp.Error.Code, rawRsp.Error.Message)
+		return commonErrors.FromCode(rawRsp.Error.Module, rawRsp.Error.Code, rawRsp.Error.Message)
 	}
 
 	if rsp != nil {
@@ -528,13 +535,25 @@ func (c *client) call(
 	return nil
 }
 
+// Implements Client.
+func (c *client) Close(peerID core.PeerID) error {
+	var errs error
+	for _, conn := range c.host.Network().ConnsToPeer(peerID) {
+		err := conn.Close()
+		errs = errors.Join(errs, err)
+	}
+	return errs
+}
+
+// Implements Client.
 func (c *client) RegisterListener(l ClientListener) {
 	c.listeners.Lock()
 	defer c.listeners.Unlock()
 
 	c.listeners.m[l] = struct{}{}
 }
 
+// Implements Client.
 func (c *client) UnregisterListener(l ClientListener) {
 	c.listeners.Lock()
 	defer c.listeners.Unlock()

go/p2p/rpc/nop.go

@@ -82,6 +82,13 @@ func (c *nopClient) CallMulti(
 	return nil, nil, errUnsupported
 }
 
+// Implements Client.
+func (c *nopClient) Close(
+	peer.ID,
+) error {
+	return nil
+}
+
 // Implements Client.
 func (c *nopClient) RegisterListener(ClientListener) {}