frostfs-node/pkg/local_object_storage/writecache/benchmark/writecache_test.go

package benchmark

import (
	"context"
	"encoding/binary"
	"fmt"
	"os"
	"path/filepath"
	"testing"

	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/common"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/teststore"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/internal/testutil"
	meta "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/metabase"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard/mode"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/writecache"
	"github.com/stretchr/testify/require"
)

func BenchmarkWriteAllocatedSeq(b *testing.B) {
	const objectCount = 100
	sizes := []uint64{64 << 10}
	for _, size := range sizes {
		objGen := testutil.RandObjGenerator{ObjSize: size}
		b.Run(fmt.Sprintf("ftruncate %d", size), func(b *testing.B) {
			b.ResetTimer()
			for run := 0; run < b.N; run++ {
				b.StopTimer()
				var dataOffset uint32
				dir := b.TempDir()
				filePrefix := fmt.Sprintf("writecache_%d_%d", run, size)
				data, err := os.OpenFile(
					filepath.Join(dir, fmt.Sprintf("%s_%d", filePrefix, dataOffset)),
					os.O_WRONLY|os.O_CREATE|os.O_EXCL|os.O_SYNC,
					0o700,
				)
				require.NoError(b, err, "open writecache file")

				obj := objGen.Next()
				rawData, err := obj.Marshal()
				require.NoError(b, err, "marshaling object")
				require.NoError(b, data.Truncate(int64((len(rawData)+4)*objectCount)))
				b.StartTimer()

				for range objectCount {
					obj := objGen.Next()
					rawData, err := obj.Marshal()
					require.NoError(b, err, "marshaling object")

					objectStruct := make([]byte, 4+len(rawData))
					binary.LittleEndian.PutUint32(objectStruct, uint32(len(rawData)))
					copy(objectStruct[4:], rawData)
					c, err := data.WriteAt(objectStruct, int64(dataOffset))
					require.NoError(b, err)
					require.True(b, c == len(objectStruct))
					require.NoError(b, os.Rename(
						filepath.Join(dir, fmt.Sprintf("%s_%d", filePrefix, dataOffset)),
						filepath.Join(dir, fmt.Sprintf("%s_%d", filePrefix, dataOffset+uint32(len(objectStruct)))),
					))
					dataOffset += uint32(len(objectStruct))
				}

				require.NoError(b, data.Close())
			}
		})

		b.Run(fmt.Sprintf("fstree %d", size), func(b *testing.B) {
			cache := newCache(b)
			benchmarkPutPrepare(b, cache)
			defer func() { require.NoError(b, cache.Close(context.Background())) }()

			b.ResetTimer()
			for range b.N {
				for range objectCount {
					obj := objGen.Next()
					rawData, err := obj.Marshal()
					require.NoError(b, err, "marshaling object")
					prm := common.PutPrm{
						Address: testutil.AddressFromObject(b, obj),
						Object:  obj,
						RawData: rawData,
					}
					if _, err := cache.Put(context.Background(), prm); err != nil {
						b.Fatalf("putting: %v", err)
					}
				}
			}
		})
	}
}

func BenchmarkWritecacheSeq(b *testing.B) {
	const payloadSize = 8 << 10
	b.Run("bbolt_seq", func(b *testing.B) {
		benchmarkPutSeq(b, newCache(b), payloadSize)
	})
}

func BenchmarkWritecachePar(b *testing.B) {
	const payloadSize = 8 << 10
	b.Run("bbolt_par", func(b *testing.B) {
		benchmarkPutPar(b, newCache(b), payloadSize)
	})
}

func BenchmarkWriteAfterDelete(b *testing.B) {
	const payloadSize = 32 << 10
	const parallel = 25

	cache := newCache(b)
	benchmarkPutPrepare(b, cache)
	b.Run(fmt.Sprintf("%dB_before", payloadSize), func(b *testing.B) {
		b.SetParallelism(parallel)
		benchmarkRunPar(b, cache, payloadSize)
	})
	require.NoError(b, cache.Flush(context.Background(), false, false))
	b.Run(fmt.Sprintf("%dB_after", payloadSize), func(b *testing.B) {
		b.SetParallelism(parallel)
		benchmarkRunPar(b, cache, payloadSize)
	})
	require.NoError(b, cache.Close(context.Background()))
}

func benchmarkPutSeq(b *testing.B, cache writecache.Cache, size uint64) {
	benchmarkPutPrepare(b, cache)
	defer func() { require.NoError(b, cache.Close(context.Background())) }()

	ctx := context.Background()
	objGen := testutil.RandObjGenerator{ObjSize: size}

	b.ResetTimer()
	for range b.N {
		obj := objGen.Next()
		rawData, err := obj.Marshal()
		require.NoError(b, err, "marshaling object")
		prm := common.PutPrm{
			Address: testutil.AddressFromObject(b, obj),
			Object:  obj,
			RawData: rawData,
		}
		if _, err := cache.Put(ctx, prm); err != nil {
			b.Fatalf("putting: %v", err)
		}
	}
}

func benchmarkPutPar(b *testing.B, cache writecache.Cache, size uint64) {
	benchmarkPutPrepare(b, cache)
	defer func() { require.NoError(b, cache.Close(context.Background())) }()

	benchmarkRunPar(b, cache, size)
}

func benchmarkRunPar(b *testing.B, cache writecache.Cache, size uint64) {
	ctx := context.Background()

	b.ResetTimer()
	b.RunParallel(func(pb *testing.PB) {
		objGen := testutil.RandObjGenerator{ObjSize: size}
		for pb.Next() {
			obj := objGen.Next()
			rawData, err := obj.Marshal()
			require.NoError(b, err, "marshaling object")
			prm := common.PutPrm{
				Address: testutil.AddressFromObject(b, obj),
				Object:  obj,
				RawData: rawData,
			}
			if _, err := cache.Put(ctx, prm); err != nil {
				b.Fatalf("putting: %v", err)
			}
		}
	})
}

func benchmarkPutPrepare(b *testing.B, cache writecache.Cache) {
	require.NoError(b, cache.Open(context.Background(), mode.ReadWrite), "opening")
	require.NoError(b, cache.Init(context.Background()), "initializing")
}

type testMetabase struct{}

func (testMetabase) UpdateStorageID(context.Context, meta.UpdateStorageIDPrm) (meta.UpdateStorageIDRes, error) {
	return meta.UpdateStorageIDRes{}, nil
}

func newCache(b *testing.B) writecache.Cache {
	bs := teststore.New(
		teststore.WithPut(func(pp common.PutPrm) (common.PutRes, error) { return common.PutRes{}, nil }),
	)
	return writecache.New(
		writecache.WithPath(b.TempDir()),
		writecache.WithBlobstor(bs),
		writecache.WithMetabase(testMetabase{}),
		writecache.WithMaxCacheSize(256<<30),
	)
}