diff --git a/go.mod b/go.mod index 7302092..45317b3 100644 --- a/go.mod +++ b/go.mod @@ -39,6 +39,7 @@ require ( ) require ( + github.com/anacrolix/mmsg v1.1.1 // indirect github.com/beorn7/perks v1.0.1 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/davecgh/go-spew v1.1.1 // indirect diff --git a/go.sum b/go.sum index 030d6ef..3346e3e 100644 --- a/go.sum +++ b/go.sum @@ -6,6 +6,8 @@ github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuy github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0= github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0= github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho= +github.com/anacrolix/mmsg v1.1.1 h1:4ce/3I5kM7qSF6T5A8MOmDCfac3UqYlk5Bzh5XsWebM= +github.com/anacrolix/mmsg v1.1.1/go.mod h1:lPCXEN1eDDQtKktdKEzdw+roswx6wWPpeXAl/WpWVDU= github.com/anmitsu/go-shlex v0.0.0-20200514113438-38f4b401e2be h1:9AeTilPcZAjCFIImctFaOjnTIavg87rW78vTPkQqLI8= github.com/anmitsu/go-shlex v0.0.0-20200514113438-38f4b401e2be/go.mod h1:ySMOLuWl6zY27l47sB3qLNK6tF2fkHG55UZxx8oIVo4= github.com/armon/go-radix v1.0.0 h1:F4z6KzEeeQIMeLFa97iZU6vupzoecKdU5TX24SNppXI= diff --git a/udp/udp_linux.go b/udp/udp_linux.go index 2a747e2..458fe31 100644 --- a/udp/udp_linux.go +++ b/udp/udp_linux.go @@ -5,7 +5,6 @@ package udp import ( "context" - "encoding/binary" "fmt" "net" "net/netip" @@ -16,11 +15,13 @@ import ( "github.com/rcrowley/go-metrics" "github.com/sirupsen/logrus" "github.com/slackhq/nebula/config" + "golang.org/x/net/ipv6" "golang.org/x/sys/unix" ) type StdConn struct { - c *net.UDPConn + c *ipv6.PacketConn + uc *net.UDPConn rc syscall.RawConn isV4 bool l *logrus.Logger @@ -65,8 +66,7 @@ func NewListener(l *logrus.Logger, ip netip.Addr, port int, multi bool, batch in _ = c.Close() return nil, fmt.Errorf("unable to open sysfd: %w", err) } - - return &StdConn{c: uc, rc: rc, isV4: ip.Is4(), l: l, batch: batch}, err + return &StdConn{c: ipv6.NewPacketConn(c), rc: rc, uc: uc, isV4: ip.Is4(), l: l, batch: batch}, err } func (u *StdConn) Rebind() error { @@ -143,36 +143,48 @@ func (u *StdConn) GetSoMark() (int, error) { } func (u *StdConn) LocalAddr() (netip.AddrPort, error) { - sa := u.c.LocalAddr() + sa := u.uc.LocalAddr() return netip.ParseAddrPort(sa.String()) } func (u *StdConn) ListenOut(r EncReader) { var ip netip.Addr + var port int - u.msgs, u.buffers, u.names = u.PrepareRawMessages(u.batch) - read := u.ReadMulti - if u.batch == 1 { - read = u.ReadSingle + //u.msgs, u.buffers, u.names = u.PrepareRawMessages(u.batch) + //read := u.ReadMulti + //if u.batch == 1 { + // read = u.ReadSingle + //} + + var err error + var n int + msgs := make([]ipv6.Message, u.batch) + for i := range msgs { + msgs[i].Buffers = [][]byte{make([]byte, MTU)} } for { - read() - if u.err != nil { + //read() + n, err = u.c.ReadBatch(msgs, 0) + if err != nil { //TODO: remove logging, return error - u.l.WithError(u.err).Error("udp socket is closed, exiting read loop") + u.l.WithError(err).Error("udp socket is closed, exiting read loop") return } - for i := 0; i < int(u.n); i++ { - // Its ok to skip the ok check here, the slicing is the only error that can occur and it will panic - if u.isV4 { - ip, _ = netip.AddrFromSlice(u.names[i][4:8]) - } else { - ip, _ = netip.AddrFromSlice(u.names[i][8:24]) + for i := 0; i < n; i++ { + switch addr := msgs[i].Addr.(type) { + case *net.UDPAddr: + // Its ok to skip the ok check here, the slicing is the only error that can occur and it will panic + ip, _ = netip.AddrFromSlice(addr.IP) + port = addr.Port + default: + //TODO: this is an error, return? } + //u.l.Error("GOT A PACKET", msgs[i].Len) - r(netip.AddrPortFrom(ip.Unmap(), binary.BigEndian.Uint16(u.names[i][2:4])), u.buffers[i][:u.msgs[i].Len]) + r(netip.AddrPortFrom(ip.Unmap(), uint16(port)), msgs[i].Buffers[0][:msgs[i].N]) } } } @@ -245,7 +257,7 @@ func (u *StdConn) innerReadMulti(fd uintptr) bool { } func (u *StdConn) WriteTo(b []byte, ip netip.AddrPort) error { - _, err := u.c.WriteToUDPAddrPort(b, ip) + _, err := u.uc.WriteToUDPAddrPort(b, ip) return err } @@ -318,7 +330,7 @@ func (u *StdConn) getMemInfo(meminfo *[unix.SK_MEMINFO_VARS]uint32) error { } func (u *StdConn) Close() error { - err := u.c.Close() + err := u.uc.Close() return err }