Increase link payload/ack queue sizes and make them configurable. Fixes #3706

plorenz · plorenz · commit 6e505bbb1541 · 2026-03-20T21:38:04.000-04:00
diff --git a/router/env/config.go b/router/env/config.go
@@ -151,9 +151,11 @@ type Config struct {
 		Listeners             []*CtrlListenerConfig
 	}
 	Link struct {
-		Listeners  []map[interface{}]interface{}
-		Dialers    []map[interface{}]interface{}
-		Heartbeats channel.HeartbeatOptions
+		Listeners          []map[interface{}]interface{}
+		Dialers            []map[interface{}]interface{}
+		Heartbeats         channel.HeartbeatOptions
+		PayloadSenderQueueSize int
+		AckSenderQueueSize     int
 	}
 	Dialers   map[string]xgress.OptionsData
 	Listeners []ListenerBinding
@@ -205,6 +207,9 @@ const (
 
 	DefaultLinkHeartbeatSendInterval = 10 * time.Second
 	DefaultLinkUnresponsiveTimeout   = time.Minute
+
+	DefaultLinkPayloadSenderQueueSize = 128
+	DefaultLinkAckSenderQueueSize     = 64
 )
 
 // CreateBackup will attempt to use the current path value to create a backup of
@@ -574,6 +579,8 @@ func LoadConfigWithOptions(path string, loadIdentity bool) (*Config, error) {
 	cfg.Link.Heartbeats = *channel.DefaultHeartbeatOptions()
 	cfg.Link.Heartbeats.SendInterval = DefaultLinkHeartbeatSendInterval
 	cfg.Link.Heartbeats.CloseUnresponsiveTimeout = DefaultLinkUnresponsiveTimeout
+	cfg.Link.PayloadSenderQueueSize = DefaultLinkPayloadSenderQueueSize
+	cfg.Link.AckSenderQueueSize = DefaultLinkAckSenderQueueSize
 
 	if value, found := cfgmap["link"]; found {
 		if submap, ok := value.(map[interface{}]interface{}); ok {
@@ -629,6 +636,28 @@ func LoadConfigWithOptions(path string, loadIdentity bool) (*Config, error) {
 					cfg.Link.Heartbeats = *options
 				}
 			}
+
+			if value, found := submap["payloadSenderQueueSize"]; found {
+				if intVal, ok := value.(int); ok {
+					if intVal < 1 {
+						return nil, fmt.Errorf("[link/payloadSenderQueueSize] must be at least 1, got %d", intVal)
+					}
+					cfg.Link.PayloadSenderQueueSize = intVal
+				} else {
+					return nil, fmt.Errorf("[link/payloadSenderQueueSize] must be an integer, got %T", value)
+				}
+			}
+
+			if value, found := submap["ackSenderQueueSize"]; found {
+				if intVal, ok := value.(int); ok {
+					if intVal < 1 {
+						return nil, fmt.Errorf("[link/ackSenderQueueSize] must be at least 1, got %d", intVal)
+					}
+					cfg.Link.AckSenderQueueSize = intVal
+				} else {
+					return nil, fmt.Errorf("[link/ackSenderQueueSize] must be an integer, got %T", value)
+				}
+			}
 		}
 	}
 
