From f67e10cf3866fe3a809c217ae0f47a9c47854ce1 Mon Sep 17 00:00:00 2001 From: Nicolas Lacasse Date: Thu, 2 May 2024 20:58:46 -0700 Subject: [PATCH] Automated rollback of changelist 629815554 PiperOrigin-RevId: 630263974 --- pkg/tcpip/header/ipv4.go | 12 - pkg/tcpip/header/ipv6.go | 12 - pkg/tcpip/link/fdbased/BUILD | 4 - pkg/tcpip/link/fdbased/endpoint.go | 8 +- pkg/tcpip/link/fdbased/endpoint_test.go | 6 +- pkg/tcpip/link/fdbased/mmap.go | 71 ++--- pkg/tcpip/link/fdbased/mmap_stub.go | 2 +- pkg/tcpip/link/fdbased/mmap_unsafe.go | 4 +- pkg/tcpip/link/fdbased/packet_dispatchers.go | 76 ++++-- pkg/tcpip/link/fdbased/processors.go | 263 ------------------- pkg/tcpip/stack/packet_buffer_list.go | 12 - runsc/config/flags.go | 2 +- 12 files changed, 102 insertions(+), 370 deletions(-) delete mode 100644 pkg/tcpip/link/fdbased/processors.go diff --git a/pkg/tcpip/header/ipv4.go b/pkg/tcpip/header/ipv4.go index 84a5b1d77..3168f1f7a 100644 --- a/pkg/tcpip/header/ipv4.go +++ b/pkg/tcpip/header/ipv4.go @@ -346,18 +346,6 @@ func (b IPv4) DestinationAddress() tcpip.Address { return tcpip.AddrFrom4([4]byte(b[dstAddr : dstAddr+IPv4AddressSize])) } -// SourceAddressSlice returns the "source address" field of the IPv4 header as a -// byte slice. -func (b IPv4) SourceAddressSlice() []byte { - return []byte(b[srcAddr : srcAddr+IPv4AddressSize]) -} - -// DestinationAddressSlice returns the "destination address" field of the IPv4 -// header as a byte slice. -func (b IPv4) DestinationAddressSlice() []byte { - return []byte(b[dstAddr : dstAddr+IPv4AddressSize]) -} - // SetSourceAddressWithChecksumUpdate implements ChecksummableNetwork. func (b IPv4) SetSourceAddressWithChecksumUpdate(new tcpip.Address) { b.SetChecksum(^checksumUpdate2ByteAlignedAddress(^b.Checksum(), b.SourceAddress(), new)) diff --git a/pkg/tcpip/header/ipv6.go b/pkg/tcpip/header/ipv6.go index 4260095c6..ed30f77b3 100644 --- a/pkg/tcpip/header/ipv6.go +++ b/pkg/tcpip/header/ipv6.go @@ -225,18 +225,6 @@ func (b IPv6) DestinationAddress() tcpip.Address { return tcpip.AddrFrom16([16]byte(b[v6DstAddr:][:IPv6AddressSize])) } -// SourceAddressSlice returns the "source address" field of the ipv6 header as a -// byte slice. -func (b IPv6) SourceAddressSlice() []byte { - return []byte(b[v6SrcAddr:][:IPv6AddressSize]) -} - -// DestinationAddressSlice returns the "destination address" field of the ipv6 -// header as a byte slice. -func (b IPv6) DestinationAddressSlice() []byte { - return []byte(b[v6DstAddr:][:IPv6AddressSize]) -} - // Checksum implements Network.Checksum. Given that IPv6 doesn't have a // checksum, it just returns 0. func (IPv6) Checksum() uint16 { diff --git a/pkg/tcpip/link/fdbased/BUILD b/pkg/tcpip/link/fdbased/BUILD index 8bb1ca63d..301a79874 100644 --- a/pkg/tcpip/link/fdbased/BUILD +++ b/pkg/tcpip/link/fdbased/BUILD @@ -14,17 +14,13 @@ go_library( "mmap_stub.go", "mmap_unsafe.go", "packet_dispatchers.go", - "processors.go", ], visibility = ["//visibility:public"], deps = [ "//pkg/atomicbitops", "//pkg/buffer", - "//pkg/rand", - "//pkg/sleep", "//pkg/sync", "//pkg/tcpip", - "//pkg/tcpip/hash/jenkins", "//pkg/tcpip/header", "//pkg/tcpip/link/rawfile", "//pkg/tcpip/link/stopfd", diff --git a/pkg/tcpip/link/fdbased/endpoint.go b/pkg/tcpip/link/fdbased/endpoint.go index bff2d0475..fa64a703b 100644 --- a/pkg/tcpip/link/fdbased/endpoint.go +++ b/pkg/tcpip/link/fdbased/endpoint.go @@ -42,7 +42,6 @@ package fdbased import ( "fmt" - "runtime" "golang.org/x/sys/unix" "gvisor.dev/gvisor/pkg/atomicbitops" @@ -323,9 +322,6 @@ func New(opts *Options) (stack.LinkEndpoint, error) { e.gsoMaxSize = opts.GSOMaxSize } } - if opts.ProcessorsPerChannel == 0 { - opts.ProcessorsPerChannel = max(1, runtime.GOMAXPROCS(0)/len(opts.FDs)) - } inboundDispatcher, err := createInboundDispatcher(e, fd, isSocket, fid, opts) if err != nil { @@ -340,7 +336,7 @@ func New(opts *Options) (stack.LinkEndpoint, error) { func createInboundDispatcher(e *endpoint, fd int, isSocket bool, fID int32, opts *Options) (linkDispatcher, error) { // By default use the readv() dispatcher as it works with all kinds of // FDs (tap/tun/unix domain sockets and af_packet). - inboundDispatcher, err := newReadVDispatcher(fd, e, opts) + inboundDispatcher, err := newReadVDispatcher(fd, e) if err != nil { return nil, fmt.Errorf("newReadVDispatcher(%d, %+v) = %v", fd, e, err) } @@ -380,7 +376,7 @@ func createInboundDispatcher(e *endpoint, fd int, isSocket bool, fID int32, opts switch e.packetDispatchMode { case PacketMMap: - inboundDispatcher, err = newPacketMMapDispatcher(fd, e, opts) + inboundDispatcher, err = newPacketMMapDispatcher(fd, e) if err != nil { return nil, fmt.Errorf("newPacketMMapDispatcher(%d, %+v) = %v", fd, e, err) } diff --git a/pkg/tcpip/link/fdbased/endpoint_test.go b/pkg/tcpip/link/fdbased/endpoint_test.go index 1bbb01953..353bf4c29 100644 --- a/pkg/tcpip/link/fdbased/endpoint_test.go +++ b/pkg/tcpip/link/fdbased/endpoint_test.go @@ -582,7 +582,7 @@ func (*fakeNetworkDispatcher) DeliverLinkPacket(tcpip.NetworkProtocolNumber, *st func TestDispatchPacketFormat(t *testing.T) { for _, test := range []struct { name string - newDispatcher func(fd int, e *endpoint, opts *Options) (linkDispatcher, error) + newDispatcher func(fd int, e *endpoint) (linkDispatcher, error) }{ { name: "readVDispatcher", @@ -590,7 +590,7 @@ func TestDispatchPacketFormat(t *testing.T) { }, { name: "recvMMsgDispatcher", - newDispatcher: newRecvMMsgDispatcher, + newDispatcher: func(fd int, e *endpoint) (linkDispatcher, error) { return newRecvMMsgDispatcher(fd, e, &Options{}) }, }, } { t.Run(test.name, func(t *testing.T) { @@ -618,7 +618,7 @@ func TestDispatchPacketFormat(t *testing.T) { d, err := test.newDispatcher(fds[0], &endpoint{ hdrSize: header.EthernetMinimumSize, dispatcher: sink, - }, &Options{ProcessorsPerChannel: 1}) + }) if err != nil { t.Fatal(err) } diff --git a/pkg/tcpip/link/fdbased/mmap.go b/pkg/tcpip/link/fdbased/mmap.go index 6473e95a3..a136a9fdd 100644 --- a/pkg/tcpip/link/fdbased/mmap.go +++ b/pkg/tcpip/link/fdbased/mmap.go @@ -130,17 +130,11 @@ type packetMMapDispatcher struct { // ringOffset is the current offset into the ring buffer where the next // inbound packet will be placed by the kernel. ringOffset int - - // mgr is the processor goroutine manager. - mgr *processorManager } -func (d *packetMMapDispatcher) release() { - d.mgr.close() -} +func (*packetMMapDispatcher) release() {} -func (d *packetMMapDispatcher) readMMappedPackets() (stack.PacketBufferList, bool, tcpip.Error) { - var pkts stack.PacketBufferList +func (d *packetMMapDispatcher) readMMappedPacket() (*buffer.View, bool, tcpip.Error) { hdr := tPacketHdr(d.ringBuffer[d.ringOffset*tpFrameSize:]) for hdr.tpStatus()&tpStatusUser == 0 { stopped, errno := rawfile.BlockingPollUntilStopped(d.EFD, d.fd, unix.POLLIN|unix.POLLERR) @@ -148,10 +142,10 @@ func (d *packetMMapDispatcher) readMMappedPackets() (stack.PacketBufferList, boo if errno == unix.EINTR { continue } - return pkts, stopped, rawfile.TranslateErrno(errno) + return nil, stopped, rawfile.TranslateErrno(errno) } if stopped { - return pkts, true, nil + return nil, true, nil } if hdr.tpStatus()&tpStatusCopy != 0 { // This frame is truncated so skip it after flipping the @@ -163,39 +157,50 @@ func (d *packetMMapDispatcher) readMMappedPackets() (stack.PacketBufferList, boo } } - for hdr.tpStatus()&tpStatusUser == 1 { - // Copy out the packet from the mmapped frame to a locally owned buffer. - pkts.PushBack(stack.NewPacketBuffer(stack.PacketBufferOptions{ - Payload: buffer.MakeWithView(buffer.NewViewWithData(hdr.Payload())), - })) - // Release packet to kernel. - hdr.setTPStatus(tpStatusKernel) - d.ringOffset = (d.ringOffset + 1) % tpFrameNR - hdr = tPacketHdr(d.ringBuffer[d.ringOffset*tpFrameSize:]) - } - return pkts, false, nil + // Copy out the packet from the mmapped frame to a locally owned buffer. + pkt := buffer.NewView(int(hdr.tpSnapLen())) + pkt.Write(hdr.Payload()) + // Release packet to kernel. + hdr.setTPStatus(tpStatusKernel) + d.ringOffset = (d.ringOffset + 1) % tpFrameNR + return pkt, false, nil } // dispatch reads packets from an mmaped ring buffer and dispatches them to the // network stack. func (d *packetMMapDispatcher) dispatch() (bool, tcpip.Error) { - pkts, stopped, err := d.readMMappedPackets() - defer pkts.Reset() + pkt, stopped, err := d.readMMappedPacket() if err != nil || stopped { return false, err } - for _, pkt := range pkts.AsSlice() { - if d.e.hdrSize > 0 { - hdr, ok := pkt.LinkHeader().Consume(d.e.hdrSize) - if !ok { - panic(fmt.Sprintf("LinkHeader().Consume(%d) must succeed", d.e.hdrSize)) - } - pkt.NetworkProtocolNumber = header.Ethernet(hdr).Type() + var p tcpip.NetworkProtocolNumber + if d.e.hdrSize > 0 { + p = header.Ethernet(pkt.AsSlice()).Type() + } else { + // We don't get any indication of what the packet is, so try to guess + // if it's an IPv4 or IPv6 packet. + switch header.IPVersion(pkt.AsSlice()) { + case header.IPv4Version: + p = header.IPv4ProtocolNumber + case header.IPv6Version: + p = header.IPv6ProtocolNumber + default: + return true, nil } - d.mgr.queuePacket(pkt, d.e.hdrSize > 0) } - if pkts.Len() > 0 { - d.mgr.wakeReady() + + pbuf := stack.NewPacketBuffer(stack.PacketBufferOptions{ + Payload: buffer.MakeWithView(pkt), + }) + defer pbuf.DecRef() + if d.e.hdrSize > 0 { + if _, ok := pbuf.LinkHeader().Consume(d.e.hdrSize); !ok { + panic(fmt.Sprintf("LinkHeader().Consume(%d) must succeed", d.e.hdrSize)) + } } + d.e.mu.RLock() + dsp := d.e.dispatcher + d.e.mu.RUnlock() + dsp.DeliverNetworkPacket(p, pbuf) return true, nil } diff --git a/pkg/tcpip/link/fdbased/mmap_stub.go b/pkg/tcpip/link/fdbased/mmap_stub.go index c76c4b655..9d8679502 100644 --- a/pkg/tcpip/link/fdbased/mmap_stub.go +++ b/pkg/tcpip/link/fdbased/mmap_stub.go @@ -19,6 +19,6 @@ package fdbased // Stubbed out version for non-linux/non-amd64/non-arm64 platforms. -func newPacketMMapDispatcher(fd int, e *endpoint, opts *Options) (linkDispatcher, error) { +func newPacketMMapDispatcher(fd int, e *endpoint) (linkDispatcher, error) { return nil, nil } diff --git a/pkg/tcpip/link/fdbased/mmap_unsafe.go b/pkg/tcpip/link/fdbased/mmap_unsafe.go index c324d1b4b..abe07a5e1 100644 --- a/pkg/tcpip/link/fdbased/mmap_unsafe.go +++ b/pkg/tcpip/link/fdbased/mmap_unsafe.go @@ -47,7 +47,7 @@ func (t tPacketHdr) setTPStatus(status uint32) { (*atomicbitops.Uint32)(statusPtr).Store(status) } -func newPacketMMapDispatcher(fd int, e *endpoint, opts *Options) (linkDispatcher, error) { +func newPacketMMapDispatcher(fd int, e *endpoint) (linkDispatcher, error) { stopFD, err := stopfd.New() if err != nil { return nil, err @@ -77,8 +77,6 @@ func newPacketMMapDispatcher(fd int, e *endpoint, opts *Options) (linkDispatcher if err != nil { return nil, fmt.Errorf("unix.Mmap(...,0, %v, ...) failed = %v", sz, err) } - d.mgr = newProcessorManager(opts, e) - d.mgr.start() d.ringBuffer = buf return d, nil } diff --git a/pkg/tcpip/link/fdbased/packet_dispatchers.go b/pkg/tcpip/link/fdbased/packet_dispatchers.go index 95d19dd97..ed418ff3d 100644 --- a/pkg/tcpip/link/fdbased/packet_dispatchers.go +++ b/pkg/tcpip/link/fdbased/packet_dispatchers.go @@ -155,12 +155,9 @@ type readVDispatcher struct { // buf is the iovec buffer that contains the packet contents. buf *iovecBuffer - - // mgr is the processor goroutine manager. - mgr *processorManager } -func newReadVDispatcher(fd int, e *endpoint, opts *Options) (linkDispatcher, error) { +func newReadVDispatcher(fd int, e *endpoint) (linkDispatcher, error) { stopFD, err := stopfd.New() if err != nil { return nil, err @@ -172,14 +169,11 @@ func newReadVDispatcher(fd int, e *endpoint, opts *Options) (linkDispatcher, err } skipsVnetHdr := d.e.gsoKind == stack.HostGSOSupported d.buf = newIovecBuffer(BufConfig, skipsVnetHdr) - d.mgr = newProcessorManager(opts, e) - d.mgr.start() return d, nil } func (d *readVDispatcher) release() { d.buf.release() - d.mgr.close() } // dispatch reads one packet from the file descriptor and dispatches it. @@ -194,14 +188,35 @@ func (d *readVDispatcher) dispatch() (bool, tcpip.Error) { }) defer pkt.DecRef() + var p tcpip.NetworkProtocolNumber if d.e.hdrSize > 0 { if !d.e.parseHeader(pkt) { return false, nil } - pkt.NetworkProtocolNumber = header.Ethernet(pkt.LinkHeader().Slice()).Type() + p = header.Ethernet(pkt.LinkHeader().Slice()).Type() + } else { + // We don't get any indication of what the packet is, so try to guess + // if it's an IPv4 or IPv6 packet. + // IP version information is at the first octet, so pulling up 1 byte. + h, ok := pkt.Data().PullUp(1) + if !ok { + return true, nil + } + switch header.IPVersion(h) { + case header.IPv4Version: + p = header.IPv4ProtocolNumber + case header.IPv6Version: + p = header.IPv6ProtocolNumber + default: + return true, nil + } } - d.mgr.queuePacket(pkt, d.e.hdrSize > 0) - d.mgr.wakeReady() + + d.e.mu.RLock() + dsp := d.e.dispatcher + d.e.mu.RUnlock() + dsp.DeliverNetworkPacket(p, pkt) + return true, nil } @@ -229,9 +244,6 @@ type recvMMsgDispatcher struct { // gro coalesces incoming packets to increase throughput. gro gro.GRO - - // mgr is the processor goroutine manager. - mgr *processorManager } const ( @@ -257,8 +269,6 @@ func newRecvMMsgDispatcher(fd int, e *endpoint, opts *Options) (linkDispatcher, d.bufs[i] = newIovecBuffer(BufConfig, skipsVnetHdr) } d.gro.Init(opts.GRO) - d.mgr = newProcessorManager(opts, e) - d.mgr.start() return d, nil } @@ -267,7 +277,6 @@ func (d *recvMMsgDispatcher) release() { for _, iov := range d.bufs { iov.release() } - d.mgr.close() } // recvMMsgDispatch reads more than one packet at a time from the file @@ -309,17 +318,44 @@ func (d *recvMMsgDispatcher) dispatch() (bool, tcpip.Error) { // Mark that this iovec has been processed. d.msgHdrs[k].Msg.Iovlen = 0 + var p tcpip.NetworkProtocolNumber if d.e.hdrSize > 0 { hdr, ok := pkt.LinkHeader().Consume(d.e.hdrSize) if !ok { return false, nil } - pkt.NetworkProtocolNumber = header.Ethernet(hdr).Type() + p = header.Ethernet(hdr).Type() + } else { + // We don't get any indication of what the packet is, so try to guess + // if it's an IPv4 or IPv6 packet. + // IP version information is at the first octet, so pulling up 1 byte. + h, ok := pkt.Data().PullUp(1) + if !ok { + // Skip this packet. + continue + } + switch header.IPVersion(h) { + case header.IPv4Version: + p = header.IPv4ProtocolNumber + case header.IPv6Version: + p = header.IPv6ProtocolNumber + default: + // Skip this packet. + continue + } + } + + // Only use GRO if there's more than one packet. + if nMsgs > 1 { + pkt.NetworkProtocolNumber = p + pkt.RXChecksumValidated = d.e.caps&stack.CapabilityRXChecksumOffload != 0 + d.gro.Enqueue(pkt) + } else { + dsp.DeliverNetworkPacket(p, pkt) + return true, nil } - pkt.RXChecksumValidated = d.e.caps&stack.CapabilityRXChecksumOffload != 0 - d.mgr.queuePacket(pkt, d.e.hdrSize > 0) } - d.mgr.wakeReady() + d.gro.Flush() return true, nil } diff --git a/pkg/tcpip/link/fdbased/processors.go b/pkg/tcpip/link/fdbased/processors.go deleted file mode 100644 index 1bb7c298c..000000000 --- a/pkg/tcpip/link/fdbased/processors.go +++ /dev/null @@ -1,263 +0,0 @@ -// Copyright 2024 The gVisor Authors. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -//go:build linux -// +build linux - -package fdbased - -import ( - "encoding/binary" - - "gvisor.dev/gvisor/pkg/rand" - "gvisor.dev/gvisor/pkg/sleep" - "gvisor.dev/gvisor/pkg/sync" - "gvisor.dev/gvisor/pkg/tcpip" - "gvisor.dev/gvisor/pkg/tcpip/hash/jenkins" - "gvisor.dev/gvisor/pkg/tcpip/header" - "gvisor.dev/gvisor/pkg/tcpip/stack" - "gvisor.dev/gvisor/pkg/tcpip/stack/gro" -) - -type processor struct { - mu sync.Mutex - // +checklocks:mu - pkts stack.PacketBufferList - - e *endpoint - gro gro.GRO - sleeper sleep.Sleeper - packetWaker sleep.Waker - closeWaker sleep.Waker -} - -func (p *processor) start(wg *sync.WaitGroup) { - defer wg.Done() - defer p.sleeper.Done() - for { - switch w := p.sleeper.Fetch(true); { - case w == &p.packetWaker: - p.deliverPackets() - case w == &p.closeWaker: - p.mu.Lock() - p.pkts.Reset() - p.mu.Unlock() - return - } - } -} - -func (p *processor) deliverPackets() { - p.e.mu.RLock() - p.gro.Dispatcher = p.e.dispatcher - p.e.mu.RUnlock() - - p.mu.Lock() - for p.pkts.Len() > 0 { - pkt := p.pkts.PopFront() - p.mu.Unlock() - p.gro.Enqueue(pkt) - pkt.DecRef() - p.mu.Lock() - } - p.mu.Unlock() - p.gro.Flush() -} - -// processorManager handles starting, closing, and queuing packets on processor -// goroutines. -type processorManager struct { - processors []processor - seed uint32 - wg sync.WaitGroup - e *endpoint - ready []bool -} - -// newProcessorManager creates a new processor manager. -func newProcessorManager(opts *Options, e *endpoint) *processorManager { - m := &processorManager{} - m.seed = rand.Uint32() - m.ready = make([]bool, opts.ProcessorsPerChannel) - m.processors = make([]processor, opts.ProcessorsPerChannel) - m.e = e - m.wg.Add(opts.ProcessorsPerChannel) - - for i := range m.processors { - p := &m.processors[i] - p.sleeper.AddWaker(&p.packetWaker) - p.sleeper.AddWaker(&p.closeWaker) - p.gro.Init(opts.GRO) - p.e = e - } - - return m -} - -// start starts the processor goroutines if the processor manager is configured -// with more than one processor. -func (m *processorManager) start() { - for i := range m.processors { - p := &m.processors[i] - // Only start processor in a separate goroutine if we have multiple of them. - if len(m.processors) > 1 { - go p.start(&m.wg) - } - } -} - -func (m *processorManager) connectionHash(cid *connectionID) uint32 { - var payload [4]byte - binary.LittleEndian.PutUint16(payload[0:], cid.srcPort) - binary.LittleEndian.PutUint16(payload[2:], cid.dstPort) - - h := jenkins.Sum32(m.seed) - h.Write(payload[:]) - h.Write(cid.srcAddr) - h.Write(cid.dstAddr) - return h.Sum32() -} - -// queuePacket queues a packet to be delivered to the appropriate processor. -func (m *processorManager) queuePacket(pkt *stack.PacketBuffer, hasEthHeader bool) { - var pIdx int - if len(m.processors) > 1 { - cid, nonConnectionPkt := tcpipConnectionID(pkt) - if !hasEthHeader { - if nonConnectionPkt { - // If there's no eth header this should be a standard tcpip packet. If - // it isn't the packet is invalid so drop it. - return - } - pkt.NetworkProtocolNumber = cid.proto - } - // If the packet is not associated with an active connection, use the - // first processor. - if nonConnectionPkt { - pIdx = 0 - } else { - pIdx = int(m.connectionHash(&cid)) % len(m.processors) - } - } else { - pIdx = 0 - } - p := &m.processors[pIdx] - p.mu.Lock() - defer p.mu.Unlock() - pkt.IncRef() - p.pkts.PushBack(pkt) - m.ready[pIdx] = true -} - -type connectionID struct { - proto tcpip.NetworkProtocolNumber - srcAddr, dstAddr []byte - srcPort, dstPort uint16 -} - -// tcpipConnectionID returns a tcpip connection id tuple based on the data found -// in the packet. It returns true if the packet is not associated with an active -// connection (e.g ARP, NDP, etc). The method assumes link headers have already -// been processed if they were present. -func tcpipConnectionID(pkt *stack.PacketBuffer) (connectionID, bool) { - var cid connectionID - h, ok := pkt.Data().PullUp(1) - if !ok { - // Skip this packet. - return cid, true - } - - const tcpSrcDstPortLen = 4 - switch header.IPVersion(h) { - case header.IPv4Version: - hdrLen := header.IPv4(h).HeaderLength() - h, ok = pkt.Data().PullUp(int(hdrLen) + tcpSrcDstPortLen) - if !ok { - return cid, true - } - ipHdr := header.IPv4(h[:hdrLen]) - tcpHdr := header.TCP(h[hdrLen:][:tcpSrcDstPortLen]) - - cid.srcAddr = ipHdr.SourceAddressSlice() - cid.dstAddr = ipHdr.DestinationAddressSlice() - cid.srcPort = tcpHdr.SourcePort() - cid.dstPort = tcpHdr.DestinationPort() - cid.proto = header.IPv4ProtocolNumber - case header.IPv6Version: - h, ok = pkt.Data().PullUp(header.IPv6FixedHeaderSize + tcpSrcDstPortLen) - if !ok { - return cid, true - } - ipHdr := header.IPv6(h) - - var tcpHdr header.TCP - if tcpip.TransportProtocolNumber(ipHdr.NextHeader()) == header.TCPProtocolNumber { - tcpHdr = header.TCP(h[header.IPv6FixedHeaderSize:][:tcpSrcDstPortLen]) - } else { - // Slow path for IPv6 options :(. - dataBuf := pkt.Data().ToBuffer() - dataBuf.TrimFront(header.IPv6MinimumSize) - it := header.MakeIPv6PayloadIterator(header.IPv6ExtensionHeaderIdentifier(ipHdr.NextHeader()), dataBuf) - defer it.Release() - for { - _, done, err := it.Next() - if done || err != nil { - break - } - } - h, ok = pkt.Data().PullUp(int(it.HeaderOffset()) + tcpSrcDstPortLen) - if !ok { - return cid, true - } - tcpHdr = header.TCP(h[it.HeaderOffset():][:tcpSrcDstPortLen]) - } - cid.srcAddr = ipHdr.SourceAddressSlice() - cid.dstAddr = ipHdr.DestinationAddressSlice() - cid.srcPort = tcpHdr.SourcePort() - cid.dstPort = tcpHdr.DestinationPort() - cid.proto = header.IPv6ProtocolNumber - - default: - return cid, true - } - return cid, false -} - -func (m *processorManager) close() { - if len(m.processors) < 2 { - return - } - for i := range m.processors { - p := &m.processors[i] - p.closeWaker.Assert() - } -} - -// wakeReady wakes up all processors that have a packet queued. If there is only -// one processor, the method delivers the packet inline without waking a -// goroutine. -func (m *processorManager) wakeReady() { - for i, ready := range m.ready { - if !ready { - continue - } - p := &m.processors[i] - if len(m.processors) > 1 { - p.packetWaker.Assert() - } else { - p.deliverPackets() - } - m.ready[i] = false - } -} diff --git a/pkg/tcpip/stack/packet_buffer_list.go b/pkg/tcpip/stack/packet_buffer_list.go index 363059a9b..226b3e495 100644 --- a/pkg/tcpip/stack/packet_buffer_list.go +++ b/pkg/tcpip/stack/packet_buffer_list.go @@ -62,18 +62,6 @@ func (pl *PacketBufferList) PushBack(pb *PacketBuffer) { pl.pbs = append(pl.pbs, pb) } -// PopFront removes the first element in the list if it exists and returns it. -// -//go:nosplit -func (pl *PacketBufferList) PopFront() *PacketBuffer { - if len(pl.pbs) == 0 { - return nil - } - pkt := pl.pbs[0] - pl.pbs = pl.pbs[1:] - return pkt -} - // DecRef decreases the reference count on each PacketBuffer // stored in the list. // diff --git a/runsc/config/flags.go b/runsc/config/flags.go index 96d6bd1f3..53b4cb024 100644 --- a/runsc/config/flags.go +++ b/runsc/config/flags.go @@ -119,7 +119,7 @@ func RegisterFlags(flagSet *flag.FlagSet) { flagSet.Bool("rx-checksum-offload", true, "enable RX checksum offload.") flagSet.Var(queueingDisciplinePtr(QDiscFIFO), "qdisc", "specifies which queueing discipline to apply by default to the non loopback nics used by the sandbox.") flagSet.Int("num-network-channels", 1, "number of underlying channels(FDs) to use for network link endpoints.") - flagSet.Int("network-processors-per-channel", 0, "number of goroutines in each channel for processng inbound packets. If 0, the link endpoint will divide GOMAXPROCS evenly among the number of channels specified by num-network-channels.") + flagSet.Int("network-processors-per-channel", 1, "number of goroutines in each channel for processng inbound packets. If 0, the link endpoint will divide GOMAXPROCS evenly among the number of channels specified by num-network-channels.") flagSet.Bool("buffer-pooling", true, "DEPRECATED: this flag has no effect. Buffer pooling is always enabled.") flagSet.Var(&xdpConfig, "EXPERIMENTAL-xdp", `whether and how to use XDP. Can be one of: "off" (default), "ns", "redirect:", or "tunnel:"`) flagSet.Bool("EXPERIMENTAL-xdp-need-wakeup", true, "EXPERIMENTAL. Use XDP_USE_NEED_WAKEUP with XDP sockets.") // TODO(b/240191988): Figure out whether this helps and remove it as a flag.