feat: opt-in Swarm.ResourceMgr (go-libp2p v0.18)

config/swarm.go

@@ -49,6 +49,9 @@ type SwarmConfig struct {
 
 	// ConnMgr configures the connection manager.
 	ConnMgr ConnMgr
+
+	// ResourceMgr configures the libp2p Network Resource Manager
+	ResourceMgr ResourceMgr
 }
 
 type RelayClient struct {
@@ -129,3 +132,36 @@ type ConnMgr struct {
 	HighWater   int
 	GracePeriod string
 }
+
+// ResourceMgr defines configuration options for the libp2p Network Resource Manager
+// <https://github.com/libp2p/go-libp2p-resource-manager#readme>
+type ResourceMgr struct {
+	// Enables the Network Resource Manager feature
+	Enabled Flag `json:",omitempty"`
+
+	// Limits is a map of Resource Scope.
+	Limits map[string]ResourceMgrScopeConfig `json:",omitempty"`
+}
+
+const (
+	ResourceMgrSystemScope         = "system"
+	ResourceMgrTransientScope      = "transient"
+	ResourceMgrServiceScopePrefix  = "svc:"
+	ResourceMgrProtocolScopePrefix = "proto:"
+	ResourceMgrPeerScopePrefix     = "peer:"
+)
+
+// libp2p Network Resource Manager config for a scope (ipfs swarm stats|limit)
+type ResourceMgrScopeConfig struct {
+	Dynamic bool `json:",omitempty"`
+	// set if Dynamic is false
+	Memory int64 `json:",omitempty"`
+	// set if Dynamic is true
+	MemoryFraction float64 `json:",omitempty"`
+	MinMemory      int64   `json:",omitempty"`
+	MaxMemory      int64   `json:",omitempty"`
+
+	Streams, StreamsInbound, StreamsOutbound int
+	Conns, ConnsInbound, ConnsOutbound       int
+	FD                                       int
+}

core/commands/commands_test.go

@@ -237,11 +237,13 @@ func TestCommands(t *testing.T) {
 		"/swarm/filters",
 		"/swarm/filters/add",
 		"/swarm/filters/rm",
+		"/swarm/limit",
 		"/swarm/peers",
 		"/swarm/peering",
 		"/swarm/peering/add",
 		"/swarm/peering/ls",
 		"/swarm/peering/rm",
+		"/swarm/stats",
 		"/tar",
 		"/tar/add",
 		"/tar/cat",

core/commands/config.go

@@ -215,18 +215,20 @@ NOTE: For security reasons, this command will omit your private key and remote s
 		return cmds.EmitOnce(res, &cfg)
 	},
 	Encoders: cmds.EncoderMap{
-		cmds.Text: cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, out *map[string]interface{}) error {
-			buf, err := config.HumanOutput(out)
-			if err != nil {
-				return err
-			}
-			buf = append(buf, byte('\n'))
-			_, err = w.Write(buf)
-			return err
-		}),
+		cmds.Text: HumanJsonEncoder,
 	},
 }
 
