From f33d034fecd7723a1e560ccc62aeeba328454fd0 Mon Sep 17 00:00:00 2001 From: Bhasker Hariharan Date: Mon, 7 Feb 2022 19:49:17 -0800 Subject: [PATCH] Fix bug in fdbased when used with TAP/TUN fd. TAP/TUN fd's are not sockets and using the WritePackets calls results in errors as it always defaults to using SendMMsg which is not supported for tap/tun device fds. This CL changes WritePackets to gracefully degrade to using writev instead of sendmmsg if the underlying fd is not a socket. Fixes #7125 PiperOrigin-RevId: 427083193 --- pkg/tcpip/link/fdbased/endpoint.go | 58 +++++++++++++++++-------- pkg/tcpip/link/muxed/injectable_test.go | 5 ++- 2 files changed, 45 insertions(+), 18 deletions(-) diff --git a/pkg/tcpip/link/fdbased/endpoint.go b/pkg/tcpip/link/fdbased/endpoint.go index 60348afe4..a683bab99 100644 --- a/pkg/tcpip/link/fdbased/endpoint.go +++ b/pkg/tcpip/link/fdbased/endpoint.go @@ -100,11 +100,16 @@ func (p PacketDispatchMode) String() string { var _ stack.LinkEndpoint = (*endpoint)(nil) var _ stack.GSOEndpoint = (*endpoint)(nil) +type fdInfo struct { + fd int + isSocket bool +} + type endpoint struct { // fds is the set of file descriptors each identifying one inbound/outbound // channel. The endpoint will dispatch from all inbound channels as well as // hash outbound packets to specific channels based on the packet hash. - fds []int + fds []fdInfo // mtu (maximum transmission unit) is the maximum size of a packet. mtu uint32 @@ -260,7 +265,6 @@ func New(opts *Options) (stack.LinkEndpoint, error) { } e := &endpoint{ - fds: opts.FDs, mtu: opts.MTU, caps: caps, closed: opts.ClosedFunc, @@ -281,8 +285,7 @@ func New(opts *Options) (stack.LinkEndpoint, error) { fid := atomic.AddInt32(&fanoutID, 1) // Create per channel dispatchers. - for i := 0; i < len(e.fds); i++ { - fd := e.fds[i] + for _, fd := range opts.FDs { if err := unix.SetNonblock(fd, true); err != nil { return nil, fmt.Errorf("unix.SetNonblock(%v) failed: %v", fd, err) } @@ -291,6 +294,7 @@ func New(opts *Options) (stack.LinkEndpoint, error) { if err != nil { return nil, err } + e.fds = append(e.fds, fdInfo{fd: fd, isSocket: isSocket}) if isSocket { if opts.GSOMaxSize != 0 { if opts.SoftwareGSOEnabled { @@ -501,7 +505,8 @@ func (e *endpoint) AddHeader(pkt *stack.PacketBuffer) { // writePacket writes outbound packets to the file descriptor. If it is not // currently writable, the packet is dropped. func (e *endpoint) writePacket(pkt *stack.PacketBuffer) tcpip.Error { - fd := e.fds[pkt.Hash%uint32(len(e.fds))] + fdInfo := e.fds[pkt.Hash%uint32(len(e.fds))] + fd := fdInfo.fd var vnetHdrBuf []byte if e.gsoKind == stack.HWGSOSupported { vnetHdr := virtioNetHdr{} @@ -549,8 +554,23 @@ func (e *endpoint) writePacket(pkt *stack.PacketBuffer) tcpip.Error { return rawfile.NonBlockingWriteIovec(fd, iovecs) } -func (e *endpoint) sendBatch(batchFD int, pkts []*stack.PacketBuffer) (int, tcpip.Error) { +func (e *endpoint) sendBatch(batchFDInfo fdInfo, pkts []*stack.PacketBuffer) (int, tcpip.Error) { + // Degrade to writePacket if underlying fd is not a socket. + if !batchFDInfo.isSocket { + written := 0 + for i := 0; i < len(pkts); i++ { + if err := e.writePacket(pkts[i]); err != nil { + if written > 0 { + return written, nil + } + return 0, err + } + } + return written, nil + } + // Send a batch of packets through batchFD. + batchFD := batchFDInfo.fd mmsgHdrsStorage := make([]rawfile.MMsgHdr, 0, len(pkts)) packets := 0 for packets < len(pkts) { @@ -657,29 +677,29 @@ func (e *endpoint) WritePackets(pkts stack.PacketBufferList) (int, tcpip.Error) // byte segment. const batchSz = 47 batch := make([]*stack.PacketBuffer, 0, batchSz) - batchFD := -1 + batchFDInfo := fdInfo{fd: -1, isSocket: false} sentPackets := 0 for pkt := pkts.Front(); pkt != nil; pkt = pkt.Next() { if len(batch) == 0 { - batchFD = e.fds[pkt.Hash%uint32(len(e.fds))] + batchFDInfo = e.fds[pkt.Hash%uint32(len(e.fds))] } - pktFD := e.fds[pkt.Hash%uint32(len(e.fds))] - if sendNow := pktFD != batchFD; !sendNow { + pktFDInfo := e.fds[pkt.Hash%uint32(len(e.fds))] + if sendNow := pktFDInfo != batchFDInfo; !sendNow { batch = append(batch, pkt) continue } - n, err := e.sendBatch(batchFD, batch) + n, err := e.sendBatch(batchFDInfo, batch) sentPackets += n if err != nil { return sentPackets, err } batch = batch[:0] batch = append(batch, pkt) - batchFD = pktFD + batchFDInfo = pktFDInfo } if len(batch) != 0 { - n, err := e.sendBatch(batchFD, batch) + n, err := e.sendBatch(batchFDInfo, batch) sentPackets += n if err != nil { return sentPackets, err @@ -695,7 +715,7 @@ func viewsEqual(vs1, vs2 []buffer.View) bool { // InjectOutobund implements stack.InjectableEndpoint.InjectOutbound. func (e *endpoint) InjectOutbound(dest tcpip.Address, packet []byte) tcpip.Error { - return rawfile.NonBlockingWrite(e.fds[0], packet) + return rawfile.NonBlockingWrite(e.fds[0].fd, packet) } // dispatchLoop reads packets from the file descriptor in a loop and dispatches @@ -750,13 +770,17 @@ func (e *InjectableEndpoint) InjectInbound(protocol tcpip.NetworkProtocolNumber, } // NewInjectable creates a new fd-based InjectableEndpoint. -func NewInjectable(fd int, mtu uint32, capabilities stack.LinkEndpointCapabilities) *InjectableEndpoint { +func NewInjectable(fd int, mtu uint32, capabilities stack.LinkEndpointCapabilities) (*InjectableEndpoint, error) { unix.SetNonblock(fd, true) + isSocket, err := isSocketFD(fd) + if err != nil { + return nil, err + } return &InjectableEndpoint{endpoint: endpoint{ - fds: []int{fd}, + fds: []fdInfo{{fd: fd, isSocket: isSocket}}, mtu: mtu, caps: capabilities, writevMaxIovs: rawfile.MaxIovs, - }} + }}, nil } diff --git a/pkg/tcpip/link/muxed/injectable_test.go b/pkg/tcpip/link/muxed/injectable_test.go index ac2e49c0d..0cb66d2ab 100644 --- a/pkg/tcpip/link/muxed/injectable_test.go +++ b/pkg/tcpip/link/muxed/injectable_test.go @@ -107,7 +107,10 @@ func makeTestInjectableEndpoint(t *testing.T) (*InjectableEndpoint, *os.File, tc if err != nil { t.Fatal("Failed to create socket pair:", err) } - underlyingEndpoint := fdbased.NewInjectable(pair[1], 6500, stack.CapabilityNone) + underlyingEndpoint, err := fdbased.NewInjectable(pair[1], 6500, stack.CapabilityNone) + if err != nil { + t.Fatalf("fdbased.NewInjectable(%d, 6500, stack.CapabilityNone) failed: %s", pair[1], err) + } routes := map[tcpip.Address]stack.InjectableLinkEndpoint{dstIP: underlyingEndpoint} endpoint := NewInjectableEndpoint(routes) return endpoint, os.NewFile(uintptr(pair[0]), "test route end"), dstIP