@@ -5,8 +5,13 @@ package p2p
55
66import (
77 "context"
8+ "errors"
9+ "fmt"
10+ "math"
11+ "sync"
812 "time"
913
14+ "github.com/prometheus/client_golang/prometheus"
1015 "go.uber.org/zap"
1116
1217 "github.com/ava-labs/avalanchego/ids"
2732 _ Handler = (* NoOpHandler )(nil )
2833 _ Handler = (* TestHandler )(nil )
2934 _ Handler = (* ValidatorHandler )(nil )
35+
36+ errPeriodMustBePositive = errors .New ("period must be positive" )
37+ errRequestsPerPeerMustBeNonNegative = errors .New ("requests-per-peer must be non-negative" )
3038)
3139
3240// Handler is the server-side logic for virtual machine application protocols.
@@ -57,6 +65,117 @@ func (NoOpHandler) AppRequest(context.Context, ids.NodeID, time.Time, []byte) ([
5765 return nil , nil
5866}
5967
68+ type DynamicThrottlerHandler struct {
69+ handler * ThrottlerHandler
70+ validatorSet ValidatorSet
71+ requestsPerPeer float64
72+
73+ throttler * SlidingWindowThrottler
74+ throttleLimitMetric prometheus.Gauge
75+ lock sync.Mutex
76+ prevNumConnectedValidators int
77+ }
78+
79+ func (d * DynamicThrottlerHandler ) AppGossip (
80+ ctx context.Context ,
81+ nodeID ids.NodeID ,
82+ gossipBytes []byte ,
83+ ) {
84+ d .checkUpdateThrottlingLimit (ctx )
85+
86+ d .handler .AppGossip (ctx , nodeID , gossipBytes )
87+ }
88+
89+ func (d * DynamicThrottlerHandler ) AppRequest (
90+ ctx context.Context ,
91+ nodeID ids.NodeID ,
92+ deadline time.Time ,
93+ requestBytes []byte ,
94+ ) ([]byte , * common.AppError ) {
95+ d .checkUpdateThrottlingLimit (ctx )
96+
97+ return d .handler .AppRequest (ctx , nodeID , deadline , requestBytes )
98+ }
99+
100+ func (d * DynamicThrottlerHandler ) checkUpdateThrottlingLimit (ctx context.Context ) {
101+ d .lock .Lock ()
102+ defer d .lock .Unlock ()
103+
104+ numValidators := d .validatorSet .Len (ctx )
105+
106+ if numValidators == d .prevNumConnectedValidators {
107+ return
108+ }
109+
110+ d .prevNumConnectedValidators = numValidators
111+
112+ if numValidators == 0 {
113+ d .setLimit (0 )
114+ return
115+ }
116+
117+ n := float64 (numValidators )
118+
119+ // guaranteed to not overflow an int
120+ expectedSamples := d .requestsPerPeer / n
121+ variance := d .requestsPerPeer * (n - 1 ) / (n * n )
122+ stdDeviation := math .Sqrt (variance )
123+
124+ // Throttle anything beyond 4 standard deviations which should throttle
125+ // anything beyond the 99.994 percentile of expected requests.
126+ limit := expectedSamples + 4 * stdDeviation
127+ d .setLimit (limit )
128+ }
129+
130+ func (d * DynamicThrottlerHandler ) setLimit (limit float64 ) {
131+ d .throttler .setLimit (limit )
132+ d .throttleLimitMetric .Set (limit )
133+ }
134+
135+ // NewDynamicThrottlerHandler wraps a handler with defaults.
136+ // Period is the throttling evaluation period during which this node is
137+ // expecting each peer to make requestsPerPeer requests to the network. The
138+ // throttling limit is dynamically updated to be inversely proportional to the
139+ // number of connected network validators.
140+ func NewDynamicThrottlerHandler (
141+ log logging.Logger ,
142+ handler Handler ,
143+ validatorSet ValidatorSet ,
144+ period time.Duration ,
145+ requestsPerPeer float64 ,
146+ metrics prometheus.Registerer ,
147+ namespace string ,
148+ ) (* DynamicThrottlerHandler , error ) {
149+ if period <= 0 {
150+ return nil , errPeriodMustBePositive
151+ }
152+
153+ if math .IsNaN (requestsPerPeer ) || requestsPerPeer < 0 {
154+ return nil , errRequestsPerPeerMustBeNonNegative
155+ }
156+
157+ // Throttling limit will be initialized when a request is handled
158+ throttler := NewSlidingWindowThrottler (period , 0 )
159+
160+ throttleLimitMetric := prometheus .NewGauge (prometheus.GaugeOpts {
161+ Namespace : namespace ,
162+ Name : "throttle_limit" ,
163+ Help : "maximum number of requests per peer for a single throttling period" ,
164+ })
165+
166+ if err := metrics .Register (throttleLimitMetric ); err != nil {
167+ return nil , fmt .Errorf ("failed to register throttle limit metric: %w" , err )
168+ }
169+
170+ return & DynamicThrottlerHandler {
171+ handler : NewThrottlerHandler (handler , throttler , log ),
172+ validatorSet : validatorSet ,
173+ requestsPerPeer : requestsPerPeer ,
174+ throttler : throttler ,
175+ throttleLimitMetric : throttleLimitMetric ,
176+ }, nil
177+ }
178+
60179func NewValidatorHandler (
61180 handler Handler ,
62181 validatorSet ValidatorSet ,
0 commit comments