+var HumanJsonEncoder = cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, out *map[string]interface{}) error {
+	buf, err := config.HumanOutput(out)
+	if err != nil {
+		return err
+	}
+	buf = append(buf, byte('\n'))
+	_, err = w.Write(buf)
+	return err
+})
+
 // Scrubs value and returns error if missing
 func scrubValue(m map[string]interface{}, key []string) (map[string]interface{}, error) {
 	return scrubMapInternal(m, key, false)

core/commands/swarm.go

@@ -1,7 +1,9 @@
 package commands
 
 import (
+	"bytes"
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
@@ -10,15 +12,17 @@ import (
 	"sync"
 	"time"
 
-	commands "github.com/ipfs/go-ipfs/commands"
-	cmdenv "github.com/ipfs/go-ipfs/core/commands/cmdenv"
-	repo "github.com/ipfs/go-ipfs/repo"
-	fsrepo "github.com/ipfs/go-ipfs/repo/fsrepo"
+	files "github.com/ipfs/go-ipfs-files"
+	"github.com/ipfs/go-ipfs/commands"
+	"github.com/ipfs/go-ipfs/config"
+	"github.com/ipfs/go-ipfs/core/commands/cmdenv"
+	"github.com/ipfs/go-ipfs/core/node/libp2p"
+	"github.com/ipfs/go-ipfs/repo"
+	"github.com/ipfs/go-ipfs/repo/fsrepo"
 
 	cmds "github.com/ipfs/go-ipfs-cmds"
-	config "github.com/ipfs/go-ipfs/config"
 	inet "github.com/libp2p/go-libp2p-core/network"
-	peer "github.com/libp2p/go-libp2p-core/peer"
+	"github.com/libp2p/go-libp2p-core/peer"
 	ma "github.com/multiformats/go-multiaddr"
 	madns "github.com/multiformats/go-multiaddr-dns"
 	mamask "github.com/whyrusleeping/multiaddr-filter"
@@ -52,6 +56,8 @@ ipfs peers in the internet.
 		"filters":    swarmFiltersCmd,
 		"peers":      swarmPeersCmd,
 		"peering":    swarmPeeringCmd,
+		"stats":      swarmStatsCmd, // libp2p Network Resource Manager
+		"limit":      swarmLimitCmd, // libp2p Network Resource Manager
 	},
 }
 
@@ -304,6 +310,133 @@ var swarmPeersCmd = &cmds.Command{
 	Type: connInfos{},
 }
 
+var swarmStatsCmd = &cmds.Command{
+	Status: cmds.Experimental,
+	Helptext: cmds.HelpText{
+		Tagline: "Report resource usage for a scope.",
+		LongDescription: `Report resource usage for a scope.
+The scope can be one of the following:
+- system        -- reports the system aggregate resource usage.
+- transient     -- reports the transient resource usage.
+- svc:<service> -- reports the resource usage of a specific service.
+- proto:<proto> -- reports the resource usage of a specific protocol.
+- peer:<peer>   -- reports the resource usage of a specific peer.
+- all           -- reports the resource usage for all currently active scopes.
+
+The output of this command is JSON.
+`},
+	Arguments: []cmds.Argument{
+		cmds.StringArg("scope", true, false, "scope of the stat report"),
+	},
+	Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
+		node, err := cmdenv.GetNode(env)
+		if err != nil {
+			return err
+		}
+
+		if node.ResourceManager == nil {
+			return fmt.Errorf("no resource manager available, make sure the daemon is running")
+		}
+
+		if len(req.Arguments) != 1 {
+			return fmt.Errorf("must specify exactly one scope")
+		}
+		scope := req.Arguments[0]
+		result, err := libp2p.NetStat(node.ResourceManager, scope)
+		if err != nil {
+			return err
+		}
+
+		b := new(bytes.Buffer)
+		enc := json.NewEncoder(b)
+		err = enc.Encode(result)
+		if err != nil {
+			return err
+		}
+		return cmds.EmitOnce(res, b)
+	},
+	Encoders: cmds.EncoderMap{
+		cmds.Text: HumanJsonEncoder,
+	},
+}
+
+var swarmLimitCmd = &cmds.Command{
+	Status: cmds.Experimental,
+	Helptext: cmds.HelpText{
+		Tagline: "Get or set resource limits for a scope.",
+		LongDescription: `Get or set resource limits for a scope.
+The scope can be one of the following:
+- system        -- limits for the system aggregate resource usage.
+- transient     -- limits for the transient resource usage.
+- svc:<service> -- limits for the resource usage of a specific service.
+- proto:<proto> -- limits for the resource usage of a specific protocol.
+- peer:<peer>   -- limits for the resource usage of a specific peer.
+
+The output of this command is JSON.
+
+It is possible to use this command to inspect and tweak limits at runtime:
+
+	$ ipfs swarm limit system > limit.json
+	$ vi limit.json
+	$ ipfs swarm limit system limit.json
+
+Changes made via command line are discarded on node shutdown.
+For permanent limits set Swarm.ResourceMgr.Limits in the $IPFS_PATH/config file.
+`},
+	Arguments: []cmds.Argument{
+		cmds.StringArg("scope", true, false, "scope of the limit"),
+		cmds.FileArg("limit.json", false, false, "limits to be set").EnableStdin(),
+	},
+	Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
+		node, err := cmdenv.GetNode(env)
+		if err != nil {
+			return err
+		}
+
+		if node.ResourceManager == nil {
+			return fmt.Errorf("no resource manager available: make sure the daemon is running with Swarm.ResourceMgr.Enabled in the config")
+		}
+
+		scope := req.Arguments[0]
+
+		//  set scope limit to new values (when limit.json is passed as a second arg)
+		if req.Files != nil {
+			var newLimit config.ResourceMgrScopeConfig
+			it := req.Files.Entries()
+			if it.Next() {
+				file := files.FileFromEntry(it)
+				if file == nil {
+					return errors.New("expected a JSON file")
+				}
+				if err := json.NewDecoder(file).Decode(&newLimit); err != nil {
+					return errors.New("failed to decode JSON as ResourceMgrScopeConfig")
+				}
+				return libp2p.NetSetLimit(node.ResourceManager, scope, newLimit)
+			}
+			if err := it.Err(); err != nil {
+				return fmt.Errorf("error opening limit JSON file: %w", err)
+			}
+		}
+
+		// get scope limit
+		result, err := libp2p.NetLimit(node.ResourceManager, scope)
+		if err != nil {
+			return err
+		}
+
+		b := new(bytes.Buffer)
+		enc := json.NewEncoder(b)
+		err = enc.Encode(result)
+		if err != nil {
+			return err
+		}
+		return cmds.EmitOnce(res, b)
+	},
+	Encoders: cmds.EncoderMap{
+		cmds.Text: HumanJsonEncoder,
+	},
+}
+
 type streamInfo struct {
 	Protocol string
 }