diff --git a/router/handler_xgress/data_plane.go b/router/handler_xgress/data_plane.go
@@ -29,15 +29,13 @@ import (
 type dataPlaneAdapter struct {
 	acker           xgress.AckSender
 	forwarder       *forwarder.Forwarder
-	retransmitter   *xgress.Retransmitter
 	payloadIngester *xgress.PayloadIngester
 	metrics         xgress.Metrics
 }
 
 type DataPlaneAdapterConfig struct {
 	Acker           xgress.AckSender
 	Forwarder       *forwarder.Forwarder
-	Retransmitter   *xgress.Retransmitter
 	PayloadIngester *xgress.PayloadIngester
 	Metrics         xgress.Metrics
 }
@@ -46,7 +44,6 @@ func NewXgressDataPlaneAdapter(cfg DataPlaneAdapterConfig) xgress.DataPlaneAdapt
 	return &dataPlaneAdapter{
 		acker:           cfg.Acker,
 		forwarder:       cfg.Forwarder,
-		retransmitter:   cfg.Retransmitter,
 		payloadIngester: cfg.PayloadIngester,
 		metrics:         cfg.Metrics,
 	}
@@ -82,10 +79,6 @@ func (adapter *dataPlaneAdapter) ForwardAcknowledgement(ack *xgress.Acknowledgem
 	adapter.acker.SendAck(ack, address)
 }
 
-func (adapter *dataPlaneAdapter) GetRetransmitter() *xgress.Retransmitter {
-	return adapter.retransmitter
-}
-
 func (adapter *dataPlaneAdapter) GetPayloadIngester() *xgress.PayloadIngester {
 	return adapter.payloadIngester
 }
diff --git a/router/router.go b/router/router.go
@@ -170,6 +170,14 @@ func (self *Router) GetMetricsRegistry() metrics.UsageRegistry {
 	return self.metricsRegistry
 }
 
+func (self *Router) GetLinkPayloadSenderQueueSize() int {
+	return self.config.Link.PayloadSenderQueueSize
+}
+
+func (self *Router) GetLinkAckSenderQueueSize() int {
+	return self.config.Link.AckSenderQueueSize
+}
+
 func (self *Router) GetXgressRegistry() *env.Registry {
 	return self.xgRegistry
 }
@@ -353,12 +361,10 @@ func (self *Router) GetAlerter() env.Alerter {
 func (self *Router) createDataPlaneAdapter() xgress.DataPlaneAdapter {
 	payloadIngester := xgress.NewPayloadIngesterWithConfig(64, self.shutdownC)
 	ackSender := xgress_router.NewAcker(self.forwarder, self.metricsRegistry, self.shutdownC)
-	retransmitter := xgress.NewRetransmitter(self.forwarder, self.metricsRegistry, self.GetCloseNotify())
 
 	return handler_xgress.NewXgressDataPlaneAdapter(handler_xgress.DataPlaneAdapterConfig{
 		Acker:           ackSender,
 		Forwarder:       self.forwarder,
-		Retransmitter:   retransmitter,
 		PayloadIngester: payloadIngester,
 		Metrics:         xgress.NewMetrics(self.metricsRegistry),
 	})
diff --git a/router/xgress_test/mock_adapter.go b/router/xgress_test/mock_adapter.go
@@ -189,11 +189,6 @@ func (self *MockDataPlaneAdapter) ForwardAcknowledgement(ack *xgress.Acknowledge
 	}
 }
 
-// GetRetransmitter returns the retransmitter from the environment
-func (self *MockDataPlaneAdapter) GetRetransmitter() *xgress.Retransmitter {
-	return self.env.GetRetransmitter()
-}
-
 // GetPayloadIngester returns the payload ingester from the environment
 func (self *MockDataPlaneAdapter) GetPayloadIngester() *xgress.PayloadIngester {
 	return self.env.GetPayloadIngester()
diff --git a/router/xgress_test/mock_env.go b/router/xgress_test/mock_env.go
@@ -3,7 +3,6 @@ package xgress_test
 import (
 	"time"
 
-	"github.com/openziti/metrics"
 	"github.com/openziti/sdk-golang/xgress"
 )
 
@@ -14,6 +13,8 @@ func (m mockFaulter) ReportForwardingFault(circuitId string, ctrlId string) {
 
 type noopMetrics struct{}
 
+func (n noopMetrics) MarkRetransmission()            {}
+func (n noopMetrics) MarkRetransmissionFailure()     {}
 func (n noopMetrics) MarkAckReceived()               {}
 func (n noopMetrics) MarkPayloadDropped()            {}
 func (n noopMetrics) MarkDuplicateAck()              {}
@@ -28,25 +29,18 @@ func (n noopMetrics) SendPayloadBuffered(int64)      {}
 func (n noopMetrics) SendPayloadDelivered(int64)     {}
 
 type MockEnv struct {
-	retransmitter   *xgress.Retransmitter
 	payloadIngester *xgress.PayloadIngester
 	metrics         xgress.Metrics
 }
 
 func NewMockEnv() *MockEnv {
 	closeNotify := make(chan struct{})
-	metricsRegistry := metrics.NewRegistry("test", nil)
 	return &MockEnv{
-		retransmitter:   xgress.NewRetransmitter(mockFaulter{}, metricsRegistry, closeNotify),
 		payloadIngester: xgress.NewPayloadIngester(closeNotify),
 		metrics:         noopMetrics{},
 	}
 }
 
-func (self *MockEnv) GetRetransmitter() *xgress.Retransmitter {
-	return self.retransmitter
-}
-
 func (self *MockEnv) GetPayloadIngester() *xgress.PayloadIngester {
 	return self.payloadIngester
 }
diff --git a/router/xlink_transport/channel.go b/router/xlink_transport/channel.go
@@ -17,11 +17,11 @@ const (
 	ChannelTypeDefault string = "link.default"
 )
 
-func NewBaseLinkChannel(underlay channel.Underlay) *BaseLinkChannel {
+func NewBaseLinkChannel(underlay channel.Underlay, payloadSenderQueueSize, ackSenderQueueSize int) *BaseLinkChannel {
 	senderContext := channel.NewSenderContext()
 
-	defaultMsgChan := make(chan channel.Sendable, 64)
-	controlMsgChan := make(chan channel.Sendable, 4)
+	defaultMsgChan := make(chan channel.Sendable, payloadSenderQueueSize)
+	controlMsgChan := make(chan channel.Sendable, ackSenderQueueSize)
 	retryMsgChan := make(chan channel.Sendable, 4)
 
 	result := &BaseLinkChannel{
@@ -127,13 +127,15 @@ type DialLinkChannelConfig struct {
 	Underlay               channel.Underlay
 	MaxDefaultChannels     int
 	MaxAckChannel          int
+	PayloadSenderQueueSize int
+	AckSenderQueueSize     int
 	StartupDelay           time.Duration
 	UnderlayChangeCallback func(ch *DialLinkChannel)
 }
 
 func NewDialLinkChannel(config DialLinkChannelConfig) UnderlayHandlerLinkChannel {
 	result := &DialLinkChannel{
-		BaseLinkChannel: *NewBaseLinkChannel(config.Underlay),
+		BaseLinkChannel: *NewBaseLinkChannel(config.Underlay, config.PayloadSenderQueueSize, config.AckSenderQueueSize),
 		dialer:          config.Dialer,
 		changeCallback:  config.UnderlayChangeCallback,
 		syncRequired:    map[string]struct{}{},
@@ -266,9 +268,9 @@ func (self *DialLinkChannel) CreateGroupedUnderlay(groupId string, groupSecret [
 	})
 }
 
-func NewListenerLinkChannel(underlay channel.Underlay) UnderlayHandlerLinkChannel {
+func NewListenerLinkChannel(underlay channel.Underlay, payloadSenderQueueSize, ackSenderQueueSize int) UnderlayHandlerLinkChannel {
 	result := &ListenerLinkChannel{
-		BaseLinkChannel: *NewBaseLinkChannel(underlay),
+		BaseLinkChannel: *NewBaseLinkChannel(underlay, payloadSenderQueueSize, ackSenderQueueSize),
 	}
 
 	result.constraints.AddConstraint(ChannelTypeDefault, 1, 1)
diff --git a/router/xlink_transport/dialer.go b/router/xlink_transport/dialer.go
@@ -283,11 +283,13 @@ func (self *dialer) dialMulti(linkId *identity.TokenId, address transport.Addres
 
 	if isGrouped, _ := channel.Headers(underlay.Headers()).GetBoolHeader(channel.IsGroupedHeader); isGrouped {
 		dialLinkChangeConfig := DialLinkChannelConfig{
-			Dialer:             linkDialer,
-			Underlay:           underlay,
-			MaxDefaultChannels: int(self.config.maxDefaultConnections),
-			MaxAckChannel:      int(self.config.maxAckConnections),
-			StartupDelay:       self.config.startupDelay,
+			Dialer:                 linkDialer,
+			Underlay:               underlay,
+			MaxDefaultChannels:     int(self.config.maxDefaultConnections),
+			MaxAckChannel:          int(self.config.maxAckConnections),
+			PayloadSenderQueueSize: self.env.GetLinkPayloadSenderQueueSize(),
+			AckSenderQueueSize:     self.env.GetLinkAckSenderQueueSize(),
+			StartupDelay:           self.config.startupDelay,
 			UnderlayChangeCallback: func(ch *DialLinkChannel) {
 				self.notifyOfLinkChange(ch, bindHandler.link)
 			},
diff --git a/router/xlink_transport/factory.go b/router/xlink_transport/factory.go
@@ -59,6 +59,8 @@ type LinkEnv interface {
 	GetRateLimiterPool() goroutines.Pool
 	GetCloseNotify() <-chan struct{}
 	GetRouterId() *identity.TokenId
+	GetLinkPayloadSenderQueueSize() int
+	GetLinkAckSenderQueueSize() int
 }
 
 func NewFactory(accepter xlink.Acceptor,
diff --git a/router/xlink_transport/listener.go b/router/xlink_transport/listener.go
@@ -91,7 +91,7 @@ func (self *listener) GetLocalBinding() string {
 }
 
 func (self *listener) handleGroupedUnderlay(underlay channel.Underlay, closeCallback func()) (channel.MultiChannel, error) {
-	linkChannel := NewListenerLinkChannel(underlay)
+	linkChannel := NewListenerLinkChannel(underlay, self.env.GetLinkPayloadSenderQueueSize(), self.env.GetLinkAckSenderQueueSize())
 	multiConfig := channel.MultiChannelConfig{
 		LogicalName:     "link/" + underlay.Id(),
 		Options:         self.config.options,
diff --git a/zititest/models/circuit-perf-diff/configs/loop-client-ert.yml.tmpl b/zititest/models/circuit-perf-diff/configs/loop-client-ert.yml.tmpl
@@ -32,25 +32,25 @@ metrics:
   clientId: {{ .Component.Id }}
 
 workloads:
-{{ if .Component.BoolVariable "testSdkXgHost" }}
+{{ if and (.Component.BoolVariable "testSdkXgHost") (.Component.BoolVariable "testThroughput") }}
   - name:         throughput-xg
     connector:    throughput-xg
     {{ $.Model.MustVariable "throughputWorkload" }}
 {{end}}
 
-{{ if .Component.BoolVariable "testErtHost" }}
+{{ if and (.Component.BoolVariable "testErtHost") (.Component.BoolVariable "testThroughput") }}
   - name:         throughput-ert
     connector:    throughput-ert
     {{ $.Model.MustVariable "throughputWorkload" }}
 {{end}}
 
-{{ if .Component.BoolVariable "testSdkXgHost" }}
+{{ if and (.Component.BoolVariable "testSdkXgHost") (.Component.BoolVariable "testLatency") }}
   - name:         latency-xg
     connector:    latency-xg
     {{ $.Model.MustVariable "latencyWorkload" }}
 {{end}}
 
-{{ if .Component.BoolVariable "testErtHost" }}
+{{ if and (.Component.BoolVariable "testErtHost") (.Component.BoolVariable "testLatency") }}
   - name:         latency-ert
     connector:    latency-ert
     {{ $.Model.MustVariable "latencyWorkload" }}
diff --git a/zititest/models/circuit-perf-diff/configs/loop-client-xg.yml.tmpl b/zititest/models/circuit-perf-diff/configs/loop-client-xg.yml.tmpl
@@ -20,28 +20,28 @@ metrics:
   clientId: {{ .Component.Id }}
 
 workloads:
-{{ if .Component.BoolVariable "testSdkXgHost" }}
+{{ if and (.Component.BoolVariable "testSdkXgHost") (.Component.BoolVariable "testThroughput") }}
   - name:         throughput-xg
     service_name: throughput-xg
     connector:    default
     {{ $.Model.MustVariable "throughputWorkload" }}
 {{end}}
 
-{{ if .Component.BoolVariable "testErtHost" }}
+{{ if and (.Component.BoolVariable "testErtHost") (.Component.BoolVariable "testThroughput") }}
   - name:         throughput-ert
     service_name: throughput-ert
     connector:    default
     {{ $.Model.MustVariable "throughputWorkload" }}
 {{end}}
 
-{{ if .Component.BoolVariable "testSdkXgHost" }}
+{{ if and (.Component.BoolVariable "testSdkXgHost") (.Component.BoolVariable "testLatency") }}
   - name:         latency-xg
     service_name: latency-xg
     connector:    default
     {{ $.Model.MustVariable "latencyWorkload" }}
 {{end}}
 
-{{ if .Component.BoolVariable "testErtHost" }}
+{{ if and (.Component.BoolVariable "testErtHost") (.Component.BoolVariable "testLatency") }}
   - name:         latency-ert
     service_name: latency-ert
     connector:    default
diff --git a/zititest/models/circuit-perf-diff/main.go b/zititest/models/circuit-perf-diff/main.go
@@ -72,6 +72,25 @@ var latencyWorkload = "" +
       payloadMaxBytes:  10000
 `
 
+// standaloneLatencyWorkload is used when testMode is "latency". Higher iteration count
+// for tighter statistics, without competing throughput traffic.
+var standaloneLatencyWorkload = "" +
+	`concurrency:  10
+    iterations:  500
+    dialer:
+      txRequests:       1
+      rxTimeout:        5s
+      payloadMinBytes:  64
+      payloadMaxBytes:  256
+      latencyFrequency: 1
+    listener:
+      txRequests:       1
+      txAfterRx:        true
+      rxTimeout:        5s
+      payloadMinBytes:  2048
+      payloadMaxBytes:  10000
+`
+
 var perfReport = NewPerfDiffReport()
 var metricsDumper = NewMetricsDumper()
 
@@ -102,6 +121,12 @@ var m = &model.Model{
 			"throughputWorkload": throughputWorkload,
 			"latencyWorkload":    latencyWorkload,
 
+			// testMode controls which workloads run. Values: "all" (default), "throughput", "latency".
+			// Override via TESTMODE env var or in code.
+			"testMode":       "all",
+			"testThroughput": true,
+			"testLatency":    true,
+
 			"testErtClient":   true,
 			"testSdkClient":   false,
 			"testSdkXgClient": true,
@@ -112,6 +137,22 @@ var m = &model.Model{
 		},
 	},
 	StructureFactories: []model.Factory{
+		// Apply testMode to select workloads. Resolved early so templates see the right values.
+		model.FactoryFunc(func(m *model.Model) error {
+			testMode, _ := m.GetStringVariable("testMode")
+			switch testMode {
+			case "latency":
+				m.Scope.PutVariable("testThroughput", false)
+				m.Scope.PutVariable("latencyWorkload", standaloneLatencyWorkload)
+			case "throughput":
+				m.Scope.PutVariable("testLatency", false)
+			case "all", "":
+				// Default: run both (no changes needed)
+			default:
+				return fmt.Errorf("unknown testMode %q, expected: all, throughput, or latency", testMode)
+			}
+			return nil
+		}),
 		model.FactoryFunc(func(m *model.Model) error {
 			// Set controller to baseline version
 			return m.ForEachComponent(".ctrl", 1, func(c *model.Component) error {

Original file line number	Diff line number	Diff line change
`@@ -189,11 +189,6 @@ func (self MockDataPlaneAdapter) ForwardAcknowledgement(ack xgress.Acknowledge`
`189`	`189`	`}`
`190`	`190`	`}`
`191`	`191`
`192`		`-// GetRetransmitter returns the retransmitter from the environment`
`193`		`-func (self MockDataPlaneAdapter) GetRetransmitter() xgress.Retransmitter {`
`194`		`- return self.env.GetRetransmitter()`
`195`		`-}`
`196`		`-`
`197`	`192`	`// GetPayloadIngester returns the payload ingester from the environment`
`198`	`193`	`func (self MockDataPlaneAdapter) GetPayloadIngester() xgress.PayloadIngester {`
`199`	`194`	`return self.env.GetPayloadIngester()`
Original file line number	Diff line number	Diff line change
`@@ -59,6 +59,8 @@ type LinkEnv interface {`
`59`	`59`	`GetRateLimiterPool() goroutines.Pool`
`60`	`60`	`GetCloseNotify() <-chan struct{}`
`61`	`61`	`GetRouterId() *identity.TokenId`
	`62`	`+ GetLinkPayloadSenderQueueSize() int`
	`63`	`+ GetLinkAckSenderQueueSize() int`
`62`	`64`	`}`
`63`	`65`
`64`	`66`	`func NewFactory(accepter xlink.Acceptor,`
Original file line number	Diff line number	Diff line change
`@@ -91,7 +91,7 @@ func (self *listener) GetLocalBinding() string {`
`91`	`91`	`}`
`92`	`92`
`93`	`93`	`func (self *listener) handleGroupedUnderlay(underlay channel.Underlay, closeCallback func()) (channel.MultiChannel, error) {`
`94`		`- linkChannel := NewListenerLinkChannel(underlay)`
	`94`	`+ linkChannel := NewListenerLinkChannel(underlay, self.env.GetLinkPayloadSenderQueueSize(), self.env.GetLinkAckSenderQueueSize())`
`95`	`95`	`multiConfig := channel.MultiChannelConfig{`
`96`	`96`	`LogicalName: "link/" + underlay.Id(),`
`97`	`97`	`Options: self.config.options,`