Merge pull request #48 from libp2p/rework-limits

rewrite limits to allow auto-scaling

go.mod

@@ -7,8 +7,9 @@ require (
 	github.com/libp2p/go-libp2p-core v0.19.0
 	github.com/multiformats/go-multiaddr v0.6.0
 	github.com/pbnjay/memory v0.0.0-20210728143218-7b4eea64cf58
-	github.com/stretchr/testify v1.7.0
+	github.com/stretchr/testify v1.8.0
 	go.opencensus.io v0.23.0
+	golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c
 )
 
 require (
@@ -38,6 +39,5 @@ require (
 	go.uber.org/multierr v1.6.0 // indirect
 	go.uber.org/zap v1.19.1 // indirect
 	golang.org/x/crypto v0.0.0-20210506145944-38f3c27a63bf // indirect
-	golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c // indirect
-	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -133,10 +133,13 @@ github.com/spacemonkeygo/spacelog v0.0.0-20180420211403-2296661a0572 h1:RC6RW7j+
 github.com/spacemonkeygo/spacelog v0.0.0-20180420211403-2296661a0572/go.mod h1:w0SWMsp6j9O/dk4/ZpIhL+3CkG8ofA2vuv7k+ltqUMc=
 github.com/spaolacci/murmur3 v1.1.0/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
-github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
+github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
 go.opencensus.io v0.23.0 h1:gqCw0LfLxScz8irSi8exQc7fyQ0fKQU/qnC/X8+V/1M=
 go.opencensus.io v0.23.0/go.mod h1:XItmlyltB5F7CS4xOC1DcqMoFqwtC6OG2xF7mCv7P7E=
@@ -238,7 +241,8 @@ gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.8 h1:obN1ZagJSUGI0Ek/LBmuj4SNLPfIny3KsKFopxRdj10=
 gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b h1:h8qDotaEPuJATrMmW04NCwg7v22aHH28wwpauUhK9Oo=
 gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=

limit.go

@@ -1,6 +1,9 @@
 package rcmgr
 
 import (
+	"encoding/json"
+	"io"
+
 	"github.com/libp2p/go-libp2p-core/network"
 	"github.com/libp2p/go-libp2p-core/peer"
 	"github.com/libp2p/go-libp2p-core/protocol"
@@ -20,19 +23,6 @@ type Limit interface {
 	GetConnTotalLimit() int
 	// GetFDLimit returns the file descriptor limit.
 	GetFDLimit() int
-
-	// WithMemoryLimit creates a copy of this limit object, with memory limit adjusted to
-	// the specified memFraction of its current value, bounded by minMemory and maxMemory.
-	WithMemoryLimit(memFraction float64, minMemory, maxMemory int64) Limit
-	// WithStreamLimit creates a copy of this limit object, with stream limits adjusted
-	// as specified.
-	WithStreamLimit(numStreamsIn, numStreamsOut, numStreams int) Limit
-	// WithConnLimit creates a copy of this limit object, with connetion limits adjusted
-	// as specified.
-	WithConnLimit(numConnsIn, numConnsOut, numConns int) Limit
-	// WithFDLimit creates a copy of this limit object, with file descriptor limits adjusted
-	// as specified
-	WithFDLimit(numFD int) Limit
 }
 
 // Limiter is the interface for providing limits to the resource manager.
@@ -48,25 +38,41 @@ type Limiter interface {
 	GetConnLimits() Limit
 }
 
-// BasicLimiter is a limiter with fixed limits.
-type BasicLimiter struct {
-	SystemLimits              Limit
-	TransientLimits           Limit
-	DefaultServiceLimits      Limit
-	DefaultServicePeerLimits  Limit
-	ServiceLimits             map[string]Limit
-	ServicePeerLimits         map[string]Limit
-	DefaultProtocolLimits     Limit
-	DefaultProtocolPeerLimits Limit
-	ProtocolLimits            map[protocol.ID]Limit
-	ProtocolPeerLimits        map[protocol.ID]Limit
-	DefaultPeerLimits         Limit
-	PeerLimits                map[peer.ID]Limit
-	ConnLimits                Limit
-	StreamLimits              Limit
+// NewDefaultLimiterFromJSON creates a new limiter by parsing a json configuration,
+// using the default limits for fallback.
+func NewDefaultLimiterFromJSON(in io.Reader) (Limiter, error) {
+	return NewLimiterFromJSON(in, DefaultLimits.AutoScale())
+}
+
+// NewLimiterFromJSON creates a new limiter by parsing a json configuration.
+func NewLimiterFromJSON(in io.Reader, defaults LimitConfig) (Limiter, error) {
+	cfg, err := readLimiterConfigFromJSON(in, defaults)
+	if err != nil {
+		return nil, err
+	}
+	return &fixedLimiter{cfg}, nil
+}
+
+func readLimiterConfigFromJSON(in io.Reader, defaults LimitConfig) (LimitConfig, error) {
+	var cfg LimitConfig
+	if err := json.NewDecoder(in).Decode(&cfg); err != nil {
+		return LimitConfig{}, err
+	}
+	cfg.Apply(defaults)
+	return cfg, nil
+}
+
+// fixedLimiter is a limiter with fixed limits.
+type fixedLimiter struct {
+	LimitConfig
 }
 
-var _ Limiter = (*BasicLimiter)(nil)
+var _ Limiter = (*fixedLimiter)(nil)
+
+func NewFixedLimiter(conf LimitConfig) Limiter {
+	log.Debugw("initializing new limiter with config", "limits", conf)
+	return &fixedLimiter{LimitConfig: conf}
+}
 
 // BaseLimit is a mixin type for basic resource limits.
 type BaseLimit struct {
@@ -77,13 +83,77 @@ type BaseLimit struct {
 	ConnsInbound    int
 	ConnsOutbound   int
 	FD              int
+	Memory          int64
 }
 
-// MemoryLimit is a mixin type for memory limits
-type MemoryLimit struct {
-	MemoryFraction float64
-	MinMemory      int64
-	MaxMemory      int64
+// Apply overwrites all zero-valued limits with the values of l2
+// Must not use a pointer receiver.
+func (l *BaseLimit) Apply(l2 BaseLimit) {
+	if l.Streams == 0 {
+		l.Streams = l2.Streams
+	}
+	if l.StreamsInbound == 0 {
+		l.StreamsInbound = l2.StreamsInbound
+	}
+	if l.StreamsOutbound == 0 {
+		l.StreamsOutbound = l2.StreamsOutbound
+	}
+	if l.Conns == 0 {
+		l.Conns = l2.Conns
+	}
+	if l.ConnsInbound == 0 {
+		l.ConnsInbound = l2.ConnsInbound
+	}
+	if l.ConnsOutbound == 0 {
+		l.ConnsOutbound = l2.ConnsOutbound
+	}
+	if l.Memory == 0 {
+		l.Memory = l2.Memory
+	}
+	if l.FD == 0 {
+		l.FD = l2.FD
+	}
+}
+
+// BaseLimitIncrease is the increase per GB of system memory.
+type BaseLimitIncrease struct {
+	Streams         int
+	StreamsInbound  int
+	StreamsOutbound int
+	Conns           int
+	ConnsInbound    int
+	ConnsOutbound   int
+	Memory          int64
+	FDFraction      float64
+}
+
+// Apply overwrites all zero-valued limits with the values of l2
+// Must not use a pointer receiver.
+func (l *BaseLimitIncrease) Apply(l2 BaseLimitIncrease) {
+	if l.Streams == 0 {
+		l.Streams = l2.Streams
+	}
+	if l.StreamsInbound == 0 {
+		l.StreamsInbound = l2.StreamsInbound
+	}
+	if l.StreamsOutbound == 0 {
+		l.StreamsOutbound = l2.StreamsOutbound
+	}
+	if l.Conns == 0 {
+		l.Conns = l2.Conns
+	}
+	if l.ConnsInbound == 0 {
+		l.ConnsInbound = l2.ConnsInbound
+	}
+	if l.ConnsOutbound == 0 {
+		l.ConnsOutbound = l2.ConnsOutbound
+	}
+	if l.Memory == 0 {
+		l.Memory = l2.Memory
+	}
+	if l.FDFraction == 0 {
+		l.FDFraction = l2.FDFraction
+	}
 }
 
 func (l *BaseLimit) GetStreamLimit(dir network.Direction) int {
@@ -114,74 +184,62 @@ func (l *BaseLimit) GetFDLimit() int {
 	return l.FD
 }
 
-func (l *BasicLimiter) GetSystemLimits() Limit {
-	return l.SystemLimits
+func (l *BaseLimit) GetMemoryLimit() int64 {
+	return l.Memory
 }
 
-func (l *BasicLimiter) GetTransientLimits() Limit {
-	return l.TransientLimits
+func (l *fixedLimiter) GetSystemLimits() Limit {
+	return &l.System
 }
 
-func (l *BasicLimiter) GetServiceLimits(svc string) Limit {
-	sl, ok := l.ServiceLimits[svc]
-	if !ok {
-		return l.DefaultServiceLimits
-	}
-	return sl
+func (l *fixedLimiter) GetTransientLimits() Limit {
+	return &l.Transient
 }
 
-func (l *BasicLimiter) GetServicePeerLimits(svc string) Limit {
-	pl, ok := l.ServicePeerLimits[svc]
+func (l *fixedLimiter) GetServiceLimits(svc string) Limit {
+	sl, ok := l.Service[svc]
 	if !ok {
-		return l.DefaultServicePeerLimits
+		return &l.ServiceDefault
 	}
-	return pl
+	return &sl
 }
 
-func (l *BasicLimiter) GetProtocolLimits(proto protocol.ID) Limit {
-	pl, ok := l.ProtocolLimits[proto]
+func (l *fixedLimiter) GetServicePeerLimits(svc string) Limit {
+	pl, ok := l.ServicePeer[svc]
 	if !ok {
-		return l.DefaultProtocolLimits
+		return &l.ServicePeerDefault
 	}
-	return pl
+	return &pl
 }
 
-func (l *BasicLimiter) GetProtocolPeerLimits(proto protocol.ID) Limit {
-	pl, ok := l.ProtocolPeerLimits[proto]
+func (l *fixedLimiter) GetProtocolLimits(proto protocol.ID) Limit {
+	pl, ok := l.Protocol[proto]
 	if !ok {
-		return l.DefaultProtocolPeerLimits
+		return &l.ProtocolDefault
 	}
-	return pl
+	return &pl
 }
 
-func (l *BasicLimiter) GetPeerLimits(p peer.ID) Limit {
-	pl, ok := l.PeerLimits[p]
+func (l *fixedLimiter) GetProtocolPeerLimits(proto protocol.ID) Limit {
+	pl, ok := l.ProtocolPeer[proto]
 	if !ok {
-		return l.DefaultPeerLimits
+		return &l.ProtocolPeerDefault
 	}
-	return pl
+	return &pl
 }
 
-func (l *BasicLimiter) GetStreamLimits(p peer.ID) Limit {
-	return l.StreamLimits
-}
-
-func (l *BasicLimiter) GetConnLimits() Limit {
-	return l.ConnLimits
+func (l *fixedLimiter) GetPeerLimits(p peer.ID) Limit {
+	pl, ok := l.Peer[p]
+	if !ok {
+		return &l.PeerDefault
+	}
+	return &pl
 }
 
-func (l *MemoryLimit) GetMemory(memoryCap int64) int64 {
-	return memoryLimit(memoryCap, l.MemoryFraction, l.MinMemory, l.MaxMemory)
+func (l *fixedLimiter) GetStreamLimits(_ peer.ID) Limit {
+	return &l.Stream
 }
 
-func memoryLimit(memoryCap int64, memFraction float64, minMemory, maxMemory int64) int64 {
-	memoryCap = int64(float64(memoryCap) * memFraction)
-	switch {
-	case memoryCap < minMemory:
-		return minMemory
-	case memoryCap > maxMemory:
-		return maxMemory
-	default:
-		return memoryCap
-	}
+func (l *fixedLimiter) GetConnLimits() Limit {
+	return &l.Conn
 }