Fast and extensible multi-platform HTTP/1-2-3 web server with automatic HTTPS
Mirrored from https://github.com/Slazee/caddy.git
Synced Nov 12, 22:00 UTC
core: Apply SO_REUSEPORT to UDP sockets (#5725)
* core: Apply SO_REUSEPORT to UDP sockets For some reason, 10 months ago when I implemented SO_REUSEPORT for TCP, I didn't realize, or forgot, that it can be used for UDP too. It is a much better solution than using deadline hacks to reuse a socket, at least for TCP. Then https://github.com/mholt/caddy-l4/issues/132 was posted, in which we see that UDP servers never actually stopped when the L4 app was stopped. I verified this using this command: $ nc -u 127.0.0.1 55353 combined with POSTing configs to the /load admin endpoint (which alternated between an echo server and a proxy server so I could tell which config was being used). I refactored the code to use SO_REUSEPORT for UDP, but of course we still need graceful reloads on all platforms, not just Unix, so I also implemented a deadline hack similar to what we used for TCP before. That implementation for TCP was not perfect, possibly having a logical (not data) race condition; but for UDP so far it seems to be working. Verified the same way I verified that SO_REUSEPORT works. I think this code is slightly cleaner and I'm fairly confident this code is effective. * Check error * Fix return * Fix var name * implement Unwrap interface and clean up * move unix packet conn to platform specific file * implement Unwrap for unix packet conn * Move sharedPacketConn into proper file * Fix Windows * move sharedPacketConn and fakeClosePacketConn to proper file --------- Co-authored-by: Weidi Deng <weidi_deng@icloud.com>
Matt Holt committed
Oct 16, 2023 at 22:17 UTC
174c19a9539443e6651eecf23ab86c7fd5d8c293
3 files changed
+181
-140
listen.go
+97
-9
@@ -30,18 +30,34 @@ func reuseUnixSocket(network, addr string) (any, error) {
30
return nil, nil
31
}
32
33
-func listenTCPOrUnix(ctx context.Context, lnKey string, network, address string, config net.ListenConfig) (net.Listener, error) {
34
- sharedLn, _, err := listenerPool.LoadOrNew(lnKey, func() (Destructor, error) {
35
- ln, err := config.Listen(ctx, network, address)
33
+func listenReusable(ctx context.Context, lnKey string, network, address string, config net.ListenConfig) (any, error) {
34
+ switch network {
35
+ case "udp", "udp4", "udp6", "unixgram":
36
+ sharedPc, _, err := listenerPool.LoadOrNew(lnKey, func() (Destructor, error) {
37
+ pc, err := config.ListenPacket(ctx, network, address)
38
+ if err != nil {
39
+ return nil, err
40
+ }
41
+ return &sharedPacketConn{PacketConn: pc, key: lnKey}, nil
42
+ })
43
if err != nil {
44
return nil, err
45
}
39
- return &sharedListener{Listener: ln, key: lnKey}, nil
40
- })
41
- if err != nil {
42
- return nil, err
46
+ return &fakeClosePacketConn{sharedPacketConn: sharedPc.(*sharedPacketConn)}, nil
47
+
48
+ default:
49
+ sharedLn, _, err := listenerPool.LoadOrNew(lnKey, func() (Destructor, error) {
50
+ ln, err := config.Listen(ctx, network, address)
51
+ if err != nil {
52
+ return nil, err
53
+ }
54
+ return &sharedListener{Listener: ln, key: lnKey}, nil
55
+ })
56
+ if err != nil {
57
+ return nil, err
58
+ }
59
+ return &fakeCloseListener{sharedListener: sharedLn.(*sharedListener), keepAlivePeriod: config.KeepAlive}, nil
60
}
44
- return &fakeCloseListener{sharedListener: sharedLn.(*sharedListener), keepAlivePeriod: config.KeepAlive}, nil
61
}
62
63
// fakeCloseListener is a private wrapper over a listener that
@@ -98,7 +114,7 @@ func (fcl *fakeCloseListener) Accept() (net.Conn, error) {
114
// so that it's clear in the code that side-effects are shared with other
115
// users of this listener, not just our own reference to it; we also don't
116
// do anything with the error because all we could do is log it, but we
101
- // expliclty assign it to nothing so we don't forget it's there if needed
117
+ // explicitly assign it to nothing so we don't forget it's there if needed
118
_ = fcl.sharedListener.clearDeadline()
119
120
if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
@@ -172,3 +188,75 @@ func (sl *sharedListener) setDeadline() error {
188
func (sl *sharedListener) Destruct() error {
189
return sl.Listener.Close()
190
}
191
+
192
+// fakeClosePacketConn is like fakeCloseListener, but for PacketConns,
193
+// or more specifically, *net.UDPConn
194
+type fakeClosePacketConn struct {
195
+ closed int32 // accessed atomically; belongs to this struct only
196
+ *sharedPacketConn // embedded, so we also become a net.PacketConn; its key is used in Close
197
+}
198
+
199
+func (fcpc *fakeClosePacketConn) ReadFrom(p []byte) (n int, addr net.Addr, err error) {
200
+ // if the listener is already "closed", return error
201
+ if atomic.LoadInt32(&fcpc.closed) == 1 {
202
+ return 0, nil, &net.OpError{
203
+ Op: "readfrom",
204
+ Net: fcpc.LocalAddr().Network(),
205
+ Addr: fcpc.LocalAddr(),
206
+ Err: errFakeClosed,
207
+ }
208
+ }
209
+
210
+ // call underlying readfrom
211
+ n, addr, err = fcpc.sharedPacketConn.ReadFrom(p)
212
+ if err != nil {
213
+ // this server was stopped, so clear the deadline and let
214
+ // any new server continue reading; but we will exit
215
+ if atomic.LoadInt32(&fcpc.closed) == 1 {
216
+ if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
217
+ if err = fcpc.SetReadDeadline(time.Time{}); err != nil {
218
+ return
219
+ }
220
+ }
221
+ }
222
+ return
223
+ }
224
+
225
+ return
226
+}
227
+
228
+// Close won't close the underlying socket unless there is no more reference, then listenerPool will close it.
229
+func (fcpc *fakeClosePacketConn) Close() error {
230
+ if atomic.CompareAndSwapInt32(&fcpc.closed, 0, 1) {
231
+ _ = fcpc.SetReadDeadline(time.Now()) // unblock ReadFrom() calls to kick old servers out of their loops
232
+ _, _ = listenerPool.Delete(fcpc.sharedPacketConn.key)
233
+ }
234
+ return nil
235
+}
236
+
237
+func (fcpc *fakeClosePacketConn) Unwrap() net.PacketConn {
238
+ return fcpc.sharedPacketConn.PacketConn
239
+}
240
+
241
+// sharedPacketConn is like sharedListener, but for net.PacketConns.
242
+type sharedPacketConn struct {
243
+ net.PacketConn
244
+ key string
245
+}
246
+
247
+// Destruct closes the underlying socket.
248
+func (spc *sharedPacketConn) Destruct() error {
249
+ return spc.PacketConn.Close()
250
+}
251
+
252
+// Unwrap returns the underlying socket
253
+func (spc *sharedPacketConn) Unwrap() net.PacketConn {
254
+ return spc.PacketConn
255
+}
256
+
257
+// Interface guards (see https://github.com/caddyserver/caddy/issues/3998)
258
+var (
259
+ _ (interface {
260
+ Unwrap() net.PacketConn
261
+ }) = (*fakeClosePacketConn)(nil)
262
+)
listen_unix.go
+72
-3
@@ -22,8 +22,10 @@ package caddy
22
import (
23
"context"
24
"errors"
25
+ "io"
26
"io/fs"
27
"net"
28
+ "os"
29
"sync/atomic"
30
"syscall"
31
@@ -87,7 +89,7 @@ func reuseUnixSocket(network, addr string) (any, error) {
89
return nil, nil
90
}
91
90
-func listenTCPOrUnix(ctx context.Context, lnKey string, network, address string, config net.ListenConfig) (net.Listener, error) {
92
+func listenReusable(ctx context.Context, lnKey string, network, address string, config net.ListenConfig) (any, error) {
93
// wrap any Control function set by the user so we can also add our reusePort control without clobbering theirs
94
oldControl := config.Control
95
config.Control = func(network, address string, c syscall.RawConn) error {
@@ -103,7 +105,14 @@ func listenTCPOrUnix(ctx context.Context, lnKey string, network, address string,
105
// we still put it in the listenerPool so we can count how many
106
// configs are using this socket; necessary to ensure we can know
107
// whether to enforce shutdown delays, for example (see #5393).
106
- ln, err := config.Listen(ctx, network, address)
108
+ var ln io.Closer
109
+ var err error
110
+ switch network {
111
+ case "udp", "udp4", "udp6", "unixgram":
112
+ ln, err = config.ListenPacket(ctx, network, address)
113
+ default:
114
+ ln, err = config.Listen(ctx, network, address)
115
+ }
116
if err == nil {
117
listenerPool.LoadOrStore(lnKey, nil)
118
}
@@ -117,9 +126,23 @@ func listenTCPOrUnix(ctx context.Context, lnKey string, network, address string,
126
unixSockets[lnKey] = ln.(*unixListener)
127
}
128
129
+ // TODO: Not 100% sure this is necessary, but we do this for net.UnixListener in listen_unix.go, so...
130
+ if unix, ok := ln.(*net.UnixConn); ok {
131
+ ln = &unixConn{unix, address, lnKey, &one}
132
+ unixSockets[lnKey] = ln.(*unixConn)
133
+ }
134
+
135
// lightly wrap the listener so that when it is closed,
136
// we can decrement the usage pool counter
122
- return deleteListener{ln, lnKey}, err
137
+ switch specificLn := ln.(type) {
138
+ case net.Listener:
139
+ return deleteListener{specificLn, lnKey}, err
140
+ case net.PacketConn:
141
+ return deletePacketConn{specificLn, lnKey}, err
142
+ }
143
+
144
+ // other types, I guess we just return them directly
145
+ return ln, err
146
}
147
148
// reusePort sets SO_REUSEPORT. Ineffective for unix sockets.
@@ -158,6 +181,36 @@ func (uln *unixListener) Close() error {
181
return uln.UnixListener.Close()
182
}
183
184
+type unixConn struct {
185
+ *net.UnixConn
186
+ filename string
187
+ mapKey string
188
+ count *int32 // accessed atomically
189
+}
190
+
191
+func (uc *unixConn) Close() error {
192
+ newCount := atomic.AddInt32(uc.count, -1)
193
+ if newCount == 0 {
194
+ defer func() {
195
+ unixSocketsMu.Lock()
196
+ delete(unixSockets, uc.mapKey)
197
+ unixSocketsMu.Unlock()
198
+ _ = syscall.Unlink(uc.filename)
199
+ }()
200
+ }
201
+ return uc.UnixConn.Close()
202
+}
203
+
204
+func (uc *unixConn) Unwrap() net.PacketConn {
205
+ return uc.UnixConn
206
+}
207
+
208
+// unixSockets keeps track of the currently-active unix sockets
209
+// so we can transfer their FDs gracefully during reloads.
210
+var unixSockets = make(map[string]interface {
211
+ File() (*os.File, error)
212
+})
213
+
214
// deleteListener is a type that simply deletes itself
215
// from the listenerPool when it closes. It is used
216
// solely for the purpose of reference counting (i.e.
@@ -171,3 +224,19 @@ func (dl deleteListener) Close() error {
224
_, _ = listenerPool.Delete(dl.lnKey)
225
return dl.Listener.Close()
226
}
227
+
228
+// deletePacketConn is like deleteListener, but
229
+// for net.PacketConns.
230
+type deletePacketConn struct {
231
+ net.PacketConn
232
+ lnKey string
233
+}
234
+
235
+func (dl deletePacketConn) Close() error {
236
+ _, _ = listenerPool.Delete(dl.lnKey)
237
+ return dl.PacketConn.Close()
238
+}
239
+
240
+func (dl deletePacketConn) Unwrap() net.PacketConn {
241
+ return dl.PacketConn
242
+}
listeners.go
+12
-128
@@ -28,7 +28,6 @@ import (
28
"strings"
29
"sync"
30
"sync/atomic"
31
- "syscall"
31
"time"
32
33
"github.com/quic-go/quic-go"
@@ -149,11 +148,13 @@ func (na NetworkAddress) Listen(ctx context.Context, portOffset uint, config net
148
}
149
150
func (na NetworkAddress) listen(ctx context.Context, portOffset uint, config net.ListenConfig) (any, error) {
152
- var ln any
153
- var err error
154
- var address string
155
- var unixFileMode fs.FileMode
156
- var isAbtractUnixSocket bool
151
+ var (
152
+ ln any
153
+ err error
154
+ address string
155
+ unixFileMode fs.FileMode
156
+ isAbtractUnixSocket bool
157
+ )
158
159
// split unix socket addr early so lnKey
160
// is independent of permissions bits
@@ -181,27 +182,10 @@ func (na NetworkAddress) listen(ctx context.Context, portOffset uint, config net
182
183
lnKey := listenerKey(na.Network, address)
184
184
- switch na.Network {
185
- case "tcp", "tcp4", "tcp6", "unix", "unixpacket":
186
- ln, err = listenTCPOrUnix(ctx, lnKey, na.Network, address, config)
187
- case "unixgram":
188
- ln, err = config.ListenPacket(ctx, na.Network, address)
189
- case "udp", "udp4", "udp6":
190
- sharedPc, _, err := listenerPool.LoadOrNew(lnKey, func() (Destructor, error) {
191
- pc, err := config.ListenPacket(ctx, na.Network, address)
192
- if err != nil {
193
- return nil, err
194
- }
195
- return &sharedPacketConn{PacketConn: pc, key: lnKey}, nil
196
- })
197
- if err != nil {
198
- return nil, err
199
- }
200
- spc := sharedPc.(*sharedPacketConn)
201
- ln = &fakeClosePacketConn{spc: spc, UDPConn: spc.PacketConn.(*net.UDPConn)}
202
- }
185
if strings.HasPrefix(na.Network, "ip") {
186
ln, err = config.ListenPacket(ctx, na.Network, address)
187
+ } else {
188
+ ln, err = listenReusable(ctx, lnKey, na.Network, address, config)
189
}
190
if err != nil {
191
return nil, err
@@ -210,13 +194,6 @@ func (na NetworkAddress) listen(ctx context.Context, portOffset uint, config net
194
return nil, fmt.Errorf("unsupported network type: %s", na.Network)
195
}
196
213
- // TODO: Not 100% sure this is necessary, but we do this for net.UnixListener in listen_unix.go, so...
214
- if unix, ok := ln.(*net.UnixConn); ok {
215
- one := int32(1)
216
- ln = &unixConn{unix, address, lnKey, &one}
217
- unixSockets[lnKey] = unix
218
- }
219
-
197
if IsUnixNetwork(na.Network) {
198
if !isAbtractUnixSocket {
199
if err := os.Chmod(address, unixFileMode); err != nil {
@@ -555,20 +532,8 @@ func ListenQUIC(ln net.PacketConn, tlsConf *tls.Config, activeRequests *int64) (
532
// and the request counter will reflect current http server
533
ctx, cancel := sql.sqs.addState(tlsConf, activeRequests)
534
558
- // TODO: to serve QUIC over a unix socket, currently we need to hold onto
559
- // the underlying net.PacketConn (which we wrap as unixConn to keep count
560
- // of closes) because closing the quic.EarlyListener doesn't actually close
561
- // the underlying PacketConn, but we need to for unix sockets since we dup
562
- // the file descriptor and thus need to close the original; track issue:
563
- // https://github.com/quic-go/quic-go/issues/3560#issuecomment-1258959608
564
- var unix *unixConn
565
- if uc, ok := ln.(*unixConn); ok {
566
- unix = uc
567
- }
568
-
535
return &fakeCloseQuicListener{
536
sharedQuicListener: sql,
571
- uc: unix,
537
context: ctx,
538
contextCancel: cancel,
539
}, nil
@@ -677,17 +642,6 @@ func (sql *sharedQuicListener) Destruct() error {
642
return sql.packetConn.Close()
643
}
644
680
-// sharedPacketConn is like sharedListener, but for net.PacketConns.
681
-type sharedPacketConn struct {
682
- net.PacketConn
683
- key string
684
-}
685
-
686
-// Destruct closes the underlying socket.
687
-func (spc *sharedPacketConn) Destruct() error {
688
- return spc.PacketConn.Close()
689
-}
690
-
645
// fakeClosedErr returns an error value that is not temporary
646
// nor a timeout, suitable for making the caller think the
647
// listener is actually closed
@@ -707,39 +661,9 @@ func fakeClosedErr(l interface{ Addr() net.Addr }) error {
661
// socket is actually left open.
662
var errFakeClosed = fmt.Errorf("listener 'closed' 😉")
663
710
-// fakeClosePacketConn is like fakeCloseListener, but for PacketConns,
711
-// or more specifically, *net.UDPConn
712
-type fakeClosePacketConn struct {
713
- closed int32 // accessed atomically; belongs to this struct only
714
- spc *sharedPacketConn // its key is used in Close
715
- *net.UDPConn // embedded, so we also become a net.PacketConn and enable several other optimizations done by quic-go
716
-}
717
-
718
-// interface guard for extra optimizations
719
-// needed by QUIC implementation: https://github.com/caddyserver/caddy/issues/3998, https://github.com/caddyserver/caddy/issues/5605
720
-var _ quic.OOBCapablePacketConn = (*fakeClosePacketConn)(nil)
721
-
722
-// https://pkg.go.dev/golang.org/x/net/ipv4#NewPacketConn is used by quic-go and requires a net.PacketConn type assertable to a net.Conn,
723
-// but doesn't actually use these methods, the only methods needed are `ReadMsgUDP` and `SyscallConn`.
724
-var _ net.Conn = (*fakeClosePacketConn)(nil)
725
-
726
-// Unwrap returns the underlying net.UDPConn for quic-go optimization
727
-func (fcpc *fakeClosePacketConn) Unwrap() any {
728
- return fcpc.UDPConn
729
-}
730
-
731
-// Close won't close the underlying socket unless there is no more reference, then listenerPool will close it.
732
-func (fcpc *fakeClosePacketConn) Close() error {
733
- if atomic.CompareAndSwapInt32(&fcpc.closed, 0, 1) {
734
- _, _ = listenerPool.Delete(fcpc.spc.key)
735
- }
736
- return nil
737
-}
738
-
664
type fakeCloseQuicListener struct {
740
- closed int32 // accessed atomically; belongs to this struct only
741
- *sharedQuicListener // embedded, so we also become a quic.EarlyListener
742
- uc *unixConn // underlying unix socket, if UDS
665
+ closed int32 // accessed atomically; belongs to this struct only
666
+ *sharedQuicListener // embedded, so we also become a quic.EarlyListener
667
context context.Context
668
contextCancel context.CancelFunc
669
}
@@ -766,11 +690,6 @@ func (fcql *fakeCloseQuicListener) Close() error {
690
if atomic.CompareAndSwapInt32(&fcql.closed, 0, 1) {
691
fcql.contextCancel()
692
_, _ = listenerPool.Delete(fcql.sharedQuicListener.key)
769
- if fcql.uc != nil {
770
- // unix sockets need to be closed ourselves because we dup() the file
771
- // descriptor when we reuse them, so this avoids a resource leak
772
- fcql.uc.Close()
773
- }
693
}
694
return nil
695
}
@@ -796,34 +715,7 @@ func RegisterNetwork(network string, getListener ListenerFunc) {
715
networkTypes[network] = getListener
716
}
717
799
-type unixConn struct {
800
- *net.UnixConn
801
- filename string
802
- mapKey string
803
- count *int32 // accessed atomically
804
-}
805
-
806
-func (uc *unixConn) Close() error {
807
- newCount := atomic.AddInt32(uc.count, -1)
808
- if newCount == 0 {
809
- defer func() {
810
- unixSocketsMu.Lock()
811
- delete(unixSockets, uc.mapKey)
812
- unixSocketsMu.Unlock()
813
- _ = syscall.Unlink(uc.filename)
814
- }()
815
- }
816
- return uc.UnixConn.Close()
817
-}
818
-
819
-// unixSockets keeps track of the currently-active unix sockets
820
-// so we can transfer their FDs gracefully during reloads.
821
-var (
822
- unixSockets = make(map[string]interface {
823
- File() (*os.File, error)
824
- })
825
- unixSocketsMu sync.Mutex
826
-)
718
+var unixSocketsMu sync.Mutex
719
720
// getListenerFromPlugin returns a listener on the given network and address
721
// if a plugin has registered the network name. It may return (nil, nil) if
@@ -867,11 +759,3 @@ type ListenerWrapper interface {
759
var listenerPool = NewUsagePool()
760
761
const maxPortSpan = 65535
870
-
871
-// Interface guards (see https://github.com/caddyserver/caddy/issues/3998)
872
-var (
873
- _ (interface{ SetReadBuffer(int) error }) = (*fakeClosePacketConn)(nil)
874
- _ (interface {
875
- SyscallConn() (syscall.RawConn, error)
876
- }) = (*fakeClosePacketConn)(nil)
877
-)