Files
tidb/pkg/lightning/membuf/buffer_test.go

406 lines
9.1 KiB
Go

// Copyright 2021 PingCAP, Inc.
//
// 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 membuf
import (
"bytes"
"crypto/rand"
rand2 "math/rand"
"runtime"
"slices"
"sync"
"testing"
"time"
"github.com/stretchr/testify/require"
)
type testAllocator struct {
allocs int
frees int
}
func (t *testAllocator) Alloc(n int) []byte {
t.allocs++
return make([]byte, n)
}
func (t *testAllocator) Free(_ []byte) {
t.frees++
}
func TestBufferPool(t *testing.T) {
allocator := &testAllocator{}
pool := NewPool(
WithBlockNum(2),
WithAllocator(allocator),
WithBlockSize(1024),
)
defer pool.Destroy()
bytesBuf := pool.NewBuffer()
bytesBuf.AllocBytes(256)
require.Equal(t, 1, allocator.allocs)
bytesBuf.AllocBytes(512)
require.Equal(t, 1, allocator.allocs)
bytesBuf.AllocBytes(257)
require.Equal(t, 2, allocator.allocs)
bytesBuf.AllocBytes(767)
require.Equal(t, 2, allocator.allocs)
largeBytes := bytesBuf.AllocBytes(1025)
require.Equal(t, 1025, len(largeBytes))
require.Equal(t, 2, allocator.allocs)
require.Equal(t, 0, allocator.frees)
bytesBuf.Destroy()
require.Equal(t, 0, allocator.frees)
bytesBuf = pool.NewBuffer()
for range 6 {
bytesBuf.AllocBytes(512)
}
bytesBuf.Destroy()
require.Equal(t, 3, allocator.allocs)
require.Equal(t, 1, allocator.frees)
}
func TestPoolMemLimit(t *testing.T) {
limiter := NewLimiter(2*1024*1024 + 2*smallObjOverheadBatch)
// only allow to allocate one block
pool := NewPool(
WithBlockSize(2*1024*1024),
WithPoolMemoryLimiter(limiter),
)
defer pool.Destroy()
buf := pool.NewBuffer()
buf.AllocBytes(1024 * 1024)
buf.AllocBytes(1024 * 1024)
buf2 := pool.NewBuffer()
done := make(chan struct{}, 1)
go func() {
buf2.AllocBytes(1024 * 1024)
buf2.Destroy()
done <- struct{}{}
}()
// sleep a while to make sure the goroutine is started
time.Sleep(50 * time.Millisecond)
require.Len(t, done, 0)
// reset will not release memory to pool
buf.Reset()
buf.AllocBytes(1024 * 1024)
buf.AllocBytes(1024 * 1024)
require.Len(t, done, 0)
// destroy will release memory to pool
buf.Destroy()
// wait buf2 to finish
require.Eventually(t, func() bool {
return len(done) > 0
}, time.Second, 10*time.Millisecond)
// after buf2 is finished, still can allocate memory from pool
buf.AllocBytes(2 * 1024 * 1024)
buf.Destroy()
}
func TestBufferIsolation(t *testing.T) {
pool := NewPool(WithBlockSize(1024))
defer pool.Destroy()
bytesBuf := pool.NewBuffer()
defer bytesBuf.Destroy()
b1 := bytesBuf.AllocBytes(16)
b2 := bytesBuf.AllocBytes(16)
require.Len(t, b1, cap(b1))
require.Len(t, b2, cap(b2))
_, err := rand.Read(b2)
require.NoError(t, err)
b3 := slices.Clone(b2)
b1 = append(b1, 0, 1, 2, 3)
require.Equal(t, b3, b2)
require.NotEqual(t, b2, b1)
}
func TestBufferMemLimit(t *testing.T) {
pool := NewPool(WithBlockSize(10))
defer pool.Destroy()
// the actual memory limit is 10 bytes.
bytesBuf := pool.NewBuffer(WithBufferMemoryLimit(5))
got, _ := bytesBuf.AllocBytesWithSliceLocation(9)
require.NotNil(t, got)
got, _ = bytesBuf.AllocBytesWithSliceLocation(3)
require.Nil(t, got)
bytesBuf.Destroy()
// test the buffer is still usable after destroy.
got, _ = bytesBuf.AllocBytesWithSliceLocation(3)
require.NotNil(t, got)
// exactly 2 block
bytesBuf = pool.NewBuffer(WithBufferMemoryLimit(20))
got, _ = bytesBuf.AllocBytesWithSliceLocation(9)
require.NotNil(t, got)
got, _ = bytesBuf.AllocBytesWithSliceLocation(9)
require.NotNil(t, got)
got, _ = bytesBuf.AllocBytesWithSliceLocation(2)
require.Nil(t, got)
// after reset, can get same allocation again
bytesBuf.Reset()
got, _ = bytesBuf.AllocBytesWithSliceLocation(9)
require.NotNil(t, got)
got, _ = bytesBuf.AllocBytesWithSliceLocation(9)
require.NotNil(t, got)
got, _ = bytesBuf.AllocBytesWithSliceLocation(2)
require.Nil(t, got)
}
func TestGetAlignedSizeGetBlockCnt(t *testing.T) {
require.EqualValues(t, 1, getBlockCnt(10, 16))
require.EqualValues(t, 2, getBlockCnt(17, 16))
require.EqualValues(t, 16, GetAlignedSize(10, 16))
require.EqualValues(t, 32, GetAlignedSize(17, 16))
}
const dataNum = 100 * 1024 * 1024
func BenchmarkStoreSlice(b *testing.B) {
data := make([][]byte, dataNum)
for b.Loop() {
func() {
pool := NewPool()
defer pool.Destroy()
bytesBuf := pool.NewBuffer()
defer bytesBuf.Destroy()
for j := range data {
data[j] = bytesBuf.AllocBytes(10)
}
}()
}
}
func BenchmarkStoreLocation(b *testing.B) {
data := make([]SliceLocation, dataNum)
for b.Loop() {
func() {
pool := NewPool()
defer pool.Destroy()
bytesBuf := pool.NewBuffer()
defer bytesBuf.Destroy()
for j := range data {
_, data[j] = bytesBuf.AllocBytesWithSliceLocation(10)
}
}()
}
}
const sortDataNum = 1024 * 1024
func BenchmarkSortSlice(b *testing.B) {
data := make([][]byte, sortDataNum)
// fixed seed for benchmark
rnd := rand2.New(rand2.NewSource(6716))
for b.Loop() {
func() {
pool := NewPool()
defer pool.Destroy()
bytesBuf := pool.NewBuffer()
defer bytesBuf.Destroy()
for j := range data {
data[j] = bytesBuf.AllocBytes(10)
rnd.Read(data[j])
}
slices.SortFunc(data, func(a, b []byte) int {
return bytes.Compare(a, b)
})
}()
}
}
func BenchmarkSortLocation(b *testing.B) {
data := make([]SliceLocation, sortDataNum)
// fixed seed for benchmark
rnd := rand2.New(rand2.NewSource(6716))
for b.Loop() {
func() {
pool := NewPool()
defer pool.Destroy()
bytesBuf := pool.NewBuffer()
defer bytesBuf.Destroy()
for j := range data {
var buf []byte
buf, data[j] = bytesBuf.AllocBytesWithSliceLocation(10)
rnd.Read(buf)
}
slices.SortFunc(data, func(a, b SliceLocation) int {
return bytes.Compare(bytesBuf.GetSlice(&a), bytesBuf.GetSlice(&b))
})
}()
}
}
func BenchmarkSortSliceWithGC(b *testing.B) {
data := make([][]byte, sortDataNum)
// fixed seed for benchmark
rnd := rand2.New(rand2.NewSource(6716))
for b.Loop() {
func() {
pool := NewPool()
defer pool.Destroy()
bytesBuf := pool.NewBuffer()
defer bytesBuf.Destroy()
for j := range data {
data[j] = bytesBuf.AllocBytes(10)
rnd.Read(data[j])
}
runtime.GC()
slices.SortFunc(data, func(a, b []byte) int {
return bytes.Compare(a, b)
})
}()
}
}
func BenchmarkSortLocationWithGC(b *testing.B) {
data := make([]SliceLocation, sortDataNum)
// fixed seed for benchmark
rnd := rand2.New(rand2.NewSource(6716))
for b.Loop() {
func() {
pool := NewPool()
defer pool.Destroy()
bytesBuf := pool.NewBuffer()
defer bytesBuf.Destroy()
for j := range data {
var buf []byte
buf, data[j] = bytesBuf.AllocBytesWithSliceLocation(10)
rnd.Read(buf)
}
runtime.GC()
slices.SortFunc(data, func(a, b SliceLocation) int {
return bytes.Compare(bytesBuf.GetSlice(&a), bytesBuf.GetSlice(&b))
})
}()
}
}
func BenchmarkSortLocationWithEscape(b *testing.B) {
data := make([]SliceLocation, sortDataNum)
// fixed seed for benchmark
rnd := rand2.New(rand2.NewSource(6716))
for b.Loop() {
func() {
pool := NewPool()
defer pool.Destroy()
bytesBuf := pool.NewBuffer()
defer bytesBuf.Destroy()
for j := range data {
var buf []byte
buf, data[j] = bytesBuf.AllocBytesWithSliceLocation(10)
rnd.Read(buf)
}
var (
dupFound bool
dupLoc *SliceLocation
)
slices.SortFunc(data, func(a, b SliceLocation) int {
res := bytes.Compare(bytesBuf.GetSlice(&a), bytesBuf.GetSlice(&b))
if res == 0 && !dupFound {
dupFound = true
dupLoc = &a
}
return res
})
_ = dupLoc
}()
}
}
func BenchmarkSortLocationWithoutEscape(b *testing.B) {
data := make([]SliceLocation, sortDataNum)
// fixed seed for benchmark
rnd := rand2.New(rand2.NewSource(6716))
for b.Loop() {
func() {
pool := NewPool()
defer pool.Destroy()
bytesBuf := pool.NewBuffer()
defer bytesBuf.Destroy()
for j := range data {
var buf []byte
buf, data[j] = bytesBuf.AllocBytesWithSliceLocation(10)
rnd.Read(buf)
}
var (
dupFound bool
dupLoc SliceLocation
)
slices.SortFunc(data, func(a, b SliceLocation) int {
res := bytes.Compare(bytesBuf.GetSlice(&a), bytesBuf.GetSlice(&b))
if res == 0 && !dupFound {
dupFound = true
dupLoc = a
}
return res
})
_ = dupLoc
}()
}
}
func BenchmarkConcurrentAcquire(b *testing.B) {
for b.Loop() {
limiter := NewLimiter(512 * 1024 * 1024)
pool := NewPool(WithPoolMemoryLimiter(limiter), WithBlockSize(4*1024))
// start 1000 clients, each client will acquire 100B for 1000 times.
wg := sync.WaitGroup{}
clientNum := 1000
wg.Add(clientNum)
for range clientNum {
go func() {
defer wg.Done()
buf := pool.NewBuffer()
for range 1000 {
buf.AllocBytes(100)
}
buf.Destroy()
}()
}
wg.Wait()
pool.Destroy()
}
}