Migrate PacketBuffer to use pkg/buffer

Benchmark iperf3:
                    Before      After
native->runsc       5.14        5.01    (Gbps)
runsc->native       4.15        4.07    (Gbps)

It did introduce overhead, mainly at the bridge between pkg/buffer and
VectorisedView, the ExtractVV method. Once endpoints start migrating away from
VV, this overhead will be gone.

Updates #2404

PiperOrigin-RevId: 373651666
This commit is contained in:
Ting-Yu Wang
2021-05-13 13:56:16 -07:00
committed by gVisor bot
parent baa0888f11
commit 84f04cc858
12 changed files with 874 additions and 230 deletions
+1
View File
@@ -38,6 +38,7 @@ go_test(
name = "buffer_test",
size = "small",
srcs = [
"buffer_test.go",
"pool_test.go",
"safemem_test.go",
"view_test.go",
+28
View File
@@ -33,12 +33,40 @@ func (b *buffer) init(size int) {
b.data = make([]byte, size)
}
// initWithData initializes b with data, taking ownership.
func (b *buffer) initWithData(data []byte) {
b.data = data
b.read = 0
b.write = len(data)
}
// Reset resets read and write locations, effectively emptying the buffer.
func (b *buffer) Reset() {
b.read = 0
b.write = 0
}
// Remove removes r from the unread portion. It returns false if r does not
// fully reside in b.
func (b *buffer) Remove(r Range) bool {
sz := b.ReadSize()
switch {
case r.Len() != r.Intersect(Range{end: sz}).Len():
return false
case r.Len() == 0:
// Noop
case r.begin == 0:
b.read += r.end
case r.end == sz:
b.write -= r.Len()
default:
// Remove from the middle of b.data.
copy(b.data[b.read+r.begin:], b.data[b.read+r.end:b.write])
b.write -= r.Len()
}
return true
}
// Full indicates the buffer is full.
//
// This indicates there is no capacity left to write.
+111
View File
@@ -0,0 +1,111 @@
// Copyright 2021 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.
package buffer
import (
"bytes"
"testing"
)
func TestBufferRemove(t *testing.T) {
sample := []byte("01234567")
// Success cases
for _, tc := range []struct {
desc string
data []byte
rng Range
want []byte
}{
{
desc: "empty slice",
},
{
desc: "empty range",
data: sample,
want: sample,
},
{
desc: "empty range with positive begin",
data: sample,
rng: Range{begin: 1, end: 1},
want: sample,
},
{
desc: "range at beginning",
data: sample,
rng: Range{begin: 0, end: 1},
want: sample[1:],
},
{
desc: "range in middle",
data: sample,
rng: Range{begin: 2, end: 4},
want: []byte("014567"),
},
{
desc: "range at end",
data: sample,
rng: Range{begin: 7, end: 8},
want: sample[:7],
},
{
desc: "range all",
data: sample,
rng: Range{begin: 0, end: 8},
},
} {
t.Run(tc.desc, func(t *testing.T) {
var buf buffer
buf.initWithData(tc.data)
if ok := buf.Remove(tc.rng); !ok {
t.Errorf("buf.Remove(%#v) = false, want true", tc.rng)
} else if got := buf.ReadSlice(); !bytes.Equal(got, tc.want) {
t.Errorf("buf.ReadSlice() = %q, want %q", got, tc.want)
}
})
}
// Failure cases
for _, tc := range []struct {
desc string
data []byte
rng Range
}{
{
desc: "begin out-of-range",
data: sample,
rng: Range{begin: -1, end: 4},
},
{
desc: "end out-of-range",
data: sample,
rng: Range{begin: 4, end: 9},
},
{
desc: "both out-of-range",
data: sample,
rng: Range{begin: -100, end: 100},
},
} {
t.Run(tc.desc, func(t *testing.T) {
var buf buffer
buf.initWithData(tc.data)
if ok := buf.Remove(tc.rng); ok {
t.Errorf("buf.Remove(%#v) = true, want false", tc.rng)
}
})
}
}
+8 -1
View File
@@ -42,6 +42,13 @@ type pool struct {
// get gets a new buffer from p.
func (p *pool) get() *buffer {
buf := p.getNoInit()
buf.init(p.bufferSize)
return buf
}
// get gets a new buffer from p without initializing it.
func (p *pool) getNoInit() *buffer {
if p.avail == nil {
p.avail = p.embeddedStorage[:]
}
@@ -52,7 +59,6 @@ func (p *pool) get() *buffer {
p.bufferSize = defaultBufferSize
}
buf := &p.avail[0]
buf.init(p.bufferSize)
p.avail = p.avail[1:]
return buf
}
@@ -62,6 +68,7 @@ func (p *pool) put(buf *buffer) {
// Remove reference to the underlying storage, allowing it to be garbage
// collected.
buf.data = nil
buf.Reset()
}
// setBufferSize sets the size of underlying storage buffer for future
+177 -2
View File
@@ -19,6 +19,9 @@ import (
"io"
)
// Buffer is an alias to View.
type Buffer = View
// View is a non-linear buffer.
//
// All methods are thread compatible.
@@ -39,6 +42,51 @@ func (v *View) TrimFront(count int64) {
}
}
// Remove deletes data at specified location in v. It returns false if specified
// range does not fully reside in v.
func (v *View) Remove(offset, length int) bool {
if offset < 0 || length < 0 {
return false
}
tgt := Range{begin: offset, end: offset + length}
if tgt.Len() != tgt.Intersect(Range{end: int(v.size)}).Len() {
return false
}
// Scan through each buffer and remove intersections.
var curr Range
for buf := v.data.Front(); buf != nil; {
origLen := buf.ReadSize()
curr.end = curr.begin + origLen
if x := curr.Intersect(tgt); x.Len() > 0 {
if !buf.Remove(x.Offset(-curr.begin)) {
panic("buf.Remove() failed")
}
if buf.ReadSize() == 0 {
// buf fully removed, removing it from the list.
oldBuf := buf
buf = buf.Next()
v.data.Remove(oldBuf)
v.pool.put(oldBuf)
} else {
// Only partial data intersects, moving on to next one.
buf = buf.Next()
}
v.size -= int64(x.Len())
} else {
// This buffer is not in range, moving on to next one.
buf = buf.Next()
}
curr.begin += origLen
if curr.begin >= tgt.end {
break
}
}
return true
}
// ReadAt implements io.ReaderAt.ReadAt.
func (v *View) ReadAt(p []byte, offset int64) (int, error) {
var (
@@ -81,7 +129,6 @@ func (v *View) advanceRead(count int64) {
oldBuf := buf
buf = buf.Next() // Iterate.
v.data.Remove(oldBuf)
oldBuf.Reset()
v.pool.put(oldBuf)
// Update counts.
@@ -118,7 +165,6 @@ func (v *View) Truncate(length int64) {
// Drop the buffer completely; see above.
v.data.Remove(buf)
buf.Reset()
v.pool.put(buf)
v.size -= sz
}
@@ -224,6 +270,78 @@ func (v *View) Append(data []byte) {
}
}
// AppendOwned takes ownership of data and appends it to v.
func (v *View) AppendOwned(data []byte) {
if len(data) > 0 {
buf := v.pool.getNoInit()
buf.initWithData(data)
v.data.PushBack(buf)
v.size += int64(len(data))
}
}
// PullUp makes the specified range contiguous and returns the backing memory.
func (v *View) PullUp(offset, length int) ([]byte, bool) {
if length == 0 {
return nil, true
}
tgt := Range{begin: offset, end: offset + length}
if tgt.Intersect(Range{end: int(v.size)}).Len() != length {
return nil, false
}
curr := Range{}
buf := v.data.Front()
for ; buf != nil; buf = buf.Next() {
origLen := buf.ReadSize()
curr.end = curr.begin + origLen
if x := curr.Intersect(tgt); x.Len() == tgt.Len() {
// buf covers the whole requested target range.
sub := x.Offset(-curr.begin)
return buf.ReadSlice()[sub.begin:sub.end], true
} else if x.Len() > 0 {
// buf is pointing at the starting buffer we want to merge.
break
}
curr.begin += origLen
}
// Calculate the total merged length.
totLen := 0
for n := buf; n != nil; n = n.Next() {
totLen += n.ReadSize()
if curr.begin+totLen >= tgt.end {
break
}
}
// Merge the buffers.
data := make([]byte, totLen)
off := 0
for n := buf; n != nil && off < totLen; {
copy(data[off:], n.ReadSlice())
off += n.ReadSize()
// Remove buffers except for the first one, which will be reused.
if n == buf {
n = n.Next()
} else {
old := n
n = n.Next()
v.data.Remove(old)
v.pool.put(old)
}
}
// Update the first buffer with merged data.
buf.initWithData(data)
r := tgt.Offset(-curr.begin)
return buf.data[r.begin:r.end], true
}
// Flatten returns a flattened copy of this data.
//
// This method should not be used in any performance-sensitive paths. It may
@@ -267,6 +385,27 @@ func (v *View) Apply(fn func([]byte)) {
}
}
// SubApply applies fn to a given range of data in v. Any part of the range
// outside of v is ignored.
func (v *View) SubApply(offset, length int, fn func([]byte)) {
for buf := v.data.Front(); length > 0 && buf != nil; buf = buf.Next() {
d := buf.ReadSlice()
if offset >= len(d) {
offset -= len(d)
continue
}
if offset > 0 {
d = d[offset:]
offset = 0
}
if length < len(d) {
d = d[:length]
}
fn(d)
length -= len(d)
}
}
// Merge merges the provided View with this one.
//
// The other view will be appended to v, and other will be empty after this
@@ -389,3 +528,39 @@ func (v *View) ReadToWriter(w io.Writer, count int64) (int64, error) {
}
return done, err
}
// A Range specifies a range of buffer.
type Range struct {
begin int
end int
}
// Intersect returns the intersection of x and y.
func (x Range) Intersect(y Range) Range {
if x.begin < y.begin {
x.begin = y.begin
}
if x.end > y.end {
x.end = y.end
}
if x.begin >= x.end {
return Range{}
}
return x
}
// Offset returns x offset by off.
func (x Range) Offset(off int) Range {
x.begin += off
x.end += off
return x
}
// Len returns the length of x.
func (x Range) Len() int {
l := x.end - x.begin
if l < 0 {
l = 0
}
return l
}
+356
View File
@@ -17,7 +17,9 @@ package buffer
import (
"bytes"
"context"
"fmt"
"io"
"reflect"
"strings"
"testing"
@@ -237,6 +239,18 @@ func TestView(t *testing.T) {
},
},
// AppendOwned.
{
name: "append-owned",
input: "hello",
output: "hello world",
op: func(t *testing.T, v *View) {
b := []byte("Xworld")
v.AppendOwned(b)
b[0] = ' '
},
},
// Truncate.
{
name: "truncate",
@@ -495,6 +509,267 @@ func TestView(t *testing.T) {
}
}
func TestViewPullUp(t *testing.T) {
for _, tc := range []struct {
desc string
inputs []string
offset int
length int
output string
failed bool
// lengths is the lengths of each buffer node after the pull up.
lengths []int
}{
{
desc: "whole empty view",
},
{
desc: "zero pull",
inputs: []string{"hello", " world"},
lengths: []int{5, 6},
},
{
desc: "whole view",
inputs: []string{"hello", " world"},
offset: 0,
length: 11,
output: "hello world",
lengths: []int{11},
},
{
desc: "middle to end aligned",
inputs: []string{"0123", "45678", "9abcd"},
offset: 4,
length: 10,
output: "456789abcd",
lengths: []int{4, 10},
},
{
desc: "middle to end unaligned",
inputs: []string{"0123", "45678", "9abcd"},
offset: 6,
length: 8,
output: "6789abcd",
lengths: []int{4, 10},
},
{
desc: "middle aligned",
inputs: []string{"0123", "45678", "9abcd", "efgh"},
offset: 6,
length: 5,
output: "6789a",
lengths: []int{4, 10, 4},
},
// Failed cases.
{
desc: "empty view - length too long",
offset: 0,
length: 1,
failed: true,
},
{
desc: "empty view - offset too large",
offset: 1,
length: 1,
failed: true,
},
{
desc: "length too long",
inputs: []string{"0123", "45678", "9abcd"},
offset: 4,
length: 100,
failed: true,
lengths: []int{4, 5, 5},
},
{
desc: "offset too large",
inputs: []string{"0123", "45678", "9abcd"},
offset: 100,
length: 1,
failed: true,
lengths: []int{4, 5, 5},
},
} {
t.Run(tc.desc, func(t *testing.T) {
var v View
for _, s := range tc.inputs {
v.AppendOwned([]byte(s))
}
got, gotOk := v.PullUp(tc.offset, tc.length)
want, wantOk := []byte(tc.output), !tc.failed
if gotOk != wantOk || !bytes.Equal(got, want) {
t.Errorf("v.PullUp(%d, %d) = %q, %t; %q, %t", tc.offset, tc.length, got, gotOk, want, wantOk)
}
var gotLengths []int
for buf := v.data.Front(); buf != nil; buf = buf.Next() {
gotLengths = append(gotLengths, buf.ReadSize())
}
if !reflect.DeepEqual(gotLengths, tc.lengths) {
t.Errorf("lengths = %v; want %v", gotLengths, tc.lengths)
}
})
}
}
func TestViewRemove(t *testing.T) {
// Success cases
for _, tc := range []struct {
desc string
// before is the contents for each buffer node initially.
before []string
// after is the contents for each buffer node after removal.
after []string
offset int
length int
}{
{
desc: "empty view",
},
{
desc: "nothing removed",
before: []string{"hello", " world"},
after: []string{"hello", " world"},
},
{
desc: "whole view",
before: []string{"hello", " world"},
offset: 0,
length: 11,
},
{
desc: "beginning to middle aligned",
before: []string{"0123", "45678", "9abcd"},
after: []string{"9abcd"},
offset: 0,
length: 9,
},
{
desc: "beginning to middle unaligned",
before: []string{"0123", "45678", "9abcd"},
after: []string{"678", "9abcd"},
offset: 0,
length: 6,
},
{
desc: "middle to end aligned",
before: []string{"0123", "45678", "9abcd"},
after: []string{"0123"},
offset: 4,
length: 10,
},
{
desc: "middle to end unaligned",
before: []string{"0123", "45678", "9abcd"},
after: []string{"0123", "45"},
offset: 6,
length: 8,
},
{
desc: "middle aligned",
before: []string{"0123", "45678", "9abcd"},
after: []string{"0123", "9abcd"},
offset: 4,
length: 5,
},
{
desc: "middle unaligned",
before: []string{"0123", "45678", "9abcd"},
after: []string{"0123", "4578", "9abcd"},
offset: 6,
length: 1,
},
} {
t.Run(tc.desc, func(t *testing.T) {
var v View
for _, s := range tc.before {
v.AppendOwned([]byte(s))
}
if ok := v.Remove(tc.offset, tc.length); !ok {
t.Errorf("v.Remove(%d, %d) = false, want true", tc.offset, tc.length)
}
var got []string
for buf := v.data.Front(); buf != nil; buf = buf.Next() {
got = append(got, string(buf.ReadSlice()))
}
if !reflect.DeepEqual(got, tc.after) {
t.Errorf("after = %v; want %v", got, tc.after)
}
})
}
// Failure cases
for _, tc := range []struct {
desc string
// before is the contents for each buffer node initially.
before []string
offset int
length int
}{
{
desc: "offset out-of-range",
before: []string{"hello", " world"},
offset: -1,
length: 3,
},
{
desc: "length too long",
before: []string{"hello", " world"},
offset: 0,
length: 12,
},
{
desc: "length too long with positive offset",
before: []string{"hello", " world"},
offset: 3,
length: 9,
},
{
desc: "length negative",
before: []string{"hello", " world"},
offset: 0,
length: -1,
},
} {
t.Run(tc.desc, func(t *testing.T) {
var v View
for _, s := range tc.before {
v.AppendOwned([]byte(s))
}
if ok := v.Remove(tc.offset, tc.length); ok {
t.Errorf("v.Remove(%d, %d) = true, want false", tc.offset, tc.length)
}
})
}
}
func TestViewSubApply(t *testing.T) {
var v View
v.AppendOwned([]byte("0123"))
v.AppendOwned([]byte("45678"))
v.AppendOwned([]byte("9abcd"))
data := []byte("0123456789abcd")
for i := 0; i <= len(data); i++ {
for j := i; j <= len(data); j++ {
t.Run(fmt.Sprintf("SubApply(%d,%d)", i, j), func(t *testing.T) {
var got []byte
v.SubApply(i, j-i, func(b []byte) {
got = append(got, b...)
})
if want := data[i:j]; !bytes.Equal(got, want) {
t.Errorf("got = %q; want %q", got, want)
}
})
}
}
}
func doSaveAndLoad(t *testing.T, toSave, toLoad *View) {
t.Helper()
var buf bytes.Buffer
@@ -542,3 +817,84 @@ func TestSaveRestoreView(t *testing.T) {
t.Errorf("v.Flatten() = %x, want %x", got, data)
}
}
func TestRangeIntersect(t *testing.T) {
for _, tc := range []struct {
desc string
x, y, want Range
}{
{
desc: "empty intersects empty",
},
{
desc: "empty intersection",
x: Range{end: 10},
y: Range{begin: 10, end: 20},
},
{
desc: "some intersection",
x: Range{begin: 5, end: 20},
y: Range{end: 10},
want: Range{begin: 5, end: 10},
},
} {
t.Run(tc.desc, func(t *testing.T) {
if got := tc.x.Intersect(tc.y); got != tc.want {
t.Errorf("(%#v).Intersect(%#v) = %#v; want %#v", tc.x, tc.y, got, tc.want)
}
if got := tc.y.Intersect(tc.x); got != tc.want {
t.Errorf("(%#v).Intersect(%#v) = %#v; want %#v", tc.y, tc.x, got, tc.want)
}
})
}
}
func TestRangeOffset(t *testing.T) {
for _, tc := range []struct {
input Range
offset int
output Range
}{
{
input: Range{},
offset: 0,
output: Range{},
},
{
input: Range{},
offset: -1,
output: Range{begin: -1, end: -1},
},
{
input: Range{begin: 10, end: 20},
offset: -1,
output: Range{begin: 9, end: 19},
},
{
input: Range{begin: 10, end: 20},
offset: 2,
output: Range{begin: 12, end: 22},
},
} {
if got := tc.input.Offset(tc.offset); got != tc.output {
t.Errorf("(%#v).Offset(%d) = %#v, want %#v", tc.input, tc.offset, got, tc.output)
}
}
}
func TestRangeLen(t *testing.T) {
for _, tc := range []struct {
r Range
want int
}{
{r: Range{}, want: 0},
{r: Range{begin: 1, end: 1}, want: 0},
{r: Range{begin: -1, end: -1}, want: 0},
{r: Range{end: 10}, want: 10},
{r: Range{begin: 5, end: 10}, want: 5},
} {
if got := tc.r.Len(); got != tc.want {
t.Errorf("(%#v).Len() = %d, want %d", tc.r, got, tc.want)
}
}
}
+5 -1
View File
@@ -60,9 +60,13 @@ func IPv4(pkt *stack.PacketBuffer) bool {
return false
}
ipHdr = header.IPv4(hdr)
length := int(ipHdr.TotalLength()) - len(hdr)
if length < 0 {
return false
}
pkt.NetworkProtocolNumber = header.IPv4ProtocolNumber
pkt.Data().CapLength(int(ipHdr.TotalLength()) - len(hdr))
pkt.Data().CapLength(length)
return true
}
@@ -167,8 +167,7 @@ func (r *reassembler) process(first, last uint16, more bool, proto uint8, pkt *s
resPkt := r.holes[0].pkt
for i := 1; i < len(r.holes); i++ {
fragData := r.holes[i].pkt.Data()
resPkt.Data().ReadFromData(fragData, fragData.Size())
stack.MergeFragment(resPkt, r.holes[i].pkt)
}
return resPkt, r.proto, true, memConsumed, nil
}
+8 -2
View File
@@ -1472,13 +1472,19 @@ func (e *endpoint) processExtensionHeaders(h header.IPv6, pkt *stack.PacketBuffe
// If the last header in the payload isn't a known IPv6 extension header,
// handle it as if it is transport layer data.
// Calculate the number of octets parsed from data. We want to remove all
// the data except the unparsed portion located at the end, which its size
// is extHdr.Buf.Size().
trim := pkt.Data().Size() - extHdr.Buf.Size()
// For unfragmented packets, extHdr still contains the transport header.
// Get rid of it.
//
// For reassembled fragments, pkt.TransportHeader is unset, so this is a
// no-op and pkt.Data begins with the transport header.
extHdr.Buf.TrimFront(pkt.TransportHeader().View().Size())
pkt.Data().Replace(extHdr.Buf)
trim += pkt.TransportHeader().View().Size()
pkt.Data().DeleteFront(trim)
stats.PacketsDelivered.Increment()
if p := tcpip.TransportProtocolNumber(extHdr.Identifier); p == header.ICMPv6ProtocolNumber {
+1
View File
@@ -74,6 +74,7 @@ go_library(
visibility = ["//visibility:public"],
deps = [
"//pkg/atomicbitops",
"//pkg/buffer",
"//pkg/ilist",
"//pkg/log",
"//pkg/rand",
File diff suppressed because it is too large Load Diff
+2 -27
View File
@@ -444,23 +444,8 @@ func TestPacketBufferData(t *testing.T) {
checkData(t, pkt, []byte(tc.data+s))
})
// ReadFromData/VV
// ReadFromVV
for _, n := range []int{0, 1, 2, 7, 10, 14, 20} {
t.Run(fmt.Sprintf("ReadFromData%d", n), func(t *testing.T) {
s := "TO READ"
otherPkt := NewPacketBuffer(PacketBufferOptions{
Data: vv(s, s),
})
s += s
pkt := tc.makePkt(t)
pkt.Data().ReadFromData(otherPkt.Data(), n)
if n < len(s) {
s = s[:n]
}
checkData(t, pkt, []byte(tc.data+s))
})
t.Run(fmt.Sprintf("ReadFromVV%d", n), func(t *testing.T) {
s := "TO READ"
srcVV := vv(s, s)
@@ -487,16 +472,6 @@ func TestPacketBufferData(t *testing.T) {
t.Errorf("pkt.Data().ExtractVV().ToOwnedView() = %q, want %q", got, want)
}
})
// Replace
t.Run("Replace", func(t *testing.T) {
s := "REPLACED"
pkt := tc.makePkt(t)
pkt.Data().Replace(vv(s))
checkData(t, pkt, []byte(s))
})
})
}
}
@@ -568,7 +543,7 @@ func checkViewEqual(t *testing.T, what string, got, want buffer.View) {
func checkData(t *testing.T, pkt *PacketBuffer, want []byte) {
t.Helper()
if got := concatViews(pkt.Data().Views()...); !bytes.Equal(got, want) {
t.Errorf("pkt.Data().Views() = %x, want %x", got, want)
t.Errorf("pkt.Data().Views() = 0x%x, want 0x%x", got, want)
}
if got := pkt.Data().Size(); got != len(want) {
t.Errorf("pkt.Data().Size() = %d, want %d", got, len(want))