From b1cb2804b9ed8a65d0e95be2127d9ac30a65c2fc Mon Sep 17 00:00:00 2001 From: Eduard Zateev Date: Thu, 22 Oct 2026 13:23:44 +0200 Subject: [PATCH] port/builtin: retry accept after fd exhaustion Signed-off-by: Eduard Zateev --- pkg/port/builtin/parent/tcp/tcp.go | 72 +++++++++-- pkg/port/builtin/parent/tcp/tcp_test.go | 157 ++++++++++++++++++++++++ 2 files changed, 218 insertions(+), 11 deletions(-) create mode 100644 pkg/port/builtin/parent/tcp/tcp_test.go diff --git a/pkg/port/builtin/parent/tcp/tcp.go b/pkg/port/builtin/parent/tcp/tcp.go index e3053ffd..ff5c87ae 100644 --- a/pkg/port/builtin/parent/tcp/tcp.go +++ b/pkg/port/builtin/parent/tcp/tcp.go @@ -1,12 +1,15 @@ package tcp import ( + "errors" "fmt" "io" "net" "os" "strconv" "sync" + "syscall" + "time" "github.com/rootless-containers/rootlesskit/v3/pkg/port" "github.com/rootless-containers/rootlesskit/v3/pkg/port/builtin/msg" @@ -19,17 +22,7 @@ func Run(socketPath string, spec port.Spec, stopCh <-chan struct{}, stoppedCh ch return err } newConns := make(chan net.Conn) - go func() { - for { - c, err := ln.Accept() - if err != nil { - fmt.Fprintf(logWriter, "accept: %v\n", err) - close(newConns) - return - } - newConns <- c - } - }() + go acceptConnections(ln, newConns, stopCh, logWriter) go func() { defer func() { stoppedCh <- ln.Close() @@ -56,6 +49,63 @@ func Run(socketPath string, spec port.Spec, stopCh <-chan struct{}, stoppedCh ch return nil } +const ( + minAcceptRetryDelay = 5 * time.Millisecond + maxAcceptRetryDelay = time.Second +) + +func acceptConnections(ln net.Listener, newConns chan<- net.Conn, stopCh <-chan struct{}, logWriter io.Writer) { + defer close(newConns) + + var retryDelay time.Duration + for { + c, err := ln.Accept() + if err != nil { + select { + case <-stopCh: + return + default: + } + + if errors.Is(err, syscall.EMFILE) || errors.Is(err, syscall.ENFILE) { + if retryDelay == 0 { + retryDelay = minAcceptRetryDelay + } else { + retryDelay *= 2 + if retryDelay > maxAcceptRetryDelay { + retryDelay = maxAcceptRetryDelay + } + } + fmt.Fprintf(logWriter, "accept: %v; retrying in %s\n", err, retryDelay) + timer := time.NewTimer(retryDelay) + select { + case <-timer.C: + case <-stopCh: + if !timer.Stop() { + select { + case <-timer.C: + default: + } + } + return + } + continue + } + + fmt.Fprintf(logWriter, "accept: %v\n", err) + return + } + retryDelay = 0 + + select { + case newConns <- c: + case <-stopCh: + c.Close() + return + } + } +} + func copyConnToChild(c net.Conn, socketPath string, spec port.Spec, stopCh <-chan struct{}) error { defer c.Close() // get fd from the child as an SCM_RIGHTS cmsg diff --git a/pkg/port/builtin/parent/tcp/tcp_test.go b/pkg/port/builtin/parent/tcp/tcp_test.go new file mode 100644 index 00000000..72e7b0e0 --- /dev/null +++ b/pkg/port/builtin/parent/tcp/tcp_test.go @@ -0,0 +1,157 @@ +package tcp + +import ( + "bytes" + "errors" + "net" + "sync" + "syscall" + "testing" + "time" +) + +type acceptResult struct { + conn net.Conn + err error +} + +type testListener struct { + results chan acceptResult + closed chan struct{} + once sync.Once +} + +func newTestListener() *testListener { + return &testListener{ + results: make(chan acceptResult), + closed: make(chan struct{}), + } +} + +func (l *testListener) Accept() (net.Conn, error) { + select { + case result := <-l.results: + return result.conn, result.err + case <-l.closed: + return nil, net.ErrClosed + } +} + +func (l *testListener) Close() error { + l.once.Do(func() { close(l.closed) }) + return nil +} + +func (l *testListener) Addr() net.Addr { + return &net.TCPAddr{} +} + +type signalWriter struct { + wrote chan struct{} +} + +func (w signalWriter) Write(p []byte) (int, error) { + select { + case w.wrote <- struct{}{}: + default: + } + return len(p), nil +} + +func TestAcceptConnectionsRetriesFileDescriptorExhaustion(t *testing.T) { + for _, acceptErr := range []error{syscall.EMFILE, syscall.ENFILE} { + t.Run(acceptErr.Error(), func(t *testing.T) { + listener := newTestListener() + stopCh := make(chan struct{}) + newConns := make(chan net.Conn) + done := make(chan struct{}) + var log bytes.Buffer + + go func() { + acceptConnections(listener, newConns, stopCh, &log) + close(done) + }() + + listener.results <- acceptResult{err: acceptErr} + serverConn, clientConn := net.Pipe() + defer serverConn.Close() + defer clientConn.Close() + listener.results <- acceptResult{conn: serverConn} + + select { + case conn := <-newConns: + if conn != serverConn { + t.Fatalf("received unexpected connection %v", conn) + } + case <-time.After(time.Second): + t.Fatal("accept loop did not retry after descriptor exhaustion") + } + + close(stopCh) + listener.Close() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("accept loop did not stop") + } + }) + } +} + +func TestAcceptConnectionsStopsDuringRetryDelay(t *testing.T) { + listener := newTestListener() + stopCh := make(chan struct{}) + newConns := make(chan net.Conn) + done := make(chan struct{}) + wrote := make(chan struct{}, 1) + + go func() { + acceptConnections(listener, newConns, stopCh, signalWriter{wrote: wrote}) + close(done) + }() + + listener.results <- acceptResult{err: syscall.EMFILE} + select { + case <-wrote: + case <-time.After(time.Second): + t.Fatal("accept loop did not report descriptor exhaustion") + } + + close(stopCh) + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("accept loop did not stop during retry delay") + } + if conn, ok := <-newConns; ok { + t.Fatalf("expected closed connection channel, got %v", conn) + } +} + +func TestAcceptConnectionsStopsOnPermanentError(t *testing.T) { + listener := newTestListener() + newConns := make(chan net.Conn) + var log bytes.Buffer + done := make(chan struct{}) + + go func() { + acceptConnections(listener, newConns, make(chan struct{}), &log) + close(done) + }() + + wantErr := errors.New("permanent accept error") + listener.results <- acceptResult{err: wantErr} + + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("accept loop did not stop after a permanent error") + } + + if got := <-newConns; got != nil { + t.Fatalf("expected closed connection channel, got %v", got) + } + if !bytes.Contains(log.Bytes(), []byte(wantErr.Error())) { + t.Fatalf("expected error in log, got %q", log.String()) + } +}