libp2p/go-libp2p-resource-manager
1
0
Fork 0
mirror of https://github.com/libp2p/go-libp2p-resource-manager.git synced 2026-08-22 11:33:28 +08:00
Code Issues Projects Releases Wiki Activity

introduce per protocol peer limits, don't transfer resources out of protocol when setting the service

Browse Source
This commit is contained in:
vyzo
2022-01-14 13:42:08 +02:00
parent e1701c7189
commit 90d7e860a0
4 changed files with 147 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
limit.go
View File

@@ -38,6 +38,7 @@ type Limiter interface {
GetServiceLimits(svc string) Limit
GetServicePeerLimits(svc string) Limit
GetProtocolLimits(proto protocol.ID) Limit
GetProtocolPeerLimits(proto protocol.ID) Limit
GetPeerLimits(p peer.ID) Limit
GetStreamLimits(p peer.ID) Limit
GetConnLimits() Limit
@@ -45,17 +46,20 @@ type Limiter interface {
// BasicLimiter is a limiter with fixed limits.
type BasicLimiter struct {
SystemLimits Limit
TransientLimits Limit
DefaultServiceLimits Limit
ServiceLimits map[string]Limit
ServicePeerLimits map[string]Limit
DefaultProtocolLimits Limit
ProtocolLimits map[protocol.ID]Limit
DefaultPeerLimits Limit
PeerLimits map[peer.ID]Limit
ConnLimits Limit
StreamLimits Limit
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
}
var _ Limiter = (*BasicLimiter)(nil)
@@ -106,7 +110,11 @@ func (l *BasicLimiter) GetServiceLimits(svc string) Limit {
}
func (l *BasicLimiter) GetServicePeerLimits(svc string) Limit {
return l.ServicePeerLimits[svc]
pl, ok := l.ServicePeerLimits[svc]
if !ok {
return l.DefaultServicePeerLimits
}
return pl
}
func (l *BasicLimiter) GetProtocolLimits(proto protocol.ID) Limit {
@@ -117,6 +125,14 @@ func (l *BasicLimiter) GetProtocolLimits(proto protocol.ID) Limit {
return pl
}
func (l *BasicLimiter) GetProtocolPeerLimits(proto protocol.ID) Limit {
pl, ok := l.ProtocolPeerLimits[proto]
if !ok {
return l.DefaultProtocolPeerLimits
}
return pl
}
func (l *BasicLimiter) GetPeerLimits(p peer.ID) Limit {
pl, ok := l.PeerLimits[p]
if !ok {
@@ -163,6 +179,14 @@ func DefaultServiceBaseLimit() BaseLimit {
}
}
// DefaultServicePeerBaseLimit returns the default BaseLimit per peer for Service Scopes.
func DefaultServicePeerBaseLimit() BaseLimit {
return BaseLimit{
StreamsInbound: 256,
StreamsOutbound: 512,
}
}
// DefaultProtocolBaseLimit returns the default BaseLimit for Protocol Scopes.
func DefaultProtocolBaseLimit() BaseLimit {
return BaseLimit{
@@ -171,6 +195,14 @@ func DefaultProtocolBaseLimit() BaseLimit {
}
}
// DefaultProtocolPeerBaseLimit returns the default BaseLimit per peer for Protocol Scopes.
func DefaultProtocolPeerBaseLimit() BaseLimit {
return BaseLimit{
StreamsInbound: 128,
StreamsOutbound: 256,
}
}
// DefaultPeerBaseLimit returns the default BaseLimit for Peer Scopes.
func DefaultPeerBaseLimit() BaseLimit {
return BaseLimit{

29
limit_dynamic.go
View File

@@ -99,12 +99,25 @@ func NewDynamicLimiter(memFraction float64, minMemory, maxMemory int64) *BasicLi
MemoryFraction: memFraction / 4,
BaseLimit: DefaultServiceBaseLimit(),
}
svcPeer := &DynamicLimit{
MinMemory: 16 << 20,
MaxMemory: 64 << 20,
MemoryFraction: memFraction / 16,
BaseLimit: DefaultServicePeerBaseLimit(),
}
proto := &DynamicLimit{
MinMemory: 64 << 20,
MaxMemory: 128 << 20,
MemoryFraction: memFraction / 16,
BaseLimit: DefaultProtocolBaseLimit(),
}
protoPeer := &DynamicLimit{
MinMemory: 16 << 20,
MaxMemory: 64 << 20,
MemoryFraction: memFraction / 16,
BaseLimit: DefaultProtocolPeerBaseLimit(),
}
peer := &DynamicLimit{
MinMemory: 64 << 20,
MaxMemory: 128 << 20,
@@ -121,12 +134,14 @@ func NewDynamicLimiter(memFraction float64, minMemory, maxMemory int64) *BasicLi
}
return &BasicLimiter{
SystemLimits: system,
TransientLimits: transient,
DefaultServiceLimits: svc,
DefaultProtocolLimits: proto,
DefaultPeerLimits: peer,
ConnLimits: conn,
StreamLimits: stream,
SystemLimits: system,
TransientLimits: transient,
DefaultServiceLimits: svc,
DefaultServicePeerLimits: svcPeer,
DefaultProtocolLimits: proto,
DefaultProtocolPeerLimits: protoPeer,
DefaultPeerLimits: peer,
ConnLimits: conn,
StreamLimits: stream,
}
}

24
limit_static.go
View File

@@ -85,10 +85,18 @@ func newDefaultStaticLimiter(memoryCap int64) *BasicLimiter {
Memory: memoryLimit(memoryCap/4, 64<<20, 512<<20),
BaseLimit: DefaultServiceBaseLimit(),
}
svcPeer := &StaticLimit{
Memory: memoryLimit(memoryCap/16, 16<<20, 64<<20),
BaseLimit: DefaultServicePeerBaseLimit(),
}
proto := &StaticLimit{
Memory: memoryLimit(memoryCap/16, 64<<20, 128<<20),
BaseLimit: DefaultProtocolBaseLimit(),
}
protoPeer := &StaticLimit{
Memory: memoryLimit(memoryCap/16, 16<<20, 64<<20),
BaseLimit: DefaultProtocolPeerBaseLimit(),
}
peer := &StaticLimit{
Memory: memoryLimit(memoryCap/16, 64<<20, 128<<20),
BaseLimit: DefaultPeerBaseLimit(),
@@ -103,12 +111,14 @@ func newDefaultStaticLimiter(memoryCap int64) *BasicLimiter {
}
return &BasicLimiter{
SystemLimits: system,
TransientLimits: transient,
DefaultServiceLimits: svc,
DefaultProtocolLimits: proto,
DefaultPeerLimits: peer,
ConnLimits: conn,
StreamLimits: stream,
SystemLimits: system,
TransientLimits: transient,
DefaultServiceLimits: svc,
DefaultServicePeerLimits: svcPeer,
DefaultProtocolLimits: proto,
DefaultProtocolPeerLimits: protoPeer,
DefaultPeerLimits: peer,
ConnLimits: conn,
StreamLimits: stream,
}
}

87
rcmgr.go
View File

@@ -59,6 +59,8 @@ type protocolScope struct {
proto protocol.ID
rcmgr *resourceManager
peers map[peer.ID]*resourceScope
}
var _ network.ProtocolScope = (*protocolScope)(nil)
@@ -92,6 +94,9 @@ type streamScope struct {
peer *peerScope
svc *serviceScope
proto *protocolScope
peerProtoScope *resourceScope
peerSvcScope *resourceScope
}
var _ network.StreamScope = (*streamScope)(nil)
@@ -269,6 +274,18 @@ func (r *resourceManager) gc() {
}
s.Unlock()
}
for _, s := range r.proto {
s.Lock()
for _, p := range deadPeers {
ps, ok := s.peers[p]
if ok {
ps.Done()
delete(s.peers, p)
}
}
s.Unlock()
}
}
func newSystemScope(limit Limit) *systemScope {
@@ -341,9 +358,6 @@ func (s *serviceScope) getPeerScope(p peer.ID) *resourceScope {
}
l := s.rcmgr.limits.GetServicePeerLimits(s.name)
if l == nil {
return nil
}
if s.peers == nil {
s.peers = make(map[peer.ID]*resourceScope)
@@ -360,6 +374,29 @@ func (s *protocolScope) Protocol() protocol.ID {
return s.proto
}
func (s *protocolScope) getPeerScope(p peer.ID) *resourceScope {
s.Lock()
defer s.Unlock()
ps, ok := s.peers[p]
if ok {
ps.IncRef()
return ps
}
l := s.rcmgr.limits.GetProtocolPeerLimits(s.proto)
if s.peers == nil {
s.peers = make(map[peer.ID]*resourceScope)
}
ps = newResourceScope(l, nil, fmt.Sprintf("%s.peer", s.name))
s.peers[p] = ps
ps.IncRef()
return ps
}
func (s *peerScope) Peer() peer.ID {
return s.peer
}
@@ -424,12 +461,23 @@ func (s *streamScope) SetProtocol(proto protocol.ID) error {
return err
}
s.peerProtoScope = s.proto.getPeerScope(s.peer.peer)
if err := s.peerProtoScope.ReserveForChild(stat); err != nil {
s.proto.ReleaseForChild(stat)
s.proto.DecRef()
s.proto = nil
s.peerProtoScope.DecRef()
s.peerProtoScope = nil
return err
}
s.rcmgr.transient.ReleaseForChild(stat)
s.rcmgr.transient.DecRef() // removed from constraints
// update constraints
constraints := []*resourceScope{
s.peer.resourceScope,
s.peerProtoScope,
s.proto.resourceScope,
s.rcmgr.system.resourceScope,
}
@@ -466,32 +514,25 @@ func (s *streamScope) SetService(svc string) error {
}
// get the per peer service scope constraint, if any
peerSvcScope := s.svc.getPeerScope(s.peer.peer)
if peerSvcScope != nil {
if err := peerSvcScope.ReserveForChild(stat); err != nil {
s.svc.ReleaseForChild(stat)
s.svc.DecRef()
s.svc = nil
peerSvcScope.DecRef()
return err
}
s.peerSvcScope = s.svc.getPeerScope(s.peer.peer)
if err := s.peerSvcScope.ReserveForChild(stat); err != nil {
s.svc.ReleaseForChild(stat)
s.svc.DecRef()
s.svc = nil
s.peerSvcScope.DecRef()
s.peerSvcScope = nil
return err
}
// remove resources from the protocol
s.proto.ReleaseForChild(stat)
s.proto.DecRef() // removed from constraints
// update constraints
constraints := []*resourceScope{
s.peer.resourceScope,
s.peerProtoScope,
s.peerSvcScope,
s.proto.resourceScope,
s.svc.resourceScope,
s.rcmgr.system.resourceScope,
}
if peerSvcScope != nil {
constraints = append(constraints, peerSvcScope)
}
constraints = append(constraints, s.svc.resourceScope, s.rcmgr.system.resourceScope)
s.resourceScope.constraints = constraints
return nil