Optimize: tcp lifo

config/config.go

@@ -252,6 +252,7 @@ type Config struct {
 	BackendFlushInterval         time.Duration `yaml:"backend-flush-interval"`
 	ExperimentalUpgrade          bool          `yaml:"experimental-upgrade"`
 	ExperimentalUpgradeAudit     bool          `yaml:"experimental-upgrade-audit"`
+	TCPQueueTimeoutServer        time.Duration `yaml:"tcp-queue-timeout-server"`
 	ReadTimeoutServer            time.Duration `yaml:"read-timeout-server"`
 	ReadHeaderTimeoutServer      time.Duration `yaml:"read-header-timeout-server"`
 	WriteTimeoutServer           time.Duration `yaml:"write-timeout-server"`
@@ -576,6 +577,7 @@ func NewConfig() *Config {
 	flag.DurationVar(&cfg.BackendFlushInterval, "backend-flush-interval", 20*time.Millisecond, "flush interval for upgraded proxy connections")
 	flag.BoolVar(&cfg.ExperimentalUpgrade, "experimental-upgrade", false, "enable experimental feature to handle upgrade protocol requests")
 	flag.BoolVar(&cfg.ExperimentalUpgradeAudit, "experimental-upgrade-audit", false, "enable audit logging of the request line and the messages during the experimental web socket upgrades")
+	flag.DurationVar(&cfg.TCPQueueTimeoutServer, "tcp-queue-timeout-server", time.Second, "set timeout for how long TCP connections can be queued in http server connections")
 	flag.DurationVar(&cfg.ReadTimeoutServer, "read-timeout-server", 5*time.Minute, "set ReadTimeout for http server connections")
 	flag.DurationVar(&cfg.ReadHeaderTimeoutServer, "read-header-timeout-server", 60*time.Second, "set ReadHeaderTimeout for http server connections")
 	flag.DurationVar(&cfg.WriteTimeoutServer, "write-timeout-server", 60*time.Second, "set WriteTimeout for http server connections")
@@ -948,6 +950,7 @@ func (c *Config) ToOptions() skipper.Options {
 		BackendFlushInterval:         c.BackendFlushInterval,
 		ExperimentalUpgrade:          c.ExperimentalUpgrade,
 		ExperimentalUpgradeAudit:     c.ExperimentalUpgradeAudit,
+		TCPQueueTimeoutServer:        c.TCPQueueTimeoutServer,
 		ReadTimeoutServer:            c.ReadTimeoutServer,
 		ReadHeaderTimeoutServer:      c.ReadHeaderTimeoutServer,
 		WriteTimeoutServer:           c.WriteTimeoutServer,

config/config_test.go

@@ -129,6 +129,7 @@ func defaultConfig(with func(*Config)) *Config {
 		IdleConnsPerHost:                        64,
 		CloseIdleConnsPeriod:                    20 * time.Second,
 		BackendFlushInterval:                    20 * time.Millisecond,
+		TCPQueueTimeoutServer:                   time.Second,
 		ReadTimeoutServer:                       5 * time.Minute,
 		ReadHeaderTimeoutServer:                 1 * time.Minute,
 		WriteTimeoutServer:                      1 * time.Minute,

go.mod

@@ -68,6 +68,8 @@ require (
 	github.com/HdrHistogram/hdrhistogram-go v1.1.2 // indirect
 	github.com/Microsoft/go-winio v0.6.2 // indirect
 	github.com/agnivade/levenshtein v1.2.1 // indirect
+	github.com/amirylm/go-options v0.0.2 // indirect
+	github.com/amirylm/lockfree v0.0.4 // indirect
 	github.com/armon/go-metrics v0.4.1 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bmizerany/perks v0.0.0-20141205001514-d9a9656a3a4b // indirect
@@ -101,6 +103,7 @@ require (
 	github.com/go-ole/go-ole v1.2.6 // indirect
 	github.com/gobwas/glob v0.2.3 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang-collections/collections v0.0.0-20130729185459-604e922904d3 // indirect
 	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/google/btree v1.0.0 // indirect
 	github.com/google/flatbuffers v25.2.10+incompatible // indirect

go.sum

@@ -34,6 +34,10 @@ github.com/andreyvit/diff v0.0.0-20170406064948-c7f18ee00883 h1:bvNMNQO63//z+xNg
 github.com/andreyvit/diff v0.0.0-20170406064948-c7f18ee00883/go.mod h1:rCTlJbsFo29Kk6CurOXKm700vrz8f0KW0JNfpkRJY/8=
 github.com/andybalholm/brotli v1.2.0 h1:ukwgCxwYrmACq68yiUqwIWnGY0cTPox/M94sVwToPjQ=
 github.com/andybalholm/brotli v1.2.0/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY=
+github.com/amirylm/go-options v0.0.2 h1:OvuFcKUg3+7jdKeY54XrRnAIxP9Dmultlg9dS7Q3TpA=
+github.com/amirylm/go-options v0.0.2/go.mod h1:OmhJW65Aeyb74akzydI9SVgCjuwKlPNcTZeXk7TETPk=
+github.com/amirylm/lockfree v0.0.4 h1:SAC96Droepe6HjDqymFY3E6UyJ6GR2crOGvbXFlk+kY=
+github.com/amirylm/lockfree v0.0.4/go.mod h1:92tGIqOCCQdd9SR5nGLYwK4GN9PTKlQmRwXKxqfVz/U=
 github.com/arbovm/levenshtein v0.0.0-20160628152529-48b4e1c0c4d0 h1:jfIu9sQUG6Ig+0+Ap1h4unLjW6YQJpKZVmUzxsD4E/Q=
 github.com/arbovm/levenshtein v0.0.0-20160628152529-48b4e1c0c4d0/go.mod h1:t2tdKJDJF9BV14lnkjHmOQgcvEKgtqs5a1N3LNdJhGE=
 github.com/armon/go-metrics v0.4.1 h1:hR91U9KYmb6bLBYLQjyM+3j+rcd/UhE+G78SFnF8gJA=
@@ -169,6 +173,8 @@ github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7a
 github.com/gogo/protobuf v1.3.1/go.mod h1:SlYgWuQ5SjCEi6WLHjHCa1yvBfUnHcTbrrZtXPKa29o=
 github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
 github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
+github.com/golang-collections/collections v0.0.0-20130729185459-604e922904d3 h1:zN2lZNZRflqFyxVaTIU61KNKQ9C0055u9CAfpmqUvo4=
+github.com/golang-collections/collections v0.0.0-20130729185459-604e922904d3/go.mod h1:nPpo7qLxd6XL3hWJG/O60sR8ZKfMCiIoNap5GvD12KU=
 github.com/golang-jwt/jwt/v4 v4.4.2/go.mod h1:m21LjoU+eqJr34lmDMbreY2eSTRJ1cv77w39/MY0Ch0=
 github.com/golang-jwt/jwt/v4 v4.5.2 h1:YtQM7lnr8iZ+j5q71MGKkNw9Mn7AjHM68uc9g5fXeUI=
 github.com/golang-jwt/jwt/v4 v4.5.2/go.mod h1:m21LjoU+eqJr34lmDMbreY2eSTRJ1cv77w39/MY0Ch0=

metrics/metrics.go

@@ -84,6 +84,36 @@ type Metrics interface {
 	Close()
 }
 
+type NoMetric struct{}
+
+func (NoMetric) MeasureSince(string, time.Time)                      {}
+func (NoMetric) IncCounter(string)                                   {}
+func (NoMetric) IncCounterBy(string, int64)                          {}
+func (NoMetric) IncFloatCounterBy(string, float64)                   {}
+func (NoMetric) MeasureRouteLookup(time.Time)                        {}
+func (NoMetric) MeasureFilterCreate(string, time.Time)               {}
+func (NoMetric) MeasureFilterRequest(string, time.Time)              {}
+func (NoMetric) MeasureAllFiltersRequest(string, time.Time)          {}
+func (NoMetric) MeasureBackendRequestHeader(string, int)             {}
+func (NoMetric) MeasureBackend(string, time.Time)                    {}
+func (NoMetric) MeasureBackendHost(string, time.Time)                {}
+func (NoMetric) MeasureFilterResponse(string, time.Time)             {}
+func (NoMetric) MeasureAllFiltersResponse(string, time.Time)         {}
+func (NoMetric) MeasureResponse(int, string, string, time.Time)      {}
+func (NoMetric) MeasureResponseSize(string, int64)                   {}
+func (NoMetric) MeasureProxy(time.Duration, time.Duration)           {}
+func (NoMetric) MeasureServe(string, string, string, int, time.Time) {}
+func (NoMetric) IncRoutingFailures()                                 {}
+func (NoMetric) IncErrorsBackend(string)                             {}
+func (NoMetric) MeasureBackend5xx(time.Time)                         {}
+func (NoMetric) IncErrorsStreaming(string)                           {}
+func (NoMetric) RegisterHandler(string, *http.ServeMux)              {}
+func (NoMetric) UpdateGauge(string, float64)                         {}
+func (NoMetric) UpdateInvalidRoute(map[string]int)                   {}
+func (NoMetric) Close()                                              {}
+
+var _ Metrics = NoMetric{}
+
 // Options for initializing metrics collection.
 type Options struct {
 	// the metrics exposing format.

queuelistener/listener.go

@@ -20,6 +20,7 @@ const (
 	maxCalculatedQueueSize          = 50_000
 	acceptedConnectionsKey          = "listener.accepted.connections"
 	queuedConnectionsKey            = "listener.queued.connections"
+	queueTimeoutKey                 = "listener.queued.timeouts"
 	acceptLatencyKey                = "listener.accept.latency"
 )
 
@@ -79,6 +80,8 @@ type Options struct {
 }
 
 type listener struct {
+	metrics           metrics.Metrics
+	log               logging.Logger
 	options           Options
 	maxConcurrency    int64
 	maxQueueSize      int64
@@ -96,6 +99,7 @@ type listener struct {
 var (
 	token             struct{}
 	errListenerClosed = errors.New("listener closed")
+	errAcceptTimeout  = errors.New("accept timeout")
 )
 
 func (c *connection) Close() error {
@@ -132,10 +136,7 @@ func (o Options) maxQueueSize() int64 {
 		return int64(o.MaxQueueSize)
 	}
 
-	maxQueueSize := 10 * o.maxConcurrency()
-	if maxQueueSize > maxCalculatedQueueSize {
-		maxQueueSize = maxCalculatedQueueSize
-	}
+	maxQueueSize := min(10*o.maxConcurrency(), maxCalculatedQueueSize)
 
 	return maxQueueSize
 }
@@ -155,6 +156,8 @@ func listenWith(nl net.Listener, o Options) (net.Listener, error) {
 
 	l := &listener{
 		options:           o,
+		log:               o.Log,
+		metrics:           o.Metrics,
 		maxConcurrency:    o.maxConcurrency(),
 		maxQueueSize:      o.maxQueueSize(),
 		externalListener:  nl,
@@ -165,7 +168,10 @@ func listenWith(nl net.Listener, o Options) (net.Listener, error) {
 		releaseConnection: make(chan struct{}),
 		quit:              make(chan struct{}),
 	}
-	o.Log.Infof("TCP lifo listener config: %s", l)
+	if l.metrics == nil {
+		l.metrics = metrics.NoMetric{}
+	}
+	l.log.Infof("TCP lifo listener config: %s", l)
 
 	go l.listenExternal()
 	go l.listenInternal()
@@ -233,7 +239,7 @@ func (l *listener) listenExternal() {
 			//lint:ignore SA1019 Temporary is deprecated in Go 1.18, but keep it for now (https://github.com/zalando/skipper/issues/1992)
 			if nerr, ok := err.(net.Error); ok && nerr.Temporary() {
 				delay = bounce(delay)
-				l.options.Log.Errorf(
+				l.log.Errorf(
 					"queue listener: accept error: %v, retrying in %v",
 					err,
 					delay,
@@ -309,10 +315,8 @@ func (l *listener) listenInternal() {
 			)
 		}
 
-		if l.options.Metrics != nil {
-			l.options.Metrics.UpdateGauge(acceptedConnectionsKey, float64(concurrency))
-			l.options.Metrics.UpdateGauge(queuedConnectionsKey, float64(queue.size))
-		}
+		l.metrics.UpdateGauge(acceptedConnectionsKey, float64(concurrency))
+		l.metrics.UpdateGauge(queuedConnectionsKey, float64(queue.size))
 
 		select {
 		case conn := <-l.acceptExternal:
@@ -362,7 +366,7 @@ func (l *listener) listenInternal() {
 			// Closing the real listener in a separate goroutine is based on inspecting the
 			// stdlib. It's fair to just log the errors.
 			if err := l.externalListener.Close(); err != nil {
-				l.options.Log.Errorf("Failed to close network listener: %v.", err)
+				l.log.Errorf("Failed to close network listener: %v.", err)
 			}
 
 			if l.closedHook != nil {
@@ -377,9 +381,7 @@ func (l *listener) listenInternal() {
 func (l *listener) Accept() (net.Conn, error) {
 	select {
 	case c := <-l.acceptInternal:
-		if l.options.Metrics != nil {
-			l.options.Metrics.MeasureSince(acceptLatencyKey, c.external.accepted)
-		}
+		l.metrics.MeasureSince(acceptLatencyKey, c.external.accepted)
 		return c, nil
 	case err := <-l.internalError:
 		return nil, err

queuelistener/stack.go

@@ -0,0 +1,44 @@
+package queuelistener
+
+import (
+	"sync"
+)
+
+const stackSize int = 10000
+
+type naiveStack[T any] struct {
+	mu    sync.Mutex
+	top   int
+	items [stackSize]*T
+}
+
+func NewStack() *naiveStack[external] {
+	return &naiveStack[external]{
+		top: -1,
+	}
+}
+
+func (s *naiveStack[T]) Push(data *T) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if s.top == len(s.items)-1 {
+		return
+	}
+
+	s.top++
+	s.items[s.top] = data
+}
+
+func (s *naiveStack[T]) Pop() *T {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if s.top == -1 {
+		return nil
+	} else {
+		defer func() { s.top-- }()
+	}
+
+	return s.items[s.top]
+}