Repository navigation
Expand file tree
/
Copy pathudp.go
More file actions
2385 lines (2263 loc) · 84.6 KB
/
Copy pathudp.go
File metadata and controls
2385 lines (2263 loc) · 84.6 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
package mipstack
import (
"encoding/binary"
"errors"
"net"
"net/netip"
"os"
"sync"
"sync/atomic"
"syscall"
"time"
)
const (
// udpHeaderSize is the fixed UDP header length.
udpHeaderSize = 8
// udpDefaultReceiveCapacity bounds retained payload and queue metadata per
// socket unless the application selects another value.
udpDefaultReceiveCapacity = 4 * 1024 * 1024
// udpDatagramMetadataSize conservatively accounts for the slice and source
// address retained alongside each payload. In particular, empty datagrams
// must consume capacity so they cannot grow the queue without bound.
udpDatagramMetadataSize = 96
)
// UDPDatagram is the semantic representation of one UDP datagram. Source and
// Destination provide both wire ports and the IP pseudo-header addresses.
// IPPacket.UDPDatagram borrows Payload from the packet; callers must replace
// or copy it before modifying unowned input. Construction normalizes
// IPv4-mapped IPv6 addresses to IPv4.
type UDPDatagram struct {
// Source is the source IP address and UDP port.
Source netip.AddrPort
// Destination is the destination IP address and UDP port.
Destination netip.AddrPort
// ChecksumDisabled requests the optional zero IPv4 UDP checksum. It is
// invalid for IPv6.
ChecksumDisabled bool
// Payload is the datagram body.
Payload []byte
}
// udpWireLength validates UDP's fixed header and declared datagram length. It
// is intentionally small enough to inline in both the public codec and Stack's
// packet hot path; checksum policy remains with those callers.
func udpWireLength(udp []byte) (int, bool) {
if len(udp) < udpHeaderSize {
return 0, false
}
length := int(binary.BigEndian.Uint16(udp[4:6]))
if length < udpHeaderSize || length > len(udp) {
return 0, false
}
return length, true
}
// UDPDatagram validates and decodes the packet's UDP upper layer. A shorter
// UDP Length trims IP-layer padding; a length larger than the upper layer is
// invalid.
func (p IPPacket) UDPDatagram() (UDPDatagram, error) {
udp, pseudoHeaderSafe, err := p.upperLayerForProtocol(ProtocolUDP)
if err != nil {
return UDPDatagram{}, err
}
source, destination := p.Source.Unmap(), p.Destination.Unmap()
length, valid := udpWireLength(udp)
if !valid {
return UDPDatagram{}, syscall.EINVAL
}
checksumValue := binary.BigEndian.Uint16(udp[6:8])
if source.Is6() && checksumValue == 0 {
return UDPDatagram{}, syscall.EINVAL
}
if checksumValue != 0 {
if !pseudoHeaderSafe {
return UDPDatagram{}, syscall.EPROTONOSUPPORT
}
if transportChecksum(source, destination, ProtocolUDP, udp[:length]) != 0 {
return UDPDatagram{}, syscall.EINVAL
}
}
return UDPDatagram{
Source: netip.AddrPortFrom(source, binary.BigEndian.Uint16(udp[0:2])),
Destination: netip.AddrPortFrom(destination, binary.BigEndian.Uint16(udp[2:4])),
ChecksumDisabled: checksumValue == 0,
Payload: udp[udpHeaderSize:length],
}, nil
}
// MarshalBinary returns the complete UDP datagram wire encoding. Source and
// Destination contribute to its pseudo-header checksum but only their ports
// are encoded. MarshalBinary is semantically identical to AppendBinary(nil).
func (d UDPDatagram) MarshalBinary() ([]byte, error) { return d.AppendBinary(nil) }
// AppendBinary appends the complete UDP datagram wire encoding to dst. Source
// and Destination contribute to its pseudo-header checksum but only their
// ports are encoded. It validates every field before changing dst, does not
// retain any input slice, and permits the destination to share backing storage
// with Payload. On validation failure it returns the original dst unchanged.
func (d UDPDatagram) AppendBinary(dst []byte) ([]byte, error) {
normalized, totalSize, err := d.wireLayout()
if err != nil {
return dst, err
}
start := len(dst)
dst = extendForAppend(dst, totalSize)
marshalPublicUDPDatagram(dst[start:], normalized)
return dst, nil
}
// wireLayout validates d, normalizes its addresses, and returns its exact
// datagram length.
func (d UDPDatagram) wireLayout() (UDPDatagram, int, error) {
if d.Source.Addr().Zone() != "" || d.Destination.Addr().Zone() != "" {
return UDPDatagram{}, 0, syscall.EINVAL
}
source, destination := d.Source.Addr().Unmap(), d.Destination.Addr().Unmap()
if !d.Source.IsValid() || !d.Destination.IsValid() || source.Is4() != destination.Is4() || source.Is6() && d.ChecksumDisabled {
return UDPDatagram{}, 0, syscall.EINVAL
}
d.Source, d.Destination = netip.AddrPortFrom(source, d.Source.Port()), netip.AddrPortFrom(destination, d.Destination.Port())
if len(d.Payload) > 65535-udpHeaderSize {
return UDPDatagram{}, 0, syscall.EMSGSIZE
}
return d, udpHeaderSize + len(d.Payload), nil
}
// marshalPublicUDPDatagram writes one already validated semantic datagram.
func marshalPublicUDPDatagram(dst []byte, d UDPDatagram) {
copy(dst[udpHeaderSize:], d.Payload)
marshalUDPHeaderFields(dst[:udpHeaderSize], d.Source.Port(), d.Destination.Port(), len(dst))
if d.ChecksumDisabled {
return
}
writeUDPChecksumValue(dst[:udpHeaderSize], transportChecksumParts(
d.Source.Addr(), d.Destination.Addr(), ProtocolUDP, len(dst), dst[:udpHeaderSize], dst[udpHeaderSize:],
))
}
// udpDatagram is one validated inbound payload and its source endpoint.
type udpDatagram struct {
payload []byte
source netip.AddrPort
target netip.Addr
options ipPacketOptions
}
// UDPConnInfo is a point-in-time diagnostic snapshot of one UDP socket. Traffic
// byte counters measure UDP payload; receive-queue byte values also include
// the stack's per-datagram accounting overhead.
type UDPConnInfo struct {
// LocalAddress is the bound local endpoint; an unspecified address denotes
// a wildcard binding.
LocalAddress netip.AddrPort
// RemoteAddress is the connected peer, or an invalid endpoint for an
// unconnected socket.
RemoteAddress netip.AddrPort
// Closed reports whether the socket was closed when the snapshot was taken.
Closed bool
// ReceiveQueuePackets is the number of complete datagrams awaiting a read.
ReceiveQueuePackets int
// ReceiveQueueBytes is the accounted payload and metadata retained by the
// receive queue.
ReceiveQueueBytes int
// ReceiveQueueCapacity is the configured accounting-byte limit of the
// combined datagram and extended-error queues, not an exact heap limit.
ReceiveQueueCapacity int
// ReceiveErrors reports whether asynchronous network errors are retained
// for ReadError. Otherwise, unconnected sockets do not report those errors;
// connected sockets report hard errors on ordinary reads or writes. It also
// reports whether immediate failure to admit unicast or external-link
// non-unicast output is reported as ENOBUFS.
ReceiveErrors bool
// ErrorQueueEntries is the number of extended errors awaiting ReadError.
ErrorQueueEntries int
// ErrorQueueBytes is the accounted metadata and quoted packet data retained
// by the asynchronous error queue.
ErrorQueueBytes int
// ErrorsDropped counts extended-error entries discarded because the
// configured receive-buffer budget was exhausted.
ErrorsDropped uint64
// PacketsSent counts successful UDP socket write results. It includes writes
// silently lost during bounded output admission under the default
// ReceiveErrors policy and remains cumulative if bounded link scheduling
// later drops a packet.
PacketsSent uint64
// BytesSent counts payload bytes represented by those successful writes.
BytesSent uint64
// PacketsReceived counts datagrams accepted into the receive queue.
PacketsReceived uint64
// BytesReceived counts UDP payload bytes accepted into the receive queue.
BytesReceived uint64
// PacketsDropped counts datagrams rejected because the socket was closed or
// its receive queue lacked capacity.
PacketsDropped uint64
// ICMPErrors counts matching asynchronous ICMP errors delivered to the
// socket.
ICMPErrors uint64
// PathMTU is the complete-IP-packet PMTU for a connected unicast peer, or
// zero when no such path exists.
PathMTU int
// PathMTUDiscovery is the Linux-compatible source-fragmentation and PMTU
// policy used by subsequent writes.
PathMTUDiscovery PathMTUDiscovery
// HopLimit is the default unicast IPv4 TTL or IPv6 Hop Limit.
HopLimit int
// MulticastHopLimit is the default multicast IPv4 TTL or IPv6 Hop Limit.
MulticastHopLimit int
// MulticastLoopback reports whether transmitted multicast is delivered to
// matching local memberships.
MulticastLoopback bool
// Broadcast reports whether IPv4 broadcast output is permitted.
Broadcast bool
// TrafficClass is the default IPv4 TOS or IPv6 Traffic Class byte.
TrafficClass uint8
// FlowLabel is the effective IPv6 Flow Label; it is zero for IPv4 sockets.
FlowLabel uint32
// LastError is the most recently recorded socket operation or asynchronous
// network error.
LastError error
}
// udpReuseEndpoints is the optional reusable-binding dispatcher retained by
// Stack. The concrete registry is linked only when a public listen operation
// requests address or port reuse.
type udpReuseEndpoints interface {
// empty reports whether the registry contains no sockets.
empty() bool
// connections returns a snapshot of all registered sockets.
connections() []*UDPConn
// contains reports whether the registry owns a socket.
contains(connection *UDPConn) bool
// overlaps reports whether a binding conflicts with any registry entry.
overlaps(address netip.Addr, port uint16, dual bool) bool
// connection selects a socket for one local and remote endpoint pair.
connection(binding, local, remote netip.AddrPort) *UDPConn
// add registers a socket in its reusable binding group.
add(connection *UDPConn)
// remove unregisters a socket and reports whether it was present.
remove(connection *UDPConn) bool
}
// udpSocketBinding supplies the registration policy shared by ordinary and
// reusable ListenConfig sockets.
type udpSocketBinding interface {
// available reports whether the requested socket binding can be registered.
available(stack *Stack, address netip.Addr, port uint16, dual bool) bool
// register publishes one validated socket binding.
register(stack *Stack, connection *UDPConn) error
}
// udpNetwork retains the caller's validated network spelling without storing
// a string header on every socket.
type udpNetwork byte
const (
// udpNetworkGeneric preserves the dual-stack "udp" spelling.
udpNetworkGeneric udpNetwork = iota
// udpNetworkIPv4 preserves the IPv4-only "udp4" spelling.
udpNetworkIPv4
// udpNetworkIPv6 preserves the IPv6-only "udp6" spelling.
udpNetworkIPv6
)
// newUDPNetwork encodes one of the three networks accepted by UDP APIs.
func newUDPNetwork(network string) udpNetwork {
switch network {
case "udp4":
return udpNetworkIPv4
case "udp6":
return udpNetworkIPv6
default:
return udpNetworkGeneric
}
}
// name reconstructs the original network for net.OpError.
func (n udpNetwork) name() string {
switch n {
case udpNetworkIPv4:
return "udp4"
case udpNetworkIPv6:
return "udp6"
default:
return "udp"
}
}
// exclusiveUDPSocketBinding is the default one-owner bind policy.
type exclusiveUDPSocketBinding struct{}
// reuseAddressUDPSocketBinding permits overlap only with other SO_REUSEADDR-
// style sockets. Unicast delivery selects the most recently registered exact
// or wildcard binding, matching Linux UDP binding precedence.
type reuseAddressUDPSocketBinding struct{}
// UDPConn is a connected or unconnected userspace UDP socket.
type UDPConn struct {
stack *Stack
net udpNetwork
port uint16
v6 bool
dual bool
// forwarded authorizes this socket's intercepted nonlocal address as an
// explicit output source while promiscuous admission remains enabled.
forwarded bool
reuseAddress bool
reusePort bool
local netip.Addr
remote netip.AddrPort
datagramSocketWriteControl
mu sync.Mutex
receive datagramQueue[udpDatagram]
receiveSpare []byte
receiveNotify chan struct{}
receiveCapacity int
queuedBytes int
errorState *datagramSocketErrorState
receiveErrors bool
readDeadline datagramSocketDeadline
recentTargets recentDestinationCache[netip.AddrPort]
defaultOptions ipPacketOptions
pathMTUDiscovery PathMTUDiscovery
multicastHopLimit byte
multicastLoopback bool
broadcast bool
automaticLabel uint32
packetsSent atomic.Uint64
bytesSent atomic.Uint64
packetsReceived atomic.Uint64
bytesReceived atomic.Uint64
packetsDropped atomic.Uint64
}
// udpWriteParameters is one validated output-policy snapshot shared by
// contiguous and scatter/gather writes.
type udpWriteParameters struct {
source netip.Addr
target netip.AddrPort
options ipPacketOptions
pathMTUDiscovery PathMTUDiscovery
receiveErrors bool
nonUnicast bool
}
// udpDatagramWriter emits one prepared datagram without retaining its payload.
type udpDatagramWriter func(source, target netip.Addr, sourcePort, targetPort uint16, payload []byte, options ipPacketOptions, pathMTUDiscovery PathMTUDiscovery, nonUnicast bool) error
// newUDPConn creates an unregistered UDP socket after applying explicit
// creation policies to the latest Stack defaults.
func newUDPConn(stack *Stack, network string, port uint16, v6 bool, local netip.Addr, remote netip.AddrPort, options datagramSocketOptionSet) *UDPConn {
defaults := DatagramSocketDefaults{ReceiveBuffer: udpDefaultReceiveCapacity, HopLimit: 64, MulticastHopLimit: 1}
if stack != nil {
defaults = stack.network.Load().udpDefaults.DatagramSocketDefaults
}
defaults = applyDatagramSocketOptions(defaults, options, udpDatagramMetadataSize)
connection := &UDPConn{
stack: stack, net: newUDPNetwork(network), port: port, v6: v6, local: local, remote: remote,
datagramSocketWriteControl: datagramSocketWriteControl{closed: make(chan struct{})},
receiveCapacity: defaults.ReceiveBuffer,
defaultOptions: ipPacketOptions{
hopLimit: byte(defaults.HopLimit), trafficClass: defaults.TrafficClass,
flowLabel: defaults.FlowLabel, hopLimitSet: options.hopLimit.set,
trafficClassSet: options.trafficClass.set, flowLabelSet: defaults.FlowLabel != 0 || options.flowLabel.set,
},
receiveErrors: defaults.ReceiveErrors,
pathMTUDiscovery: defaults.PathMTUDiscovery,
multicastHopLimit: byte(defaults.MulticastHopLimit), multicastLoopback: !defaults.DisableMulticastLoopback,
broadcast: !defaults.DisableBroadcast,
}
if remote.IsValid() && stack != nil && local.Is6() && remote.Addr().Is6() && defaults.FlowLabel == 0 && !options.flowLabel.set {
connection.automaticLabel = stack.automaticTransportFlowLabel(local, remote.Addr(), ProtocolUDP, port, remote.Port())
}
return connection
}
// available implements exclusive UDP binding.
func (exclusiveUDPSocketBinding) available(stack *Stack, address netip.Addr, port uint16, dual bool) bool {
return stack.udpReuse == nil || !stack.udpReuse.overlaps(address, port, dual)
}
// register adds one exclusive UDP socket while Stack.mu is held.
func (exclusiveUDPSocketBinding) register(stack *Stack, connection *UDPConn) error {
stack.udp[udpKey{address: connection.local, port: connection.port}] = connection
return nil
}
// available accepts an exact group with matching dual-stack behavior. The
// shared listen core separately rejects overlap with every exclusive socket.
func (reuseAddressUDPSocketBinding) available(stack *Stack, address netip.Addr, port uint16, dual bool) bool {
return reusableUDPBindingAvailable(stack, address, port, dual, true, false)
}
// register adds one SO_REUSEADDR-style UDP socket to the shared registry.
func (reuseAddressUDPSocketBinding) register(stack *Stack, connection *UDPConn) error {
connection.reuseAddress = true
return registerReusableUDP(stack, connection)
}
// handleUDP validates and dispatches one unicast, broadcast, or multicast UDP
// datagram according to its already classified IP destination. skipChecksum
// requires a link or builder guarantee and does not allow zero IPv6 checksums.
func (s *Stack) handleUDP(packet ipPacket, destination inboundDestinationClass, skipChecksum bool) error {
udp := packet.payload
length, valid := udpWireLength(udp)
if !valid {
s.stats.inboundDroppedPackets.Add(1)
return nil
}
checksumValue := binary.BigEndian.Uint16(udp[6:8])
if packet.source.Is6() && checksumValue == 0 || !skipChecksum && checksumValue != 0 && transportChecksum(packet.source, packet.target, ProtocolUDP, udp[:length]) != 0 {
s.stats.inboundDroppedPackets.Add(1)
return nil
}
sourcePort := binary.BigEndian.Uint16(udp[0:2])
targetPort := binary.BigEndian.Uint16(udp[2:4])
source := netip.AddrPortFrom(packet.source, sourcePort)
target := netip.AddrPortFrom(packet.target, targetPort)
packet.payload = udp[:length]
if destination == inboundDestinationMulticast {
s.mu.RLock()
multicast := s.multicast
s.mu.RUnlock()
if multicast != nil {
multicast.deliverUDP(packet, sourcePort, targetPort)
}
return nil
}
if destination == inboundDestinationBroadcast {
s.deliverBroadcastUDP(packet, source, targetPort)
return nil
}
localDestination := destination == inboundDestinationLocalUnicast
s.mu.RLock()
connection := s.udpForwardedConnectionLocked(target, source)
if connection == nil && localDestination {
connection = s.udpOrdinaryConnectionLocked(target, source)
}
forwarder := s.udpForwarder
s.mu.RUnlock()
if connection == nil {
options := ipPacketOptions{hopLimit: packet.hopLimit, trafficClass: packet.trafficClass, flowLabel: packet.flowLabel}
if forwarder != nil && forwarder.handlePacket(packet, ForwarderFlow{Source: source, Destination: target}, options) {
return nil
}
if !localDestination {
return nil
}
_ = s.sendPortUnreachable(packet)
return nil
}
if (!connection.local.IsUnspecified() && packet.target != connection.local) || (connection.remote.IsValid() && source != connection.remote) {
s.stats.inboundDroppedPackets.Add(1)
return nil
}
connection.enqueue(udp[udpHeaderSize:length], source, packet.target, ipPacketOptions{
hopLimit: packet.hopLimit, trafficClass: packet.trafficClass, flowLabel: packet.flowLabel,
})
return nil
}
// deliverBroadcastUDP copies one IPv4 broadcast to every eligible wildcard
// binding. Unlike unicast REUSEPORT dispatch, Linux fans a broadcast out to
// every member of a matching reuse group.
func (s *Stack) deliverBroadcastUDP(packet ipPacket, source netip.AddrPort, targetPort uint16) {
s.mu.RLock()
var matchedStorage [8]*UDPConn
matched := matchedStorage[:0]
consider := func(connection *UDPConn) {
if connection.forwarded || connection.port != targetPort || !connection.local.IsUnspecified() ||
connection.local.Is6() && !connection.dual ||
connection.remote.IsValid() && connection.remote != source {
return
}
matched = append(matched, connection)
}
for _, connection := range s.udp {
consider(connection)
}
if s.udpReuse != nil {
for _, connection := range s.udpReuse.connections() {
consider(connection)
}
}
s.mu.RUnlock()
options := ipPacketOptions{hopLimit: packet.hopLimit, trafficClass: packet.trafficClass, flowLabel: packet.flowLabel}
for _, connection := range matched {
connection.enqueue(packet.payload[udpHeaderSize:], source, packet.target, options)
}
}
// udpConnectionLocked selects an exact binding before a family wildcard and
// a dual-stack wildcard. REUSEPORT groups hash the complete flow tuple.
func (s *Stack) udpConnectionLocked(local, remote netip.AddrPort) *UDPConn {
if connection := s.udpForwardedConnectionLocked(local, remote); connection != nil {
return connection
}
return s.udpOrdinaryConnectionLocked(local, remote)
}
// udpForwardedConnectionLocked selects a connected forwarded flow before an
// unconnected forwarded endpoint bound to the original destination.
func (s *Stack) udpForwardedConnectionLocked(local, remote netip.AddrPort) *UDPConn {
if connection := s.udpForwarded[udpFlowKey{local: local, remote: remote}]; connection != nil {
return connection
}
return s.udpForwarded[udpFlowKey{local: local}]
}
// udpOrdinaryConnectionLocked selects an ordinary exact, wildcard, or
// REUSEPORT binding without considering forwarded four-tuples.
func (s *Stack) udpOrdinaryConnectionLocked(local, remote netip.AddrPort) *UDPConn {
if connection := s.udp[udpKey{address: local.Addr(), port: local.Port()}]; connection != nil {
return connection
}
if s.udpReuse != nil {
if connection := s.udpReuse.connection(local, local, remote); connection != nil {
return connection
}
}
wildcard := netip.IPv4Unspecified()
if local.Addr().Is6() {
wildcard = netip.IPv6Unspecified()
}
wildcardLocal := netip.AddrPortFrom(wildcard, local.Port())
if connection := s.udp[udpKey{address: wildcard, port: local.Port()}]; connection != nil {
return connection
}
if s.udpReuse != nil {
if connection := s.udpReuse.connection(wildcardLocal, local, remote); connection != nil {
return connection
}
}
if local.Addr().Is4() {
dualLocal := netip.AddrPortFrom(netip.IPv6Unspecified(), local.Port())
if connection := s.udp[udpKey{address: dualLocal.Addr(), port: local.Port()}]; connection != nil && connection.dual {
return connection
}
if s.udpReuse != nil {
if connection := s.udpReuse.connection(dualLocal, local, remote); connection != nil && connection.dual {
return connection
}
}
}
return nil
}
// acceptUDP registers one handler-approved connected endpoint and queues its
// triggering datagram before making the flow visible to concurrent input.
func (f *UDPForwarder) acceptUDP(request *UDPForwarderRequest, options datagramSocketOptionSet) (*UDPConn, error) {
stack := f.stack
stack.mu.Lock()
defer stack.mu.Unlock()
if stack.closed {
return nil, ErrClosed
}
if stack.udpForwarder != f {
return nil, net.ErrClosed
}
state := stack.network.Load()
local, remote := request.flow.Destination, request.flow.Source
if !state.acceptsInboundDestination(local.Addr()) {
return nil, syscall.EADDRNOTAVAIL
}
if _, routed := state.routeFor(remote.Addr()); !routed {
return nil, syscall.ENETUNREACH
}
key := udpFlowKey{local: local, remote: remote}
if stack.udpForwardedConnectionLocked(local, remote) != nil || networkStateHasLocal(state, local.Addr()) && stack.udpOrdinaryConnectionLocked(local, remote) != nil {
return nil, syscall.EADDRINUSE
}
network := "udp4"
if local.Addr().Is6() {
network = "udp6"
}
connection := newUDPConn(stack, network, local.Port(), local.Addr().Is6(), local.Addr(), remote, options)
connection.forwarded = true
connection.enqueue(request.Payload(), remote, local.Addr(), request.options)
if stack.udpForwarded == nil {
stack.udpForwarded = make(map[udpFlowKey]*UDPConn)
}
stack.udpForwarded[key] = connection
stack.stats.activeUDPSockets.Add(1)
return connection, nil
}
// listenUDP registers one handler-approved unconnected endpoint and queues its
// triggering datagram before making the local binding visible to concurrent
// input.
func (f *UDPForwarder) listenUDP(request *UDPForwarderRequest, options datagramSocketOptionSet) (*UDPConn, error) {
stack := f.stack
stack.mu.Lock()
defer stack.mu.Unlock()
if stack.closed {
return nil, ErrClosed
}
if stack.udpForwarder != f {
return nil, net.ErrClosed
}
state := stack.network.Load()
local, remote := request.flow.Destination, request.flow.Source
if !state.acceptsInboundDestination(local.Addr()) {
return nil, syscall.EADDRNOTAVAIL
}
key := udpFlowKey{local: local}
for existing := range stack.udpForwarded {
if existing.local == local {
return nil, syscall.EADDRINUSE
}
}
if networkStateHasLocal(state, local.Addr()) &&
!stack.udpEndpointAvailableLocked(exclusiveUDPSocketBinding{}, local.Addr(), local.Port(), false) {
return nil, syscall.EADDRINUSE
}
network := "udp4"
if local.Addr().Is6() {
network = "udp6"
}
connection := newUDPConn(stack, network, local.Port(), local.Addr().Is6(), local.Addr(), netip.AddrPort{}, options)
connection.forwarded = true
connection.enqueue(request.Payload(), remote, local.Addr(), request.options)
if stack.udpForwarded == nil {
stack.udpForwarded = make(map[udpFlowKey]*UDPConn)
}
stack.udpForwarded[key] = connection
stack.stats.activeUDPSockets.Add(1)
return connection, nil
}
// validateUDPForwarderReply normalizes one caller-selected source and checks
// only properties required to serialize a reverse-flow UDP datagram.
func validateUDPForwarderReply(flow ForwarderFlow, payload []byte, source netip.AddrPort) (netip.AddrPort, error) {
address := source.Addr()
if !source.IsValid() {
return netip.AddrPort{}, syscall.EINVAL
}
address = address.WithZone("").Unmap()
if address.Is6() != flow.Source.Addr().Unmap().Is6() {
return netip.AddrPort{}, syscall.EINVAL
}
if err := validateUDPForwarderReplyPayload(address.Is6(), payload); err != nil {
return netip.AddrPort{}, err
}
return netip.AddrPortFrom(address, source.Port()), nil
}
// validateUDPForwarderReplyPayload checks the address-family datagram limit.
func validateUDPForwarderReplyPayload(ipv6 bool, payload []byte) error {
maximumPayload := 65535 - udpHeaderSize
if !ipv6 {
maximumPayload -= 20
}
if len(payload) > maximumPayload {
return syscall.EMSGSIZE
}
return nil
}
// replyUDPFlow sends one reverse-flow response from a validated caller-selected
// source without retaining a registered endpoint after the write completes.
func (f *forwarderRuntime) replyUDPFlow(flow ForwarderFlow, payload []byte, source netip.AddrPort) (int, error) {
local, remote := flow.Destination, flow.Source
state := f.stack.network.Load()
if !state.acceptsInboundDestination(local.Addr()) {
return 0, syscall.EADDRNOTAVAIL
}
if _, routed := state.routeFor(remote.Addr()); !routed {
return 0, syscall.ENETUNREACH
}
defaults := state.udpDefaults.DatagramSocketDefaults
options := ipPacketOptions{
hopLimit: byte(defaults.HopLimit), trafficClass: defaults.TrafficClass,
flowLabel: defaults.FlowLabel, flowLabelSet: defaults.FlowLabel != 0,
}
if err := f.stack.writeBestEffortUDPDatagram(source.Addr(), remote.Addr(), source.Port(), remote.Port(), payload, options, defaults.PathMTUDiscovery); err != nil {
return 0, err
}
return len(payload), nil
}
// acceptsLocal reports whether an ICMP quote belongs to this socket's local
// address binding.
func (c *UDPConn) acceptsLocal(address netip.Addr) bool {
return c.local.IsUnspecified() || c.local == address.Unmap()
}
// enqueue copies and retains a datagram unless the configured receive
// capacity is full. The capacity check precedes allocation on the drop path.
func (c *UDPConn) enqueue(payload []byte, source netip.AddrPort, target netip.Addr, options ipPacketOptions) {
size := udpDatagramSize(payload)
c.mu.Lock()
select {
case <-c.closed:
c.mu.Unlock()
c.stack.stats.inboundDroppedPackets.Add(1)
c.packetsDropped.Add(1)
return
default:
}
if size > c.receiveCapacity || c.queuedBytes+c.errorState.bytes() > c.receiveCapacity-size {
c.mu.Unlock()
c.stack.stats.inboundDroppedPackets.Add(1)
c.packetsDropped.Add(1)
return
}
var retained []byte
if len(payload) != 0 {
if cap(c.receiveSpare) >= len(payload) {
retained = c.receiveSpare[:len(payload)]
c.receiveSpare = nil
copy(retained, payload)
} else {
retained = append([]byte(nil), payload...)
}
}
datagram := udpDatagram{payload: retained, source: source, target: target, options: options}
c.receive.push(datagram)
c.queuedBytes += size
c.packetsReceived.Add(1)
c.bytesReceived.Add(uint64(len(payload)))
c.notifyReceiveLocked()
c.mu.Unlock()
}
// notifyReceiveLocked keeps one edge notification armed while data or an
// ordinary socket error remains and removes a stale token otherwise.
func (c *UDPConn) notifyReceiveLocked() {
if c.receiveNotify == nil {
return
}
if c.receive.len() != 0 || c.errorState != nil && c.errorState.pending != nil {
select {
case c.receiveNotify <- struct{}{}:
default:
}
return
}
select {
case <-c.receiveNotify:
default:
}
}
// receiveNotificationLocked returns the shared edge notification, allocating
// it only when an empty receive path is about to block.
func (c *UDPConn) receiveNotificationLocked() <-chan struct{} {
if c.receiveNotify == nil {
c.receiveNotify = make(chan struct{}, 1)
}
return c.receiveNotify
}
// udpDatagramSize returns the approximate retained-memory cost of a payload.
func udpDatagramSize(payload []byte) int {
return udpDatagramMetadataSize + len(payload)
}
// ReadFrom returns the next complete datagram or socket error.
func (c *UDPConn) ReadFrom(buffer []byte) (int, net.Addr, error) {
n, source, _, _, _, err := c.readDatagram(buffer, nil)
var address net.Addr
if source.IsValid() {
address = net.UDPAddrFromAddrPort(source)
}
if err != nil {
return n, address, c.operationError("read", c.remoteAddr(), err)
}
return n, address, nil
}
// ReadFromUDP acts like ReadFrom but returns a UDPAddr.
func (c *UDPConn) ReadFromUDP(buffer []byte) (int, *net.UDPAddr, error) {
n, source, _, _, _, err := c.readDatagram(buffer, nil)
var address *net.UDPAddr
if source.IsValid() {
address = net.UDPAddrFromAddrPort(source)
}
if err != nil {
return n, address, c.operationError("read", c.remoteAddr(), err)
}
return n, address, nil
}
// ReadFromUDPAddrPort acts like ReadFrom but returns a netip.AddrPort.
func (c *UDPConn) ReadFromUDPAddrPort(buffer []byte) (int, netip.AddrPort, error) {
n, source, _, _, _, err := c.readDatagram(buffer, nil)
if err != nil {
return n, source, c.operationError("read", c.remoteAddr(), err)
}
return n, source, nil
}
// ReadFromWithBuffer reads the next complete datagram like ReadFrom, obtaining
// the destination buffer lazily from getBuffer and returning its source
// address.
//
// If a datagram is available, getBuffer is called once after it is dequeued.
// It is not called when the operation returns before a datagram is available
// because of an error or deadline. The callback receives the complete UDP
// payload length as an advisory size hint. It runs without c's connection state
// lock and must return promptly; it must not call a read method on c. The caller
// owns the returned slice, and c does not retain it. A nil callback returns
// EINVAL. A short returned slice truncates and consumes the datagram, matching
// ReadFrom.
//
// This is an experimental API and is not covered by the package's stability
// guarantees.
func (c *UDPConn) ReadFromWithBuffer(getBuffer func(sizeHint int) []byte) (int, net.Addr, error) {
if getBuffer == nil {
return 0, nil, c.operationError("read", c.remoteAddr(), syscall.EINVAL)
}
n, source, _, _, _, err := c.readDatagram(nil, getBuffer)
var address net.Addr
if source.IsValid() {
address = net.UDPAddrFromAddrPort(source)
}
if err != nil {
return n, address, c.operationError("read", c.remoteAddr(), err)
}
return n, address, nil
}
// ReadFromUDPWithBuffer is the *net.UDPAddr form of ReadFromWithBuffer.
//
// It has the same callback, buffer ownership, truncation, and nil-callback
// semantics as ReadFromWithBuffer.
//
// This is an experimental API and is not covered by the package's stability
// guarantees.
func (c *UDPConn) ReadFromUDPWithBuffer(getBuffer func(sizeHint int) []byte) (int, *net.UDPAddr, error) {
if getBuffer == nil {
return 0, nil, c.operationError("read", c.remoteAddr(), syscall.EINVAL)
}
n, source, _, _, _, err := c.readDatagram(nil, getBuffer)
var address *net.UDPAddr
if source.IsValid() {
address = net.UDPAddrFromAddrPort(source)
}
if err != nil {
return n, address, c.operationError("read", c.remoteAddr(), err)
}
return n, address, nil
}
// ReadFromUDPAddrPortWithBuffer is the netip.AddrPort form of
// ReadFromWithBuffer.
//
// It has the same callback, buffer ownership, truncation, and nil-callback
// semantics as ReadFromWithBuffer.
//
// This is an experimental API and is not covered by the package's stability
// guarantees.
func (c *UDPConn) ReadFromUDPAddrPortWithBuffer(getBuffer func(sizeHint int) []byte) (int, netip.AddrPort, error) {
if getBuffer == nil {
return 0, netip.AddrPort{}, c.operationError("read", c.remoteAddr(), syscall.EINVAL)
}
n, source, _, _, _, err := c.readDatagram(nil, getBuffer)
if err != nil {
return n, source, c.operationError("read", c.remoteAddr(), err)
}
return n, source, nil
}
// ReadMsgUDP reads one datagram and Linux-compatible packet-info ancillary
// data. The control message identifies the packet's IP destination address.
func (c *UDPConn) ReadMsgUDP(buffer, oob []byte) (n, oobn, flags int, address *net.UDPAddr, err error) {
var source netip.AddrPort
n, oobn, flags, source, err = c.readMsgUDPAddrPort(buffer, oob)
if source.IsValid() {
address = net.UDPAddrFromAddrPort(source)
}
return
}
// ReadMsgUDPAddrPort is the netip.AddrPort form of ReadMsgUDP.
func (c *UDPConn) ReadMsgUDPAddrPort(buffer, oob []byte) (n, oobn, flags int, source netip.AddrPort, err error) {
return c.readMsgUDPAddrPort(buffer, oob)
}
// readMsgUDPAddrPort reads one datagram and encodes its IP destination as
// Linux IP_PKTINFO or IPV6_PKTINFO.
func (c *UDPConn) readMsgUDPAddrPort(buffer, oob []byte) (n, oobn, flags int, source netip.AddrPort, err error) {
var target netip.Addr
var options ipPacketOptions
var truncated bool
n, source, target, options, truncated, err = c.readDatagram(buffer, nil)
if truncated {
flags |= MessageFlagTruncated
}
if err != nil {
err = c.operationError("read", c.remoteAddr(), err)
return
}
var specDst netip.Addr
if target.Is4() {
specDst = c.stack.network.Load().inboundIPv4PacketInfoSource(source.Addr(), target)
}
control, controlErr := controlMessageForRead(specDst, target, options)
if controlErr != nil {
err = c.operationError("read", c.remoteAddr(), controlErr)
return
}
var controlTruncated bool
oobn, controlTruncated = copyLinuxControlMessages(oob, control)
if controlTruncated {
flags |= MessageFlagControlTruncated
}
return
}
// ReadBatch reads one or more UDP messages using the SocketMessage layout shared by
// x/net/ipv4 and x/net/ipv6. The first message follows the socket's blocking
// and deadline semantics; after it succeeds, the method drains only messages
// already queued. MessageFlagDontWait also makes the first read nonblocking.
func (c *UDPConn) ReadBatch(messages []SocketMessage, flags int) (int, error) {
if flags&^(MessageFlagPeek|MessageFlagDontWait|MessageFlagTruncated|MessageFlagErrorQueue) != 0 {
return 0, c.operationError("read", c.remoteAddr(), syscall.EOPNOTSUPP)
}
if flags&MessageFlagErrorQueue != 0 {
return c.readErrorBatch(messages)
}
var network *networkState
for index := range messages {
wait := index == 0 && flags&MessageFlagDontWait == 0
err := c.readBatchMessage(&messages[index], flags, wait, index == 0, &network)
if err != nil {
// recvmmsg reports a completed prefix without the error that stopped
// the next message. A retry starting at index exposes that error.
if index != 0 {
return index, nil
}
return index, err
}
}
return len(messages), nil
}
// readBatchMessage receives one scatter/gather message without waiting when
// wait is false. consumeErrors is false after a successful prefix; a pending
// asynchronous error then stops the batch without consuming it. network
// caches one immutable configuration snapshot for IPv4 packet-info fields.
func (c *UDPConn) readBatchMessage(message *SocketMessage, flags int, wait, consumeErrors bool, network **networkState) error {
if _, err := messageBufferLength(message.Buffers); err != nil {
return c.operationError("read", c.remoteAddr(), err)
}
n, source, target, options, truncated, err := c.readDatagramBuffers(message.Buffers, wait, consumeErrors, flags&MessageFlagPeek != 0, flags&MessageFlagTruncated != 0)
if err != nil {
return c.operationError("read", c.remoteAddr(), err)
}
var specDst netip.Addr
if target.Is4() {
if *network == nil {
*network = c.stack.network.Load()
}
specDst = (*network).inboundIPv4PacketInfoSource(source.Addr(), target)
}
control, err := controlMessageForRead(specDst, target, options)
if err != nil {
return c.operationError("read", c.remoteAddr(), err)
}
resultFlags := 0
if truncated {
resultFlags |= MessageFlagTruncated
}
oobn, controlTruncated := copyLinuxControlMessages(message.OOB, control)
if controlTruncated {
resultFlags |= MessageFlagControlTruncated
}
message.N, message.NN, message.Flags = n, oobn, resultFlags
if source.IsValid() {
message.Addr = net.UDPAddrFromAddrPort(source)
} else {
message.Addr = nil
}
return nil
}
// Read receives the next datagram from a connected remote endpoint.
func (c *UDPConn) Read(buffer []byte) (int, error) {
// Go net on Linux treats an empty Read as a local-closure check, not
// a datagram receive. Address and message reads still consume a packet.
if len(buffer) == 0 {
select {
case <-c.closed:
return 0, c.operationError("read", c.remoteAddr(), net.ErrClosed)
default:
}
return 0, nil
}
n, _, _, _, _, err := c.readDatagram(buffer, nil)
if err != nil {
return n, c.operationError("read", c.remoteAddr(), err)
}
return n, nil
}
// ReadWithBuffer reads the next datagram from a connected remote endpoint like