From 9fcb7e9b5295df6f4afd0dd0889c21ec15df9d6a Mon Sep 17 00:00:00 2001 From: Pat Whittingslow Date: Thu, 14 May 2026 12:48:07 -0300 Subject: [PATCH] Add udp.PacketConn (#112) * update README table and begin planning udp.PacketConn * first attempt at packetconn implementation * add udp.PacketConn to StackAsync and implement SocketNetip iface * add tests for PacketConn --- README.md | 12 +- udp/packetconn.go | 260 +++++++++++++++++++++++++++ x/xnet/stack-async.go | 16 +- x/xnet/stack-go.go | 62 ++++++- x/xnet/stack6_test.go | 4 +- x/xnet/xnet_udppktconn_test.go | 310 +++++++++++++++++++++++++++++++++ 6 files changed, 647 insertions(+), 17 deletions(-) create mode 100644 udp/packetconn.go create mode 100644 x/xnet/xnet_udppktconn_test.go diff --git a/README.md b/README.md index 42b657c..cee02a9 100644 --- a/README.md +++ b/README.md @@ -3,7 +3,7 @@ [![Go Report Card](https://goreportcard.com/badge/github.com/soypat/lneto)](https://goreportcard.com/report/github.com/soypat/lneto) [![codecov](https://codecov.io/gh/soypat/lneto/branch/main/graph/badge.svg)](https://codecov.io/gh/soypat/lneto) [![Go](https://github.com/soypat/lneto/actions/workflows/ci.yaml/badge.svg)](https://github.com/soypat/lneto/actions/workflows/ci.yaml) -[![sourcegraph](https://sourcegraph.com/github.com/soypat/lneto/-/badge.svg)](https://sourcegraph.com/github.com/soypat/lneto?badge) +[![sourcegraph](https://sourcegraph.com/github.com/soypat/lneto/-/badge.svg)](https://github.com/soypat/lneto/network/dependents) Userspace networking primitives. @@ -96,14 +96,14 @@ ok github.com/soypat/lneto/x/xnet 2.926s | UDP | RFC 768 | ✅ | `udp` | — | Handler + thread-safe `Conn` | | TCP | RFC 9293 | ✅ | `tcp` | 0 ²³ | Full state machine, SYN cookies, retransmit queue, `Conn`/`Listener` | | DNS | RFC 1035 | ✅ | `dns` | — | Client (A/AAAA query) | -| DHCPv4 | RFC 2131 | ✅ | `dhcpv4` | — | Client + Server | +| DHCPv4 | RFC 2131 | ✅ | `dhcp/dhcpv4` | — | Client + Server | | NTP | RFC 5905 | ✅ | `ntp` | — | Client | -| mDNS | RFC 6762 | ✅ | `mdns` | — | Client (service announcement + query) | +| mDNS | RFC 6762 | ✅ | `dns/mdns` | — | Client (service announcement + query) | | HTTP/1.1 headers | RFC 9110, RFC 9112 | ✅ | `http/httpraw` | 2 ⁴ | Header parse/format; no field normalization | | Ethernet PHY/MDIO | IEEE 802.3 cl.22/45 | ✅ | `phy` | — | Bare-metal PHY management via MDIO | -| IPv6 | RFC 8200 | 🟡 | `ipv6` | — | Frame parsing only; no stack handler | -| ICMPv6 | RFC 4443 | ❌ | — | — | Not implemented | -| DHCPv6 | RFC 8415 | ❌ | — | — | Not implemented | +| IPv6 | RFC 8200 | ✅ | `ipv6` | — | Frame parsing and stack handling | +| ICMPv6 | RFC 4443 | ✅ | `ipv6/icmpv6` | — | Echo+NDP frame parsing and stack handling | +| DHCPv6 | RFC 8415 | 🟡 | `dhcp/dhcpv6` | — | Frame parsing and standalone handling | | TLS 1.3 | RFC 8446 | ❌ | — | — | Not implemented | ¹ `BenchmarkARPExchange` — full ARP request/response exchange over Ethernet: **0 B/op, 0 allocs/op** diff --git a/udp/packetconn.go b/udp/packetconn.go new file mode 100644 index 0000000..6fe7d44 --- /dev/null +++ b/udp/packetconn.go @@ -0,0 +1,260 @@ +package udp + +import ( + "net" + "net/netip" + "os" + "sync" + "time" + + "github.com/soypat/lneto" + "github.com/soypat/lneto/internal" +) + +// lnetopacketconn is the lneto interpretation of +// [net.PacketConn], making use of better types. +type lnetopacketconn interface { // size=16 (0x10) + ReadFrom(p []byte) (n int, addr netip.AddrPort, err error) + WriteTo(p []byte, addr netip.AddrPort) (n int, err error) + Close() error + LocalAddr() netip.AddrPort + SetDeadline(t time.Time) error + SetReadDeadline(t time.Time) error + SetWriteDeadline(t time.Time) error +} + +var ( + _ lnetopacketconn = (*PacketConn)(nil) + _ lneto.StackNode = (*PacketConn)(nil) +) + +// PacketConn is the UDP equivalent of [net.PacketConn] and implements +// [lnetopacketconn] and [lneto.StackNode]. It is thread safe. +type PacketConn struct { + mu sync.Mutex + m muxHandler + localAddr netip.AddrPort + _backoff lneto.BackoffStrategy + rdead time.Time + wdead time.Time +} + +// PacketConnConfig configures a [PacketConn] with pre-allocated buffers and queue sizes. +type PacketConnConfig struct { + RxBuf []byte + TxBuf []byte + RxQueueSize int + TxQueueSize int + // RWBackoff sets the backoff policy when data is unavailable on ReadFrom or buffer is full on WriteTo. + // If not set a default backoff strategy will be used. See [internal.BackoffConnRW]. + RWBackoff lneto.BackoffStrategy +} + +// Configure initializes the PacketConn with the given buffer and queue configuration. +// Must be called before [PacketConn.Open]. Calling Configure on an active connection aborts it. +func (pc *PacketConn) Configure(cfg PacketConnConfig) error { + pc.mu.Lock() + defer pc.mu.Unlock() + pc.abort() + err := pc.m.Configure(MuxConfig{ + RxBuf: cfg.RxBuf, + TxBuf: cfg.TxBuf, + RxQueueSize: cfg.RxQueueSize, + TxQueueSize: cfg.TxQueueSize, + }) + if err != nil { + return err + } + pc._backoff = cfg.RWBackoff + return nil +} + +// Open sets the local address and enables port filtering for incoming datagrams. +func (pc *PacketConn) Open(localAddr netip.AddrPort) error { + pc.mu.Lock() + defer pc.mu.Unlock() + if pc.localAddr.IsValid() { + return errStillOpen + } + if !localAddr.IsValid() || localAddr.Port() == 0 { + return lneto.ErrZeroSource + } + pc.localAddr = localAddr + pc.m.FilterResetLocalPorts() + pc.m.FilterAddLocalPort(localAddr.Port(), 1) + return nil +} + +// Abort resets the connection, discarding all buffered data and clearing deadlines. +func (pc *PacketConn) Abort() { + pc.mu.Lock() + defer pc.mu.Unlock() + pc.abort() +} + +func (pc *PacketConn) abort() { + pc.m.Abort() + pc.rdead = time.Time{} + pc.wdead = time.Time{} + pc.localAddr = netip.AddrPort{} +} + +// Close marks the PacketConn as closed. Subsequent WriteTo calls return [net.ErrClosed]. +// ReadFrom continues to drain buffered datagrams until exhausted, then returns [net.ErrClosed]. +func (pc *PacketConn) Close() error { + pc.mu.Lock() + defer pc.mu.Unlock() + pc.m.Close() + return nil +} + +// LocalAddr returns the local address set by [PacketConn.Open]. +func (pc *PacketConn) LocalAddr() netip.AddrPort { + pc.mu.Lock() + defer pc.mu.Unlock() + return pc.localAddr +} + +// LocalPort implements [lneto.StackNode]. +func (pc *PacketConn) LocalPort() uint16 { + pc.mu.Lock() + defer pc.mu.Unlock() + return pc.localAddr.Port() +} + +// Protocol implements [lneto.StackNode]. +func (pc *PacketConn) Protocol() uint64 { return uint64(lneto.IPProtoUDP) } + +// ConnectionID implements [lneto.StackNode]. +func (pc *PacketConn) ConnectionID() *uint64 { return &pc.m.connid } + +// Demux implements [lneto.StackNode]. +func (pc *PacketConn) Demux(carrierData []byte, frameOffset int) error { + pc.mu.Lock() + defer pc.mu.Unlock() + return pc.m.Demux(carrierData, frameOffset) +} + +// Encapsulate implements [lneto.StackNode]. +func (pc *PacketConn) Encapsulate(carrierData []byte, offsetToIP, offsetToFrame int) (int, error) { + pc.mu.Lock() + defer pc.mu.Unlock() + return pc.m.Encapsulate(carrierData, offsetToIP, offsetToFrame) +} + +// ReadFrom dequeues the next received datagram into p and returns the sender's address. +// Blocks until a datagram is available or the read deadline is exceeded. +func (pc *PacketConn) ReadFrom(p []byte) (n int, addr netip.AddrPort, err error) { + connID, err := pc.lockConnID() + if err != nil { + return 0, netip.AddrPort{}, err + } + var backoffs uint + for { + pc.mu.Lock() + if (pc.m.closeCalled && pc.m.BufferedInput() == 0) || connID != pc.m.connid { + pc.mu.Unlock() + return 0, netip.AddrPort{}, net.ErrClosed + } + n, _, _, addr = pc.m.ReadNext(p) + pc.mu.Unlock() + if n > 0 { + return n, addr, nil + } + if pc.deadlineExceeded(&pc.rdead) { + return 0, netip.AddrPort{}, os.ErrDeadlineExceeded + } + pc.backoff(backoffs) + backoffs++ + } +} + +// WriteTo enqueues a datagram for transmission to addr. +// Blocks until buffer space is available or the write deadline is exceeded. +func (pc *PacketConn) WriteTo(p []byte, addr netip.AddrPort) (n int, err error) { + if len(p) == 0 { + return 0, nil + } + connID, err := pc.lockConnID() + if err != nil { + return 0, err + } + var backoffs uint + for { + pc.mu.Lock() + if pc.m.closeCalled || connID != pc.m.connid { + pc.mu.Unlock() + return 0, net.ErrClosed + } + werr := pc.m.WriteTo(p, pc.localAddr.Port(), addr) + pc.mu.Unlock() + if werr == nil { + return len(p), nil + } + if werr != lneto.ErrExhausted && werr != lneto.ErrBufferFull { + return 0, werr + } + if pc.deadlineExceeded(&pc.wdead) { + return 0, os.ErrDeadlineExceeded + } + pc.backoff(backoffs) + backoffs++ + } +} + +// SetDeadline sets both the read and write deadlines. A zero value disables the deadline. +func (pc *PacketConn) SetDeadline(t time.Time) error { + pc.mu.Lock() + defer pc.mu.Unlock() + if pc.m.closeCalled { + return net.ErrClosed + } + pc.rdead = t + pc.wdead = t + return nil +} + +// SetReadDeadline sets the read deadline. A zero value disables the deadline. +func (pc *PacketConn) SetReadDeadline(t time.Time) error { + pc.mu.Lock() + defer pc.mu.Unlock() + if pc.m.closeCalled { + return net.ErrClosed + } + pc.rdead = t + return nil +} + +// SetWriteDeadline sets the write deadline. A zero value disables the deadline. +func (pc *PacketConn) SetWriteDeadline(t time.Time) error { + pc.mu.Lock() + defer pc.mu.Unlock() + if pc.m.closeCalled { + return net.ErrClosed + } + pc.wdead = t + return nil +} + +func (pc *PacketConn) deadlineExceeded(deadline *time.Time) bool { + pc.mu.Lock() + defer pc.mu.Unlock() + return !deadline.IsZero() && time.Since(*deadline) > 0 +} + +func (pc *PacketConn) backoff(n uint) { + if pc._backoff != nil { + pc._backoff.Do(n) + } else { + internal.BackoffConnRW(n) + } +} + +func (pc *PacketConn) lockConnID() (uint64, error) { + pc.mu.Lock() + defer pc.mu.Unlock() + if pc.m.closeCalled && pc.m.BufferedInput() == 0 { + return 0, net.ErrClosed + } + return pc.m.connid, nil +} diff --git a/x/xnet/stack-async.go b/x/xnet/stack-async.go index fbccda5..5df5aaf 100644 --- a/x/xnet/stack-async.go +++ b/x/xnet/stack-async.go @@ -34,10 +34,8 @@ type StackAsync struct { link internet.StackEthernet ip4 internet.StackIPv4 - // ip6 internet.StackIPv6 - arp arp.Handler - icmp icmpv4.Client - // icmp6 icmpv6.Client + arp arp.Handler + icmp icmpv4.Client icmp6buf []byte udps internet.StackPortsMACFiltered tcps internet.StackPortsMACFiltered @@ -496,10 +494,6 @@ func (s *StackAsync) RegisterListener(listener *tcp.Listener) (err error) { // Can try changing listener to inspect carrierData on demux and get the IPversion to know which tcp.Conns match the IP version. s.mu.Lock() defer s.mu.Unlock() - lport := listener.LocalPort() - if lport == 0 { - return lneto.ErrZeroSource - } return s.tcps.RegisterMACFiltered(listener, nil) } @@ -518,6 +512,12 @@ func (s *StackAsync) RegisterUDP4(node lneto.StackNode, remoteAddr []byte, remot return s.udps.RegisterMACFiltered(&s.userUDPs[idx], nil) } +func (s *StackAsync) RegisterListenerUDP(pktconn *udp.PacketConn) (err error) { + s.mu.Lock() + defer s.mu.Unlock() + return s.udps.RegisterMACFiltered(pktconn, nil) +} + var errNoDNSServer = errors.New("no DNS server- did DHCP complete? You can set a predetermined DNS server in Stack configuration") func (s *StackAsync) StartLookupIP(host string) error { diff --git a/x/xnet/stack-go.go b/x/xnet/stack-go.go index 9bb6902..958cad2 100644 --- a/x/xnet/stack-go.go +++ b/x/xnet/stack-go.go @@ -2,9 +2,11 @@ package xnet import ( "context" + "math" "net" "net/netip" "syscall" + "time" "github.com/soypat/lneto" "github.com/soypat/lneto/tcp" @@ -86,7 +88,27 @@ func (s StackGo) SocketNetip(ctx context.Context, network string, family, sotype return nil, lneto.ErrUnsupported } if !raddr.IsValid() || raddr.Addr() == netip.IPv4Unspecified() { - return nil, lneto.ErrZeroDestination + // LISTEN UDP: no fixed remote → PacketConn. + var pc udppktconn + err = pc.c.Configure(udp.PacketConnConfig{ + TxBuf: make([]byte, s.plcfg.TxBufSize), + RxBuf: make([]byte, s.plcfg.RxBufSize), + TxQueueSize: s.plcfg.QueueSize, + RxQueueSize: s.plcfg.QueueSize, + }) + if err != nil { + return nil, err + } + err = pc.c.Open(laddr) + if err != nil { + return nil, err + } + pc.laddr = net.UDPAddr{IP: laddr.Addr().AsSlice(), Port: int(laddr.Port())} + err = s.blk.async.RegisterListenerUDP(&pc.c) + if err != nil { + return nil, err + } + return &pc, nil } var conn udp.Conn err = conn.Configure(udp.ConnConfig{ @@ -175,6 +197,44 @@ func (s StackGo) SocketNetip(ctx context.Context, network string, family, sotype return nil, lneto.ErrUnsupported } +// udppktconn implements [net.PacketConn] for [udp.PacketConn]. +type udppktconn struct { + c udp.PacketConn + laddr net.UDPAddr + raddr net.UDPAddr +} + +var _ net.PacketConn = (*udppktconn)(nil) + +func (u *udppktconn) ReadFrom(p []byte) (n int, addr net.Addr, err error) { + n, ap, err := u.c.ReadFrom(p) + if err != nil { + return n, nil, err + } + u.raddr.IP, _ = ap.Addr().AppendBinary(u.raddr.IP[:0]) + u.raddr.Port = int(ap.Port()) + u.raddr.Zone = "" + return n, &u.raddr, nil +} + +func (u *udppktconn) WriteTo(p []byte, addr net.Addr) (n int, err error) { + uaddr, ok := addr.(*net.UDPAddr) + ip, ok2 := netip.AddrFromSlice(uaddr.IP) + if !ok || !ok2 || uaddr.Port <= 0 || uaddr.Port > math.MaxUint16 { + return 0, lneto.ErrInvalidAddr + } + ap := netip.AddrPortFrom(ip, uint16(uaddr.Port)) + return u.c.WriteTo(p, ap) +} + +func (u *udppktconn) Close() error { return u.c.Close() } + +func (u *udppktconn) LocalAddr() net.Addr { return &u.laddr } + +func (u *udppktconn) SetDeadline(t time.Time) error { return u.c.SetDeadline(t) } +func (u *udppktconn) SetReadDeadline(t time.Time) error { return u.c.SetReadDeadline(t) } +func (u *udppktconn) SetWriteDeadline(t time.Time) error { return u.c.SetWriteDeadline(t) } + type tcplistener struct { l tcp.Listener closed bool diff --git a/x/xnet/stack6_test.go b/x/xnet/stack6_test.go index e0044f9..182a38d 100644 --- a/x/xnet/stack6_test.go +++ b/x/xnet/stack6_test.go @@ -22,7 +22,7 @@ func stack6PairConfigs(seed int64, maxports, icmpQueue uint16) (cfg1, cfg2 Stack testMAC6B = [6]byte{0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0x02} ) cfg1 = StackConfig{ - Hostname: "test-s6-1", + Hostname: "stack6-1", RandSeed: seed, StaticAddress6: testAddr6A, HardwareAddress: testMAC6A, @@ -32,7 +32,7 @@ func stack6PairConfigs(seed int64, maxports, icmpQueue uint16) (cfg1, cfg2 Stack ICMPQueueLimit: int(icmpQueue), } cfg2 = StackConfig{ - Hostname: "test-s6-2", + Hostname: "stack6-2", RandSeed: ^seed, StaticAddress6: testAddr6B, HardwareAddress: testMAC6B, diff --git a/x/xnet/xnet_udppktconn_test.go b/x/xnet/xnet_udppktconn_test.go new file mode 100644 index 0000000..979aec6 --- /dev/null +++ b/x/xnet/xnet_udppktconn_test.go @@ -0,0 +1,310 @@ +package xnet + +import ( + "bytes" + "net/netip" + "testing" + "time" + + "github.com/soypat/lneto/ethernet" + "github.com/soypat/lneto/udp" +) + +const ( + testUDPBufSize = 2048 + testUDPQueueSize = 4 +) + +// newUDPTestPair creates a server/client StackAsync pair for UDP tests with +// static addresses and each stack's gateway pointing at the peer. +func newUDPTestPair(t testing.TB, seed int64) (s1, s2 *StackAsync) { + t.Helper() + s1, s2 = new(StackAsync), new(StackAsync) + if err := s1.Reset(StackConfig{ + Hostname: "UDP-1", + RandSeed: seed, + StaticAddress4: [4]byte{10, 1, 0, 1}, + HardwareAddress: [6]byte{0xaa, 0xbb, 0, 0, 0, 1}, + MTU: ethernet.MaxMTU, + MaxActiveUDPPorts: 4, + }); err != nil { + t.Fatal("sv Reset:", err) + } + if err := s2.Reset(StackConfig{ + Hostname: "UDP-2", + RandSeed: ^seed, + StaticAddress4: [4]byte{10, 1, 0, 2}, + HardwareAddress: [6]byte{0xaa, 0xbb, 0, 0, 0, 2}, + MTU: ethernet.MaxMTU, + MaxActiveUDPPorts: 4, + }); err != nil { + t.Fatal("cl Reset:", err) + } + s1.SetGatewayHardwareAddr(s2.HardwareAddr()) + s2.SetGatewayHardwareAddr(s1.HardwareAddr()) + return s1, s2 +} + +// TestStackAsyncRegisterListenerUDP_ReceiveData registers a udp.PacketConn as +// a listener and verifies that a datagram from a dialed client is delivered +// with the correct payload and sender address. +func TestStackAsyncRegisterListenerUDP_ReceiveData(t *testing.T) { + const ( + svPort = 9000 + clPort = 9001 + ) + sv, cl := newUDPTestPair(t, 1234) + buf := make([]byte, ethernet.MaxMTU+ethernet.MaxOverheadSize) + + var pc udp.PacketConn + if err := pc.Configure(udp.PacketConnConfig{ + RxBuf: make([]byte, testUDPBufSize), TxBuf: make([]byte, testUDPBufSize), + RxQueueSize: testUDPQueueSize, TxQueueSize: testUDPQueueSize, + }); err != nil { + t.Fatal("pc Configure:", err) + } + if err := pc.Open(netip.AddrPortFrom(netip.AddrFrom4(sv.Addr4()), svPort)); err != nil { + t.Fatal("pc.Open:", err) + } + if err := sv.RegisterListenerUDP(&pc); err != nil { + t.Fatal("RegisterListenerUDP:", err) + } + + var conn udp.Conn + if err := conn.Configure(udp.ConnConfig{ + RxBuf: make([]byte, testUDPBufSize), TxBuf: make([]byte, testUDPBufSize), + RxQueueSize: testUDPQueueSize, TxQueueSize: testUDPQueueSize, + }); err != nil { + t.Fatal("conn Configure:", err) + } + if err := cl.DialUDP4(&conn, clPort, sv.Addr4(), svPort); err != nil { + t.Fatal("DialUDP4:", err) + } + + want := []byte("hello listener") + if _, err := conn.Write(want); err != nil { + t.Fatal("Write:", err) + } + if exchangeEthernetOnce(t, cl, sv, buf) == 0 { + t.Fatal("no packet sent by client") + } + + pc.SetReadDeadline(time.Now().Add(100 * time.Millisecond)) + var rbuf [testUDPBufSize]byte + n, senderAddr, err := pc.ReadFrom(rbuf[:]) + if err != nil { + t.Fatal("ReadFrom:", err) + } + if !bytes.Equal(rbuf[:n], want) { + t.Errorf("data: got %q, want %q", rbuf[:n], want) + } + wantSender := netip.AddrPortFrom(netip.AddrFrom4(cl.Addr4()), clPort) + if senderAddr != wantSender { + t.Errorf("sender addr: got %v, want %v", senderAddr, wantSender) + } +} + +// TestStackAsyncRegisterListenerUDP_ReplyToClient registers a PacketConn, receives +// a datagram, then calls WriteTo to send a reply back to the sender and verifies +// the client's udp.Conn reads it. +func TestStackAsyncRegisterListenerUDP_ReplyToClient(t *testing.T) { + const ( + svPort = 9002 + clPort = 9003 + ) + sv, cl := newUDPTestPair(t, 5678) + buf := make([]byte, ethernet.MaxMTU+ethernet.MaxOverheadSize) + + var pc udp.PacketConn + if err := pc.Configure(udp.PacketConnConfig{ + RxBuf: make([]byte, testUDPBufSize), TxBuf: make([]byte, testUDPBufSize), + RxQueueSize: testUDPQueueSize, TxQueueSize: testUDPQueueSize, + }); err != nil { + t.Fatal("pc Configure:", err) + } + if err := pc.Open(netip.AddrPortFrom(netip.AddrFrom4(sv.Addr4()), svPort)); err != nil { + t.Fatal("pc.Open:", err) + } + if err := sv.RegisterListenerUDP(&pc); err != nil { + t.Fatal("RegisterListenerUDP:", err) + } + + var conn udp.Conn + if err := conn.Configure(udp.ConnConfig{ + RxBuf: make([]byte, testUDPBufSize), TxBuf: make([]byte, testUDPBufSize), + RxQueueSize: testUDPQueueSize, TxQueueSize: testUDPQueueSize, + }); err != nil { + t.Fatal("conn Configure:", err) + } + if err := cl.DialUDP4(&conn, clPort, sv.Addr4(), svPort); err != nil { + t.Fatal("DialUDP4:", err) + } + + // Client → Server + if _, err := conn.Write([]byte("ping")); err != nil { + t.Fatal("Write:", err) + } + exchangeEthernetOnce(t, cl, sv, buf) + + pc.SetReadDeadline(time.Now().Add(100 * time.Millisecond)) + var rbuf [testUDPBufSize]byte + _, senderAddr, err := pc.ReadFrom(rbuf[:]) + if err != nil { + t.Fatal("ReadFrom:", err) + } + + // Server → Client + reply := []byte("pong") + pc.SetWriteDeadline(time.Now().Add(100 * time.Millisecond)) + if _, err := pc.WriteTo(reply, senderAddr); err != nil { + t.Fatal("WriteTo:", err) + } + if exchangeEthernetOnce(t, sv, cl, buf) == 0 { + t.Fatal("no reply packet from server") + } + + var rbuf2 [testUDPBufSize]byte + n, err := conn.Read(rbuf2[:]) + if err != nil { + t.Fatal("conn.Read:", err) + } + if !bytes.Equal(rbuf2[:n], reply) { + t.Errorf("reply data: got %q, want %q", rbuf2[:n], reply) + } +} + +// TestStackAsyncRegisterListenerUDP_MultiSource verifies that a single PacketConn +// registered via RegisterListenerUDP (no MAC filter) receives datagrams from two +// distinct clients and ReadFrom returns the correct sender address for each. +func TestStackAsyncRegisterListenerUDP_MultiSource(t *testing.T) { + const ( + svPort = 9004 + clPort = 9005 + ) + buf := make([]byte, ethernet.MaxMTU+ethernet.MaxOverheadSize) + + sv := new(StackAsync) + if err := sv.Reset(StackConfig{ + Hostname: "UDPSvMS", + RandSeed: 999, + StaticAddress4: [4]byte{10, 1, 0, 1}, + HardwareAddress: [6]byte{0xaa, 0xbb, 0, 0, 0, 1}, + MTU: ethernet.MaxMTU, + MaxActiveUDPPorts: 1, + }); err != nil { + t.Fatal("sv Reset:", err) + } + + cl1 := new(StackAsync) + cl1Addr := [4]byte{10, 1, 0, 2} + if err := cl1.Reset(StackConfig{ + Hostname: "UDPCl1", + RandSeed: 111, + StaticAddress4: cl1Addr, + HardwareAddress: [6]byte{0xaa, 0xbb, 0, 0, 0, 2}, + MTU: ethernet.MaxMTU, + MaxActiveUDPPorts: 1, + }); err != nil { + t.Fatal("cl1 Reset:", err) + } + + cl2 := new(StackAsync) + cl2Addr := [4]byte{10, 1, 0, 3} + if err := cl2.Reset(StackConfig{ + Hostname: "UDPCl2", + RandSeed: 222, + StaticAddress4: cl2Addr, + HardwareAddress: [6]byte{0xaa, 0xbb, 0, 0, 0, 3}, + MTU: ethernet.MaxMTU, + MaxActiveUDPPorts: 1, + }); err != nil { + t.Fatal("cl2 Reset:", err) + } + + cl1.SetGatewayHardwareAddr(sv.HardwareAddr()) + cl2.SetGatewayHardwareAddr(sv.HardwareAddr()) + + var pc udp.PacketConn + if err := pc.Configure(udp.PacketConnConfig{ + RxBuf: make([]byte, testUDPBufSize), TxBuf: make([]byte, testUDPBufSize), + RxQueueSize: testUDPQueueSize, TxQueueSize: testUDPQueueSize, + }); err != nil { + t.Fatal("pc Configure:", err) + } + if err := pc.Open(netip.AddrPortFrom(netip.AddrFrom4(sv.Addr4()), svPort)); err != nil { + t.Fatal("pc.Open:", err) + } + if err := sv.RegisterListenerUDP(&pc); err != nil { + t.Fatal("RegisterListenerUDP:", err) + } + + // Client 1 dials and sends. + var conn1 udp.Conn + if err := conn1.Configure(udp.ConnConfig{ + RxBuf: make([]byte, testUDPBufSize), TxBuf: make([]byte, testUDPBufSize), + RxQueueSize: testUDPQueueSize, TxQueueSize: testUDPQueueSize, + }); err != nil { + t.Fatal("conn1 Configure:", err) + } + if err := cl1.DialUDP4(&conn1, clPort, sv.Addr4(), svPort); err != nil { + t.Fatal("cl1 DialUDP4:", err) + } + msg1 := []byte("from-client-1") + if _, err := conn1.Write(msg1); err != nil { + t.Fatal("conn1 Write:", err) + } + exchangeEthernetOnce(t, cl1, sv, buf) + + // Client 2 dials and sends. + var conn2 udp.Conn + if err := conn2.Configure(udp.ConnConfig{ + RxBuf: make([]byte, testUDPBufSize), TxBuf: make([]byte, testUDPBufSize), + RxQueueSize: testUDPQueueSize, TxQueueSize: testUDPQueueSize, + }); err != nil { + t.Fatal("conn2 Configure:", err) + } + if err := cl2.DialUDP4(&conn2, clPort, sv.Addr4(), svPort); err != nil { + t.Fatal("cl2 DialUDP4:", err) + } + msg2 := []byte("from-client-2") + if _, err := conn2.Write(msg2); err != nil { + t.Fatal("conn2 Write:", err) + } + exchangeEthernetOnce(t, cl2, sv, buf) + + // Server reads both datagrams. + pc.SetReadDeadline(time.Now().Add(100 * time.Millisecond)) + type recv struct { + data []byte + from netip.AddrPort + } + var got [2]recv + var rbuf [testUDPBufSize]byte + for i := range 2 { + n, addr, err := pc.ReadFrom(rbuf[:]) + if err != nil { + t.Fatalf("ReadFrom[%d]: %v", i, err) + } + got[i] = recv{data: bytes.Clone(rbuf[:n]), from: addr} + } + + wantFrom1 := netip.AddrPortFrom(netip.AddrFrom4(cl1Addr), clPort) + wantFrom2 := netip.AddrPortFrom(netip.AddrFrom4(cl2Addr), clPort) + + // Datagrams may arrive in either order. + if got[0].from == wantFrom1 { + if !bytes.Equal(got[0].data, msg1) { + t.Errorf("[0] data: got %q, want %q", got[0].data, msg1) + } + if got[1].from != wantFrom2 || !bytes.Equal(got[1].data, msg2) { + t.Errorf("[1]: got addr=%v data=%q, want addr=%v data=%q", got[1].from, got[1].data, wantFrom2, msg2) + } + } else { + if got[0].from != wantFrom2 || !bytes.Equal(got[0].data, msg2) { + t.Errorf("[0]: got addr=%v data=%q, want addr=%v data=%q", got[0].from, got[0].data, wantFrom2, msg2) + } + if got[1].from != wantFrom1 || !bytes.Equal(got[1].data, msg1) { + t.Errorf("[1]: got addr=%v data=%q, want addr=%v data=%q", got[1].from, got[1].data, wantFrom1, msg1) + } + } +}