mirror of
https://github.com/minio/minio.git
synced 2026-02-04 18:00:15 -05:00
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
This commit is contained in:
@@ -21,6 +21,7 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
"net"
|
"net"
|
||||||
|
"slices"
|
||||||
"syscall"
|
"syscall"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -38,46 +39,39 @@ type httpListener struct {
|
|||||||
opts TCPOptions
|
opts TCPOptions
|
||||||
listeners []net.Listener // underlying TCP listeners.
|
listeners []net.Listener // underlying TCP listeners.
|
||||||
acceptCh chan acceptResult // channel where all TCP listeners write accepted connection.
|
acceptCh chan acceptResult // channel where all TCP listeners write accepted connection.
|
||||||
ctx context.Context
|
ctxDoneCh <-chan struct{}
|
||||||
ctxCanceler context.CancelFunc
|
ctxCanceler context.CancelFunc
|
||||||
}
|
}
|
||||||
|
|
||||||
// start - starts separate goroutine for each TCP listener. A valid new connection is passed to httpListener.acceptCh.
|
// start - starts separate goroutine for each TCP listener. A valid new connection is passed to httpListener.acceptCh.
|
||||||
func (listener *httpListener) start() {
|
func (listener *httpListener) start() {
|
||||||
// Closure to send acceptResult to acceptCh.
|
// Closure to handle listener until httpListener.ctxDoneCh channel is closed.
|
||||||
// It returns true if the result is sent else false if returns when doneCh is closed.
|
handleListener := func(idx int, ln net.Listener) {
|
||||||
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) {
|
|
||||||
for {
|
for {
|
||||||
conn, err := listener.Accept()
|
conn, err := ln.Accept()
|
||||||
send(acceptResult{conn, err, idx})
|
select {
|
||||||
|
case listener.acceptCh <- acceptResult{conn, err, idx}:
|
||||||
|
case <-listener.ctxDoneCh:
|
||||||
|
return
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Start separate goroutine for each TCP listener to handle connection.
|
// Start separate goroutine for each listener to handle connection.
|
||||||
for idx, tcpListener := range listener.listeners {
|
for idx, ln := range listener.listeners {
|
||||||
go handleListener(idx, tcpListener)
|
go handleListener(idx, ln)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Accept - reads from httpListener.acceptCh for one of previously accepted TCP connection and returns the same.
|
// 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) {
|
func (listener *httpListener) Accept() (conn net.Conn, err error) {
|
||||||
select {
|
select {
|
||||||
case result, ok := <-listener.acceptCh:
|
case result := <-listener.acceptCh:
|
||||||
if ok {
|
if result.err != nil {
|
||||||
return deadlineconn.New(result.conn).WithReadDeadline(listener.opts.IdleTimeout).WithWriteDeadline(listener.opts.IdleTimeout), result.err
|
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
|
return nil, syscall.EINVAL
|
||||||
}
|
}
|
||||||
@@ -101,18 +95,18 @@ func (listener *httpListener) Addr() (addr net.Addr) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
if tcpAddr, ok := addr.(*net.TCPAddr); ok {
|
if tcpAddr, ok := addr.(*net.TCPAddr); ok {
|
||||||
if ip := net.ParseIP("0.0.0.0"); ip != nil {
|
return &net.TCPAddr{
|
||||||
tcpAddr.IP = ip
|
IP: net.IPv4zero,
|
||||||
|
Port: tcpAddr.Port,
|
||||||
|
Zone: tcpAddr.Zone,
|
||||||
}
|
}
|
||||||
|
|
||||||
addr = tcpAddr
|
|
||||||
return addr
|
|
||||||
}
|
}
|
||||||
panic("unknown address type on listener")
|
panic("unknown address type on listener")
|
||||||
}
|
}
|
||||||
|
|
||||||
// Addrs - returns all address information of TCP listeners.
|
// Addrs - returns all address information of TCP listeners.
|
||||||
func (listener *httpListener) Addrs() (addrs []net.Addr) {
|
func (listener *httpListener) Addrs() (addrs []net.Addr) {
|
||||||
|
addrs = make([]net.Addr, 0, len(listener.listeners))
|
||||||
for i := range listener.listeners {
|
for i := range listener.listeners {
|
||||||
addrs = append(addrs, listener.listeners[i].Addr())
|
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))
|
listeners := make([]net.Listener, 0, len(serverAddrs))
|
||||||
listenErrs = make([]error, 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.
|
// Unix listener with special TCP options.
|
||||||
listenCfg := net.ListenConfig{
|
listenCfg := net.ListenConfig{
|
||||||
Control: setTCPParametersFn(opts),
|
Control: setTCPParametersFn(opts),
|
||||||
@@ -162,17 +160,12 @@ func newHTTPListener(ctx context.Context, serverAddrs []string, opts TCPOptions)
|
|||||||
for i, serverAddr := range serverAddrs {
|
for i, serverAddr := range serverAddrs {
|
||||||
l, e := listenCfg.Listen(ctx, "tcp", serverAddr)
|
l, e := listenCfg.Listen(ctx, "tcp", serverAddr)
|
||||||
if e != nil {
|
if e != nil {
|
||||||
if opts.Trace != nil {
|
opts.Trace("listenCfg.Listen: " + e.Error())
|
||||||
opts.Trace(fmt.Sprint("listenCfg.Listen: ", e))
|
|
||||||
}
|
|
||||||
|
|
||||||
listenErrs[i] = e
|
listenErrs[i] = e
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
opts.Trace("adding listener to " + l.Addr().String())
|
||||||
if opts.Trace != nil {
|
|
||||||
opts.Trace(fmt.Sprint("adding listener to ", l.Addr()))
|
|
||||||
}
|
|
||||||
|
|
||||||
listeners = append(listeners, l)
|
listeners = append(listeners, l)
|
||||||
}
|
}
|
||||||
@@ -181,16 +174,17 @@ func newHTTPListener(ctx context.Context, serverAddrs []string, opts TCPOptions)
|
|||||||
// No listeners initialized, no need to continue
|
// No listeners initialized, no need to continue
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
listeners = slices.Clip(listeners)
|
||||||
|
|
||||||
|
ctx, cancel := context.WithCancel(ctx)
|
||||||
listener = &httpListener{
|
listener = &httpListener{
|
||||||
listeners: listeners,
|
listeners: listeners,
|
||||||
acceptCh: make(chan acceptResult, len(listeners)),
|
acceptCh: make(chan acceptResult, len(listeners)),
|
||||||
opts: opts,
|
opts: opts,
|
||||||
|
ctxDoneCh: ctx.Done(),
|
||||||
|
ctxCanceler: cancel,
|
||||||
}
|
}
|
||||||
listener.ctx, listener.ctxCanceler = context.WithCancel(ctx)
|
opts.Trace(fmt.Sprintf("opening %d listeners", len(listener.listeners)))
|
||||||
if opts.Trace != nil {
|
|
||||||
opts.Trace(fmt.Sprint("opening ", len(listener.listeners), " listeners"))
|
|
||||||
}
|
|
||||||
listener.start()
|
listener.start()
|
||||||
|
|
||||||
return
|
return
|
||||||
|
|||||||
Reference in New Issue
Block a user