core/core.go

@@ -30,6 +30,7 @@ import (
 	ic "github.com/libp2p/go-libp2p-core/crypto"
 	p2phost "github.com/libp2p/go-libp2p-core/host"
 	metrics "github.com/libp2p/go-libp2p-core/metrics"
+	"github.com/libp2p/go-libp2p-core/network"
 	peer "github.com/libp2p/go-libp2p-core/peer"
 	pstore "github.com/libp2p/go-libp2p-core/peerstore"
 	routing "github.com/libp2p/go-libp2p-core/routing"
@@ -85,17 +86,18 @@ type IpfsNode struct {
 	RecordValidator      record.Validator
 
 	// Online
-	PeerHost      p2phost.Host            `optional:"true"` // the network host (server+client)
-	Peering       *peering.PeeringService `optional:"true"`
-	Filters       *ma.Filters             `optional:"true"`
-	Bootstrapper  io.Closer               `optional:"true"` // the periodic bootstrapper
-	Routing       routing.Routing         `optional:"true"` // the routing system. recommend ipfs-dht
-	DNSResolver   *madns.Resolver         // the DNS resolver
-	Exchange      exchange.Interface      // the block exchange + strategy (bitswap)
-	Namesys       namesys.NameSystem      // the name system, resolves paths to hashes
-	Provider      provider.System         // the value provider system
-	IpnsRepub     *ipnsrp.Republisher     `optional:"true"`
-	GraphExchange graphsync.GraphExchange `optional:"true"`
+	PeerHost        p2phost.Host            `optional:"true"` // the network host (server+client)
+	Peering         *peering.PeeringService `optional:"true"`
+	Filters         *ma.Filters             `optional:"true"`
+	Bootstrapper    io.Closer               `optional:"true"` // the periodic bootstrapper
+	Routing         routing.Routing         `optional:"true"` // the routing system. recommend ipfs-dht
+	DNSResolver     *madns.Resolver         // the DNS resolver
+	Exchange        exchange.Interface      // the block exchange + strategy (bitswap)
+	Namesys         namesys.NameSystem      // the name system, resolves paths to hashes
+	Provider        provider.System         // the value provider system
+	IpnsRepub       *ipnsrp.Republisher     `optional:"true"`
+	GraphExchange   graphsync.GraphExchange `optional:"true"`
+	ResourceManager network.ResourceManager `optional:"true"`
 
 	PubSub   *pubsub.PubSub             `optional:"true"`
 	PSRouter *psrouter.PubsubValueStore `optional:"true"`

core/coreapi/test/api_test.go

@@ -23,7 +23,7 @@ import (
 	coreiface "github.com/ipfs/interface-go-ipfs-core"
 	"github.com/ipfs/interface-go-ipfs-core/tests"
 	"github.com/libp2p/go-libp2p-core/crypto"
-	peer "github.com/libp2p/go-libp2p-core/peer"
+	"github.com/libp2p/go-libp2p-core/peer"
 	mocknet "github.com/libp2p/go-libp2p/p2p/net/mock"
 )
 
@@ -32,7 +32,7 @@ const testPeerID = "QmTFauExutTsy4XP6JbMFcw2Wa9645HJt2bTqL6qYDCKfe"
 type NodeProvider struct{}
 
 func (NodeProvider) MakeAPISwarm(ctx context.Context, fullIdentity bool, n int) ([]coreiface.CoreAPI, error) {
-	mn := mocknet.New(ctx)
+	mn := mocknet.New()
 
 	nodes := make([]*core.IpfsNode, n)
 	apis := make([]coreiface.CoreAPI, n)

core/mock/mock.go

@@ -26,12 +26,10 @@ import (
 
 // NewMockNode constructs an IpfsNode for use in tests.
 func NewMockNode() (*core.IpfsNode, error) {
-	ctx := context.Background()
-
 	// effectively offline, only peer in its network
-	return core.NewNode(ctx, &core.BuildCfg{
+	return core.NewNode(context.Background(), &core.BuildCfg{
 		Online: true,
-		Host:   MockHostOption(mocknet.New(ctx)),
+		Host:   MockHostOption(mocknet.New()),
 	})
 }