From 5a35585acd1cb1b3f6c213bcc3d0c722b9d5c573 Mon Sep 17 00:00:00 2001 From: Denis Peshkov Date: Tue, 12 Aug 2025 22:22:12 +0400 Subject: [PATCH] http/listener: fix bugs and simplify (#21514) * Store `ctx.Done` channel in a struct instead of a `ctx`. See: https://go.dev/blog/context-and-structs * Return from `handleListener` on `ctx` cancellation, preventing goroutine leaks * Simplify `handleListener` by removing the `send` closure. The `handleListener` is inlined by the compiler * Return the first error from `Close` * Preallocate slice in `Addrs` * Reduce duplication in handling `opts.Trace` * http/listener: revert error propagation from Close() * http/listener: preserve original listener address in Addr() * Preserve the original address when calling Addr() with multiple listeners * Remove unused listeners from the slice --- internal/http/listener.go | 80 ++++++++++++++++++--------------------- 1 file changed, 37 insertions(+), 43 deletions(-) diff --git a/internal/http/listener.go b/internal/http/listener.go index be3c1a1a8..8d6341468 100644 --- a/internal/http/listener.go +++ b/internal/http/listener.go @@ -21,6 +21,7 @@ import ( "context" "fmt" "net" + "slices" "syscall" "time" @@ -38,46 +39,39 @@ type httpListener struct { opts TCPOptions listeners []net.Listener // underlying TCP listeners. acceptCh chan acceptResult // channel where all TCP listeners write accepted connection. - ctx context.Context + ctxDoneCh <-chan struct{} ctxCanceler context.CancelFunc } // start - starts separate goroutine for each TCP listener. A valid new connection is passed to httpListener.acceptCh. func (listener *httpListener) start() { - // Closure to send acceptResult to acceptCh. - // It returns true if the result is sent else false if returns when doneCh is closed. - send := func(result acceptResult) bool { - select { - case listener.acceptCh <- result: - // Successfully written to acceptCh - return true - case <-listener.ctx.Done(): - return false - } - } - - // Closure to handle TCPListener until done channel is closed. - handleListener := func(idx int, listener net.Listener) { + // Closure to handle listener until httpListener.ctxDoneCh channel is closed. + handleListener := func(idx int, ln net.Listener) { for { - conn, err := listener.Accept() - send(acceptResult{conn, err, idx}) + conn, err := ln.Accept() + select { + case listener.acceptCh <- acceptResult{conn, err, idx}: + case <-listener.ctxDoneCh: + return + } } } - // Start separate goroutine for each TCP listener to handle connection. - for idx, tcpListener := range listener.listeners { - go handleListener(idx, tcpListener) + // Start separate goroutine for each listener to handle connection. + for idx, ln := range listener.listeners { + go handleListener(idx, ln) } } // Accept - reads from httpListener.acceptCh for one of previously accepted TCP connection and returns the same. func (listener *httpListener) Accept() (conn net.Conn, err error) { select { - case result, ok := <-listener.acceptCh: - if ok { - return deadlineconn.New(result.conn).WithReadDeadline(listener.opts.IdleTimeout).WithWriteDeadline(listener.opts.IdleTimeout), result.err + case result := <-listener.acceptCh: + if result.err != nil { + return nil, result.err } - case <-listener.ctx.Done(): + return deadlineconn.New(result.conn).WithReadDeadline(listener.opts.IdleTimeout).WithWriteDeadline(listener.opts.IdleTimeout), result.err + case <-listener.ctxDoneCh: } return nil, syscall.EINVAL } @@ -101,18 +95,18 @@ func (listener *httpListener) Addr() (addr net.Addr) { } if tcpAddr, ok := addr.(*net.TCPAddr); ok { - if ip := net.ParseIP("0.0.0.0"); ip != nil { - tcpAddr.IP = ip + return &net.TCPAddr{ + IP: net.IPv4zero, + Port: tcpAddr.Port, + Zone: tcpAddr.Zone, } - - addr = tcpAddr - return addr } panic("unknown address type on listener") } // Addrs - returns all address information of TCP listeners. func (listener *httpListener) Addrs() (addrs []net.Addr) { + addrs = make([]net.Addr, 0, len(listener.listeners)) for i := range listener.listeners { addrs = append(addrs, listener.listeners[i].Addr()) } @@ -154,6 +148,10 @@ func newHTTPListener(ctx context.Context, serverAddrs []string, opts TCPOptions) listeners := make([]net.Listener, 0, len(serverAddrs)) listenErrs = make([]error, len(serverAddrs)) + if opts.Trace == nil { + opts.Trace = func(msg string) {} // Noop if not defined. + } + // Unix listener with special TCP options. listenCfg := net.ListenConfig{ Control: setTCPParametersFn(opts), @@ -162,17 +160,12 @@ func newHTTPListener(ctx context.Context, serverAddrs []string, opts TCPOptions) for i, serverAddr := range serverAddrs { l, e := listenCfg.Listen(ctx, "tcp", serverAddr) if e != nil { - if opts.Trace != nil { - opts.Trace(fmt.Sprint("listenCfg.Listen: ", e)) - } + opts.Trace("listenCfg.Listen: " + e.Error()) listenErrs[i] = e continue } - - if opts.Trace != nil { - opts.Trace(fmt.Sprint("adding listener to ", l.Addr())) - } + opts.Trace("adding listener to " + l.Addr().String()) listeners = append(listeners, l) } @@ -181,16 +174,17 @@ func newHTTPListener(ctx context.Context, serverAddrs []string, opts TCPOptions) // No listeners initialized, no need to continue return } + listeners = slices.Clip(listeners) + ctx, cancel := context.WithCancel(ctx) listener = &httpListener{ - listeners: listeners, - acceptCh: make(chan acceptResult, len(listeners)), - opts: opts, - } - listener.ctx, listener.ctxCanceler = context.WithCancel(ctx) - if opts.Trace != nil { - opts.Trace(fmt.Sprint("opening ", len(listener.listeners), " listeners")) + listeners: listeners, + acceptCh: make(chan acceptResult, len(listeners)), + opts: opts, + ctxDoneCh: ctx.Done(), + ctxCanceler: cancel, } + opts.Trace(fmt.Sprintf("opening %d listeners", len(listener.listeners))) listener.start() return