diff --git a/limit.go b/limit.go index 5e93f09..9d3887b 100644 --- a/limit.go +++ b/limit.go @@ -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{ diff --git a/limit_dynamic.go b/limit_dynamic.go index 174777a..9898f51 100644 --- a/limit_dynamic.go +++ b/limit_dynamic.go @@ -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, } } diff --git a/limit_static.go b/limit_static.go index d668a36..b700193 100644 --- a/limit_static.go +++ b/limit_static.go @@ -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, } } diff --git a/rcmgr.go b/rcmgr.go index f1da328..9506799 100644 --- a/rcmgr.go +++ b/rcmgr.go @@ -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