Convert fdbased link endpoints to use pkg/buffer instead of VectorizedViews.

PiperOrigin-RevId: 447073885

pkg/buffer/view.go

@@ -33,6 +33,19 @@ type View struct {
 	pool pool
 }
 
+// NewWithData creates a new view initialized with given data.
+func NewWithData(b []byte) View {
+	v := View{
+		size: int64(len(b)),
+	}
+	if len(b) > 0 {
+		buf := v.pool.getNoInit()
+		buf.initWithData(b)
+		v.data.PushBack(buf)
+	}
+	return v
+}
+
 // TrimFront removes the first count bytes from the buffer.
 func (v *View) TrimFront(count int64) {
 	if count >= v.size {

pkg/tcpip/link/fdbased/BUILD

@@ -15,9 +15,9 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//pkg/atomicbitops",
+        "//pkg/buffer",
         "//pkg/sync",
         "//pkg/tcpip",
-        "//pkg/tcpip/buffer",
         "//pkg/tcpip/header",
         "//pkg/tcpip/link/rawfile",
         "//pkg/tcpip/stack",
@@ -31,10 +31,10 @@ go_test(
     srcs = ["endpoint_test.go"],
     library = ":fdbased",
     deps = [
+        "//pkg/buffer",
         "//pkg/refs",
         "//pkg/refsvfs2",
         "//pkg/tcpip",
-        "//pkg/tcpip/buffer",
         "//pkg/tcpip/header",
         "//pkg/tcpip/stack",
         "@com_github_google_go_cmp//cmp:go_default_library",

pkg/tcpip/link/fdbased/endpoint.go

@@ -45,9 +45,9 @@ import (
 
 	"golang.org/x/sys/unix"
 	"gvisor.dev/gvisor/pkg/atomicbitops"
+	"gvisor.dev/gvisor/pkg/buffer"
 	"gvisor.dev/gvisor/pkg/sync"
 	"gvisor.dev/gvisor/pkg/tcpip"
-	"gvisor.dev/gvisor/pkg/tcpip/buffer"
 	"gvisor.dev/gvisor/pkg/tcpip/header"
 	"gvisor.dev/gvisor/pkg/tcpip/link/rawfile"
 	"gvisor.dev/gvisor/pkg/tcpip/stack"

pkg/tcpip/link/fdbased/endpoint_test.go

@@ -29,10 +29,10 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	"golang.org/x/sys/unix"
+	"gvisor.dev/gvisor/pkg/buffer"
 	"gvisor.dev/gvisor/pkg/refs"
 	"gvisor.dev/gvisor/pkg/refsvfs2"
 	"gvisor.dev/gvisor/pkg/tcpip"
-	"gvisor.dev/gvisor/pkg/tcpip/buffer"
 	"gvisor.dev/gvisor/pkg/tcpip/header"
 	"gvisor.dev/gvisor/pkg/tcpip/stack"
 )
@@ -52,10 +52,10 @@ type packetInfo struct {
 }
 
 type packetContents struct {
-	LinkHeader      buffer.View
-	NetworkHeader   buffer.View
-	TransportHeader buffer.View
-	Data            buffer.View
+	LinkHeader      []byte
+	NetworkHeader   []byte
+	TransportHeader []byte
+	Data            []byte
 }
 
 func checkPacketInfoEqual(t *testing.T, got, want packetInfo) {
@@ -186,7 +186,7 @@ func testWritePacket(t *testing.T, plen int, eth bool, gsoMaxSize uint32, hash u
 	defer c.cleanup()
 
 	// Build payload.
-	payload := buffer.NewView(plen)
+	payload := make([]byte, plen)
 	if _, err := rand.Read(payload); err != nil {
 		t.Fatalf("rand.Read(payload): %s", err)
 	}
@@ -195,7 +195,7 @@ func testWritePacket(t *testing.T, plen int, eth bool, gsoMaxSize uint32, hash u
 	const netHdrLen = 100
 	pkt := stack.NewPacketBuffer(stack.PacketBufferOptions{
 		ReserveHeaderBytes: int(c.ep.MaxHeaderLength()) + netHdrLen,
-		Data:               payload.ToVectorisedView(),
+		Payload:            buffer.NewWithData(payload),
 	})
 	pkt.Hash = hash
 	// Every PacketBuffer must have these set:
@@ -212,7 +212,7 @@ func testWritePacket(t *testing.T, plen int, eth bool, gsoMaxSize uint32, hash u
 	}
 
 	// Write.
-	want := append(append(buffer.View(nil), b...), payload...)
+	want := append(append([]byte{}, b...), payload...)
 	const l3HdrLen = header.IPv6MinimumSize
 	if gsoMaxSize != 0 {
 		pkt.GSOOptions = stack.GSO{
@@ -336,7 +336,6 @@ func TestPreserveSrcAddress(t *testing.T) {
 		// the minimum size of the ethernet header.
 		// TODO(b/153685824): Figure out if this should use c.ep.MaxHeaderLength().
 		ReserveHeaderBytes: header.EthernetMinimumSize,
-		Data:               buffer.VectorisedView{},
 	})
 	defer pkt.DecRef()
 	// Every PacketBuffer must have these set:
@@ -387,7 +386,7 @@ func TestDeliverPacket(t *testing.T) {
 
 				wantPkt := stack.NewPacketBuffer(stack.PacketBufferOptions{
 					ReserveHeaderBytes: header.EthernetMinimumSize,
-					Data:               buffer.NewViewFromBytes(all).ToVectorisedView(),
+					Payload:            buffer.NewWithData(all),
 				})
 				defer wantPkt.DecRef()
 				if eth {
@@ -498,12 +497,12 @@ func TestIovecBuffer(t *testing.T) {
 			// later.
 			oldIovecs := append([]unix.Iovec(nil), iovecs...)
 
-			// Test the views that get pulled.
-			vv := b.pullViews(c.n)
+			// Test the buffer that get pulled.
+			buf := b.pullBuffer(c.n)
 			var lengths []int
-			for _, v := range vv.Views() {
+			buf.Apply(func(v []byte) {
 				lengths = append(lengths, len(v))
-			}
+			})
 			if !reflect.DeepEqual(lengths, c.wantLengths) {
 				t.Errorf("Pulled view lengths = %v, want %v", lengths, c.wantLengths)
 			}
@@ -550,11 +549,11 @@ func TestIovecBufferSkipVnetHdr(t *testing.T) {
 			b := newIovecBuffer([]int{10, 20, 50, 50}, true)
 			// Pretend a read happend.
 			b.nextIovecs()
-			vv := b.pullViews(test.readN)
-			if got, want := vv.Size(), test.wantLen; got != want {
+			buf := b.pullBuffer(test.readN)
+			if got, want := int(buf.Size()), test.wantLen; got != want {
 				t.Errorf("b.pullView(%d).Size() = %d; want %d", test.readN, got, want)
 			}
-			if got, want := len(vv.ToOwnedView()), test.wantLen; got != want {
+			if got, want := len(buf.Flatten()), test.wantLen; got != want {
 				t.Errorf("b.pullView(%d).ToOwnedView() has length %d; want %d", test.readN, got, want)
 			}
 		})

pkg/tcpip/link/fdbased/mmap.go

@@ -22,8 +22,8 @@ import (
 	"fmt"
 
 	"golang.org/x/sys/unix"
+	"gvisor.dev/gvisor/pkg/buffer"
 	"gvisor.dev/gvisor/pkg/tcpip"
-	"gvisor.dev/gvisor/pkg/tcpip/buffer"
 	"gvisor.dev/gvisor/pkg/tcpip/header"
 	"gvisor.dev/gvisor/pkg/tcpip/link/rawfile"
 	"gvisor.dev/gvisor/pkg/tcpip/stack"
@@ -186,7 +186,7 @@ func (d *packetMMapDispatcher) dispatch() (bool, tcpip.Error) {
 	}
 
 	pbuf := stack.NewPacketBuffer(stack.PacketBufferOptions{
-		Data: buffer.View(pkt).ToVectorisedView(),
+		Payload: buffer.NewWithData(pkt),
 	})
 	defer pbuf.DecRef()
 	if d.e.hdrSize > 0 {

pkg/tcpip/link/fdbased/packet_dispatchers.go

@@ -21,19 +21,21 @@ import (
 	"fmt"
 
 	"golang.org/x/sys/unix"
+	"gvisor.dev/gvisor/pkg/buffer"
 	"gvisor.dev/gvisor/pkg/tcpip"
-	"gvisor.dev/gvisor/pkg/tcpip/buffer"
 	"gvisor.dev/gvisor/pkg/tcpip/header"
 	"gvisor.dev/gvisor/pkg/tcpip/link/rawfile"
 	"gvisor.dev/gvisor/pkg/tcpip/stack"
 )
 
-// BufConfig defines the shape of the vectorised view used to read packets from the NIC.
+// BufConfig defines the shape of the buffer used to read packets from the NIC.
 var BufConfig = []int{128, 256, 256, 512, 1024, 2048, 4096, 8192, 16384, 32768}
 
 type iovecBuffer struct {
-	// views are the actual buffers that hold the packet contents.
-	views []buffer.View
+	// buffer is the actual buffer that holds the packet contents. Some contents
+	// are reused across calls to pullBuffer if number of requested bytes is
+	// smaller than the number of bytes allocated in the buffer.
+	buffer buffer.Buffer
 
 	// iovecs are initialized with base pointers/len of the corresponding
 	// entries in the views defined above, except when GSO is enabled
@@ -48,15 +50,22 @@ type iovecBuffer struct {
 
 	// skipsVnetHdr is true if virtioNetHdr is to skipped.
 	skipsVnetHdr bool
+
+	// pulledIndex is the index of the last []byte buffer pulled from the
+	// underlying buffer storage during a call to pullBuffers. It is -1
+	// if no buffer is pulled.
+	pulledIndex int
 }
 
 func newIovecBuffer(sizes []int, skipsVnetHdr bool) *iovecBuffer {
 	b := &iovecBuffer{
-		views:        make([]buffer.View, len(sizes)),
 		sizes:        sizes,
 		skipsVnetHdr: skipsVnetHdr,
+		// Setting pulledIndex to the length of sizes will allocate all
+		// the buffers.
+		pulledIndex: len(sizes),
 	}
-	niov := len(b.views)
+	niov := len(sizes)
 	if b.skipsVnetHdr {
 		niov++
 	}
@@ -75,45 +84,54 @@ func (b *iovecBuffer) nextIovecs() []unix.Iovec {
 		b.iovecs[0].SetLen(virtioNetHdrSize)
 		vnetHdrOff++
 	}
-	for i := range b.views {
-		if b.views[i] != nil {
+
+	var buf buffer.Buffer
+	for i, size := range b.sizes {
+		if i > b.pulledIndex {
 			break
 		}
-		v := buffer.NewView(b.sizes[i])
-		b.views[i] = v
+		v := make([]byte, size)
+		buf.AppendOwned(v)
 		b.iovecs[i+vnetHdrOff] = unix.Iovec{Base: &v[0]}
 		b.iovecs[i+vnetHdrOff].SetLen(len(v))
 	}
+	buf.Merge(&b.buffer)
+	b.buffer = buf
+	b.pulledIndex = -1
 	return b.iovecs
 }
 
-func (b *iovecBuffer) pullViews(n int) buffer.VectorisedView {
-	var views []buffer.View
+// pullBuffer extracts the enough underlying storage from b.buffer to hold n
+// bytes. It removes this storage from b.buffer, returns a new buffer
+// that holds the storage, and updates pulledIndex to indicate which part
+// of b.buffer's storage must be reallocated during the next call to
+// nextIovecs.
+func (b *iovecBuffer) pullBuffer(n int) buffer.Buffer {
+	var pulled buffer.Buffer
 	c := 0
 	if b.skipsVnetHdr {
-		c += virtioNetHdrSize
+		c = virtioNetHdrSize
 		if c >= n {
 			// Nothing in the packet.
-			return buffer.NewVectorisedView(0, nil)
+			return pulled
 		}
 	}
-	for i, v := range b.views {
-		c += len(v)
+	// Remove the used views from the buffer.
+	pulled = b.buffer.Clone()
+	for _, size := range b.sizes {
+		b.pulledIndex++
+		c += size
+		b.buffer.TrimFront(int64(size))
 		if c >= n {
-			b.views[i].CapLength(len(v) - (c - n))
-			views = append([]buffer.View(nil), b.views[:i+1]...)
 			break
 		}
 	}
-	// Remove the first len(views) used views from the state.
-	for i := range views {
-		b.views[i] = nil
-	}
 	if b.skipsVnetHdr {
 		// Exclude the size of the vnet header.
 		n -= virtioNetHdrSize
 	}
-	return buffer.NewVectorisedView(n, views)
+	pulled.Truncate(int64(n))
+	return pulled
 }
 
 // stopFd is an eventfd used to signal the stop of a dispatcher.
@@ -179,7 +197,7 @@ func (d *readVDispatcher) dispatch() (bool, tcpip.Error) {
 	}
 
 	pkt := stack.NewPacketBuffer(stack.PacketBufferOptions{
-		Data: d.buf.pullViews(n),
+		Payload: d.buf.pullBuffer(n),
 	})
 	defer pkt.DecRef()
 
@@ -285,7 +303,7 @@ func (d *recvMMsgDispatcher) dispatch() (bool, tcpip.Error) {
 	for k := 0; k < nMsgs; k++ {
 		n := int(d.msgHdrs[k].Len)
 		pkt := stack.NewPacketBuffer(stack.PacketBufferOptions{
-			Data: d.bufs[k].pullViews(n),
+			Payload: d.bufs[k].pullBuffer(n),
 		})
 		pkts.PushBack(pkt)
 

pkg/tcpip/stack/packet_buffer.go

@@ -41,15 +41,22 @@ var pkPool = sync.Pool{
 }
 
 // PacketBufferOptions specifies options for PacketBuffer creation.
+// TODO(b/230896518): Convert PacketBufferOptions.Data to be a buffer.Buffer
+// instead of a VectorisedView and remove Payload.
 type PacketBufferOptions struct {
 	// ReserveHeaderBytes is the number of bytes to reserve for headers. Total
 	// number of bytes pushed onto the headers must not exceed this value.
 	ReserveHeaderBytes int
 
 	// Data is the initial unparsed data for the new packet. If set, it will be
-	// owned by the new packet.
+	// owned by the new packet. If Data is set, Payload must be unset.
+	// Deprecated: Use Payload instead.
 	Data tcpipbuffer.VectorisedView
 
+	// Payload is the initial unparsed data for the new packet. If set, it will
+	// be owned by the new packet. If Payload is set, Data must be unset.
+	Payload buffer.Buffer
+
 	// IsForwardedPacket identifies that the PacketBuffer being created is for a
 	// forwarded packet.
 	IsForwardedPacket bool
@@ -181,6 +188,12 @@ func NewPacketBuffer(opts PacketBufferOptions) *PacketBuffer {
 		pk.buf.AppendOwned(make([]byte, opts.ReserveHeaderBytes))
 		pk.reserved = opts.ReserveHeaderBytes
 	}
+	if opts.Payload.Size() > 0 {
+		if len(opts.Data.Views()) != 0 {
+			panic("opts.Data must not be set if using Payload")
+		}
+		pk.buf.Merge(&opts.Payload)
+	}
 	for _, v := range opts.Data.Views() {
 		pk.buf.AppendOwned(v)
 	}