throttling

Documentation

Index

• func NewThrottledListener(listener net.Listener, maxConnsPerSec float64) net.Listener
• type BandwidthThrottlerConfig
• type DialThrottler
  • func NewDialThrottler(throttleLimit int) DialThrottler
  • func NewNoDialThrottler() DialThrottler
• type InboundConnUpgradeThrottler
  • func NewInboundConnUpgradeThrottler(log logging.Logger, config InboundConnUpgradeThrottlerConfig) InboundConnUpgradeThrottler
• type InboundConnUpgradeThrottlerConfig
• type InboundMsgThrottler
  • func NewInboundMsgThrottler(log logging.Logger, namespace string, registerer prometheus.Registerer, ...) (InboundMsgThrottler, error)
  • func NewNoInboundThrottler() InboundMsgThrottler
• type InboundMsgThrottlerConfig
• type MsgByteThrottlerConfig
• type OutboundMsgThrottler
  • func NewNoOutboundThrottler() OutboundMsgThrottler
  • func NewSybilOutboundMsgThrottler(log logging.Logger, namespace string, registerer prometheus.Registerer, ...) (OutboundMsgThrottler, error)
• type ReleaseFunc
• type SystemThrottler
  • func NewSystemThrottler(namespace string, reg prometheus.Registerer, config SystemThrottlerConfig, ...) (SystemThrottler, error)
• type SystemThrottlerConfig

Functions

func NewThrottledListener

func NewThrottledListener(listener net.Listener, maxConnsPerSec float64) net.Listener

Wraps [listener] and returns a net.Listener that will accept at most [maxConnsPerSec] connections per second. [maxConnsPerSec] must be non-negative.

Types

type BandwidthThrottlerConfig

type BandwidthThrottlerConfig struct {
	// Rate at which the inbound bandwidth consumable by a peer replenishes
	RefillRate uint64 `json:"bandwidthRefillRate"`
	// Max amount of consumable bandwidth that can accumulate for a given peer
	MaxBurstSize uint64 `json:"bandwidthMaxBurstRate"`
}

type DialThrottler

type DialThrottler interface {
	// Block until the event associated with this Acquire can happen.
	// If [ctx] is canceled, gives up and returns an error.
	Acquire(ctx context.Context) error
}

func NewDialThrottler

func NewDialThrottler(throttleLimit int) DialThrottler

func NewNoDialThrottler

func NewNoDialThrottler() DialThrottler

type InboundConnUpgradeThrottler

type InboundConnUpgradeThrottler interface {
	// Dispatch starts this InboundConnUpgradeThrottler.
	// Must be called before [ShouldUpgrade].
	// Blocks until [Stop] is called (i.e. should be called in a goroutine.)
	Dispatch()
	// Stop this InboundConnUpgradeThrottler and causes [Dispatch] to return.
	// Should be called when we're done with this InboundConnUpgradeThrottler.
	// This InboundConnUpgradeThrottler must not be used after [Stop] is called.
	Stop()
	// Returns whether we should upgrade an inbound connection from [ipStr].
	// Must only be called after [Dispatch] has been called.
	// If [ip] is a local IP, this method always returns true.
	// Must not be called after [Stop] has been called.
	ShouldUpgrade(ip ips.IPPort) bool
}

InboundConnUpgradeThrottler returns whether we should upgrade an inbound connection from IP [ipStr]. If ShouldUpgrade(ipStr) returns false, the connection to that IP should be closed. Note that InboundConnUpgradeThrottler rate-limits _upgrading_ of inbound connections, whereas throttledListener rate-limits _acceptance_ of inbound connections.

func NewInboundConnUpgradeThrottler

func NewInboundConnUpgradeThrottler(log logging.Logger, config InboundConnUpgradeThrottlerConfig) InboundConnUpgradeThrottler

Returns an InboundConnUpgradeThrottler that upgrades an inbound connection from a given IP at most every [UpgradeCooldown].

type InboundConnUpgradeThrottlerConfig

type InboundConnUpgradeThrottlerConfig struct {
	// ShouldUpgrade(ipStr) returns true if it has been at least [UpgradeCooldown]
	// since the last time ShouldUpgrade(ipStr) returned true or if
	// ShouldUpgrade(ipStr) has never been called.
	// If <= 0, inbound connections not rate-limited.
	UpgradeCooldown time.Duration `json:"upgradeCooldown"`
	// Maximum number of inbound connections upgraded within [UpgradeCooldown].
	// (As implemented in inboundConnUpgradeThrottler, may actually upgrade
	// [MaxRecentConnsUpgraded+1] due to a race condition but that's fine.)
	// If <= 0, inbound connections not rate-limited.
	MaxRecentConnsUpgraded int `json:"maxRecentConnsUpgraded"`
}

type InboundMsgThrottler

