From de2ba06f98f9f275afed419701fdfd80ec00d8c7 Mon Sep 17 00:00:00 2001 From: Paul Lorenz Date: Sat, 17 Oct 2026 22:07:44 -0400 Subject: [PATCH] Lift link fault handling off the control channel receive goroutine A fault was handled by spawning a goroutine per message, so a burst of faults spawned without limit. Handling them inline instead would put the burst on the receive goroutine, which carries every other default priority message on that underlay. - adds a pool for work lifted off a control channel's receive goroutine, kept separate from the rate limiter pool, whose workers park waiting for controller send capacity; sharing it would make receiving depend on the very back-pressure that lifting work off the receive goroutine exists to survive - hands fault handling to that pool, and falls back to spawning when the pool is full rather than dropping, since a dropped fault leaves a dead link marked up with no repair loop behind it - counts the fallbacks, so the pool being too small is visible rather than silent - sizes the queue generously and the worker count modestly: the work arrives in bursts, one per controller reconnect, but is per control channel and there are only ever a handful of those --- router/env/config_forwarder.go | 7 +++++++ router/env/env.go | 5 +++++ router/handler_ctrl/bind.go | 2 +- router/handler_ctrl/fault.go | 26 +++++++++++++++++++++--- router/router.go | 36 ++++++++++++++++++++++++++++++++++ 5 files changed, 72 insertions(+), 4 deletions(-) diff --git a/router/env/config_forwarder.go b/router/env/config_forwarder.go index 42ac264a4..40f2564a5 100644 --- a/router/env/config_forwarder.go +++ b/router/env/config_forwarder.go @@ -23,6 +23,13 @@ import ( ) const ( + // DefaultRxPoolQueueLength and DefaultRxPoolWorkerCount size the pool for work lifted off a control + // channel's receive goroutine. The queue is generous because the work arrives in bursts, one per + // controller reconnect, while the worker count stays modest since the work is per control channel and + // there are only ever a handful of those. + DefaultRxPoolQueueLength = 1024 + DefaultRxPoolWorkerCount = 16 + DefaultXgressCloseCheckInterval = 5 * time.Second DefaultXgressDialDwellTime = 0 DefaultFaultTxInterval = 15 * time.Second diff --git a/router/env/env.go b/router/env/env.go index 36154a005..e36d60f2d 100644 --- a/router/env/env.go +++ b/router/env/env.go @@ -49,6 +49,11 @@ type RouterEnv interface { RenderJsonConfig() (string, error) GetHeartbeatOptions() HeartbeatOptions GetRateLimiterPool() goroutines.Pool + // GetRxPool returns the pool for work lifted off a control channel's receive goroutine. Distinct from + // the rate limiter pool, whose workers park waiting for controller send capacity: sharing that pool + // would make receiving depend on the very back-pressure that lifting work off the receive goroutine is + // meant to survive. + GetRxPool() goroutines.Pool GetCtrlRateLimiter() rate.AdaptiveRateLimitTracker GetVersionInfo() versions.VersionProvider GetRouterDataModel() *common.RouterDataModel diff --git a/router/handler_ctrl/bind.go b/router/handler_ctrl/bind.go index acac5ec2d..9128e8978 100644 --- a/router/handler_ctrl/bind.go +++ b/router/handler_ctrl/bind.go @@ -116,7 +116,7 @@ func (self *bindHandler) BindChannel(binding channel.Binding) error { channel.AddReceiveHandlers(binding, self.env.GetInspectHandler()) channel.AddReceiveHandlers(binding, newCheckStaleLinksHandler(self.env)) channel.AddReceiveHandlers(binding, newSettingsHandler(self.env)) - channel.AddReceiveHandlers(binding, newFaultHandler(self.env.GetXlinkRegistry())) + channel.AddReceiveHandlers(binding, newFaultHandler(self.env)) channel.AddReceiveHandlers(binding, self.ctrlAddrChangeHandler) channel.AddReceiveHandlers(binding, self.clusterLeaderChangeHandler) diff --git a/router/handler_ctrl/fault.go b/router/handler_ctrl/fault.go index e3149cf41..50fcb07b8 100644 --- a/router/handler_ctrl/fault.go +++ b/router/handler_ctrl/fault.go @@ -19,17 +19,25 @@ package handler_ctrl import ( "github.com/michaelquigley/pfxlog" "github.com/openziti/channel/v5" + "github.com/openziti/metrics" "github.com/openziti/ziti/v2/common/pb/ctrl_pb" + "github.com/openziti/ziti/v2/router/env" "github.com/openziti/ziti/v2/router/xlink" "google.golang.org/protobuf/proto" ) type faultHandler struct { xlinkRegistry xlink.Registry + env env.RouterEnv + poolFallbacks metrics.Meter } -func newFaultHandler(registry xlink.Registry) *faultHandler { - return &faultHandler{xlinkRegistry: registry} +func newFaultHandler(routerEnv env.RouterEnv) *faultHandler { + return &faultHandler{ + xlinkRegistry: routerEnv.GetXlinkRegistry(), + env: routerEnv, + poolFallbacks: routerEnv.GetMetricsRegistry().Meter("link.fault.rx_pool_fallback"), + } } func (self *faultHandler) ContentType() int32 { @@ -45,7 +53,19 @@ func (self *faultHandler) HandleReceive(msg *channel.Message, ch channel.Channel return } - go self.handleFault(msg, ch, fault) + handle := func() { self.handleFault(msg, ch, fault) } + + // Bounded while the receive pool has room, unbounded when it does not. Spawning is the fallback rather + // than the default because a dropped fault leaves a dead link marked up with no repair loop behind it, + // and because waiting for pool capacity would put the burst on this goroutine, which also carries every + // other default priority message on this underlay. The pool is resolved here rather than at construction + // since handlers are built before the pools exist. + if pool := self.env.GetRxPool(); pool != nil && pool.QueueOrError(handle) == nil { + return + } + + self.poolFallbacks.Mark(1) + go handle() } func (self *faultHandler) handleFault(_ *channel.Message, ch channel.Channel, fault *ctrl_pb.Fault) { diff --git a/router/router.go b/router/router.go index 0d821b920..8e6459a83 100644 --- a/router/router.go +++ b/router/router.go @@ -92,6 +92,7 @@ type Router struct { config *env.Config routerCapabilities *capabilities.RouterCapabilityMask ctrls env.NetworkControllers + rxPool goroutines.Pool ctrlBindhandler channel.BindHandler faulter *forwarder.Faulter forwarder *forwarder.Forwarder @@ -667,6 +668,10 @@ func (self *Router) initGoroutinePools() error { return err } + if err := self.initRxPool(); err != nil { + return err + } + return nil } @@ -730,6 +735,37 @@ func (self *Router) GetLinkDialerPool() goroutines.Pool { return self.linkDialerPool } +// initRxPool builds the pool for work lifted off a control channel's receive goroutine. It is deliberately +// separate from the rate limiter pool: that pool's workers park waiting for controller send capacity, so +// sharing it would leave receiving to compete with, and be starved by, the back-pressure that moving work +// off the receive goroutine exists to survive. +func (self *Router) initRxPool() error { + rxPoolConfig := goroutines.PoolConfig{ + QueueSize: env.DefaultRxPoolQueueLength, + MinWorkers: 0, + MaxWorkers: env.DefaultRxPoolWorkerCount, + IdleTime: 30 * time.Second, + CloseNotify: self.GetCloseNotify(), + PanicHandler: func(err interface{}) { + pfxlog.Logger().WithField(logrus.ErrorKey, err).WithField("backtrace", string(debug.Stack())).Error("panic during control channel receive processing") + }, + } + + servermetrics.ConfigureGoroutinesPoolMetrics(&rxPoolConfig, self.GetMetricsRegistry(), "pool.router.rx") + + rxPool, err := goroutines.NewPool(rxPoolConfig) + if err != nil { + return errors.Wrap(err, "error creating control channel receive pool") + } + + self.rxPool = rxPool + return nil +} + +func (self *Router) GetRxPool() goroutines.Pool { + return self.rxPool +} + func (self *Router) GetRateLimiterPool() goroutines.Pool { return self.rateLimiterPool }