VictoriaMetrics/lib/persistentqueue/fastqueue_timing_test.go

package persistentqueue

import (
	"fmt"
	"testing"

	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
)

func BenchmarkFastQueueThroughputSerial(b *testing.B) {
	const iterationsCount = 10
	for _, blockSize := range []int{1e0, 1e1, 1e2, 1e3, 1e4, 1e5, 1e6} {
		block := make([]byte, blockSize)
		b.Run(fmt.Sprintf("block-size-%d", blockSize), func(b *testing.B) {
			b.ReportAllocs()
			b.SetBytes(int64(blockSize) * iterationsCount)
			path := fmt.Sprintf("bench-fast-queue-throughput-serial-%d", blockSize)
			fs.MustRemoveDir(path)
			fq := MustOpenFastQueue(path, "foobar", iterationsCount*2, 0, false)
			defer func() {
				fq.MustClose()
				fs.MustRemoveDir(path)
			}()
			for range b.N {
				writeReadIterationFastQueue(fq, block, iterationsCount)
			}
		})
	}
}

func BenchmarkFastQueueThroughputConcurrent(b *testing.B) {
	const iterationsCount = 10
	for _, blockSize := range []int{1e0, 1e1, 1e2, 1e3, 1e4, 1e5, 1e6} {
		block := make([]byte, blockSize)
		b.Run(fmt.Sprintf("block-size-%d", blockSize), func(b *testing.B) {
			b.ReportAllocs()
			b.SetBytes(int64(blockSize) * iterationsCount)
			path := fmt.Sprintf("bench-fast-queue-throughput-concurrent-%d", blockSize)
			fs.MustRemoveDir(path)
			fq := MustOpenFastQueue(path, "foobar", iterationsCount*cgroup.AvailableCPUs()*2, 0, false)
			defer func() {
				fq.MustClose()
				fs.MustRemoveDir(path)
			}()
			b.RunParallel(func(pb *testing.PB) {
				for pb.Next() {
					writeReadIterationFastQueue(fq, block, iterationsCount)
				}
			})
		})
	}
}

func writeReadIterationFastQueue(fq *FastQueue, block []byte, iterationsCount int) {
	for range iterationsCount {
		if !fq.TryWriteBlock(block) {
			panic(fmt.Errorf("TryWriteBlock must return true"))
		}
	}
	var ok bool
	bb := bbPool.Get()
	for range iterationsCount {
		bb.B, ok = fq.MustReadBlock(bb.B[:0])
		if !ok {
			panic(fmt.Errorf("unexpected ok=false"))
		}
	}
	bbPool.Put(bb)
}