type InboundMsgThrottler interface {
	// Blocks until [nodeID] can read a message of size [msgSize].
	// AddNode([nodeID], ...) must have been called since
	// the last time RemoveNode([nodeID]) was called, if any.
	// It's safe for multiple goroutines to concurrently call Acquire.
	// Returns immediately if [ctx] is canceled.  The returned release function
	// needs to be called so that any allocated resources will be released
	// invariant: There should be a maximum of 1 blocking call to Acquire for a
	//            given nodeID. Callers must enforce this invariant.
	Acquire(ctx context.Context, msgSize uint64, nodeID ids.NodeID) ReleaseFunc

	// Add a new node to this throttler.
	// Must be called before Acquire(..., [nodeID]) is called.
	// RemoveNode([nodeID]) must have been called since the last time
	// AddNode([nodeID], ...) was called, if any.
	AddNode(nodeID ids.NodeID)

	// Remove a node from this throttler.
	// AddNode([nodeID], ...) must have been called since
	// the last time RemoveNode([nodeID]) was called, if any.
	// Must be called when we stop reading messages from [nodeID].
	// It's safe for multiple goroutines to concurrently call RemoveNode.
	RemoveNode(nodeID ids.NodeID)
}

InboundMsgThrottler rate-limits inbound messages from the network.

func NewInboundMsgThrottler

func NewInboundMsgThrottler(
	log logging.Logger,
	namespace string,
	registerer prometheus.Registerer,
	vdrs validators.Set,
	throttlerConfig InboundMsgThrottlerConfig,
	resourceTracker tracker.ResourceTracker,
	cpuTargeter tracker.Targeter,
	diskTargeter tracker.Targeter,
) (InboundMsgThrottler, error)

Returns a new, sybil-safe inbound message throttler.

func NewNoInboundThrottler

func NewNoInboundThrottler() InboundMsgThrottler

Returns an InboundMsgThrottler where Acquire() always returns immediately.

type InboundMsgThrottlerConfig

type InboundMsgThrottlerConfig struct {
	MsgByteThrottlerConfig   `json:"byteThrottlerConfig"`
	BandwidthThrottlerConfig `json:"bandwidthThrottlerConfig"`
	CPUThrottlerConfig       SystemThrottlerConfig `json:"cpuThrottlerConfig"`
	DiskThrottlerConfig      SystemThrottlerConfig `json:"diskThrottlerConfig"`
	MaxProcessingMsgsPerNode uint64                `json:"maxProcessingMsgsPerNode"`
}

type MsgByteThrottlerConfig

type MsgByteThrottlerConfig struct {
	VdrAllocSize        uint64 `json:"vdrAllocSize"`
	AtLargeAllocSize    uint64 `json:"atLargeAllocSize"`
	NodeMaxAtLargeBytes uint64 `json:"nodeMaxAtLargeBytes"`
}

Used by the sybil-safe inbound and outbound message throttlers

type OutboundMsgThrottler

type OutboundMsgThrottler interface {
	// Returns true if we can queue the message [msg] to be sent to node [nodeID].
	// Returns false if the message should be dropped (not sent to [nodeID]).
	// If this method returns true, Release([msg], [nodeID]) must be called (!) when
	// the message is sent (or when we give up trying to send the message, if applicable.)
	// If this method returns false, do not make a corresponding call to Release.
	Acquire(msg message.OutboundMessage, nodeID ids.NodeID) bool

	// Mark that a message [msg] has been sent to [nodeID] or we have given up
	// sending the message. Must correspond to a previous call to
	// Acquire([msg], [nodeID]) that returned true.
	Release(msg message.OutboundMessage, nodeID ids.NodeID)
}

Rate-limits outgoing messages

func NewNoOutboundThrottler

func NewNoOutboundThrottler() OutboundMsgThrottler

func NewSybilOutboundMsgThrottler

func NewSybilOutboundMsgThrottler(
	log logging.Logger,
	namespace string,
	registerer prometheus.Registerer,
	vdrs validators.Set,
	config MsgByteThrottlerConfig,
) (OutboundMsgThrottler, error)

type ReleaseFunc

type ReleaseFunc func()

type SystemThrottler

type SystemThrottler interface {
	// Blocks until we can read a message from the given peer.
	// If [ctx] is canceled, returns immediately.
	Acquire(ctx context.Context, nodeID ids.NodeID)
}

SystemThrottler rate-limits based on the system metrics usage caused by each peer. We will not read messages from peers whose messages cause excessive usage until the usage caused by the peer drops to an acceptable level.

func NewSystemThrottler

func NewSystemThrottler(
	namespace string,
	reg prometheus.Registerer,
	config SystemThrottlerConfig,
	tracker tracker.Tracker,
	targeter tracker.Targeter,
) (SystemThrottler, error)

type SystemThrottlerConfig

type SystemThrottlerConfig struct {
	Clock mockable.Clock `json:"-"`
	// The maximum amount of time we'll wait before re-checking whether a call
	// to [Acquire] can return.
	MaxRecheckDelay time.Duration `json:"maxRecheckDelay"`
}