stream_bench.go

// $go test -bench=. -benchmem -count=1 -timeout=60m ./bench/
// goos: linux
// goarch: amd64
// pkg: open-api-streaming/bench
// cpu: Intel(R) Core(TM) Ultra 7 165H
// BenchmarkNonStreaming/20k-22                   1        4992228793 ns/op             20000 items        343319144 B/op   2021991 allocs/op
// BenchmarkNonStreaming/100k-22                  1        22535888759 ns/op           100000 items        1597454152 B/op 10480705 allocs/op
// BenchmarkNonStreaming/1M-22                    1        236724331269 ns/op         1000000 items        20456522784 B/op        103956295 allocs/op
// BenchmarkStreaming/20k-22                      1        4538066003 ns/op             20000 items        82356144 B/op    1997525 allocs/op
// BenchmarkStreaming/100k-22                     1        20280130851 ns/op           100000 items        393346336 B/op  10598465 allocs/op
// BenchmarkStreaming/1M-22                       1        202630474256 ns/op         1000000 items        3936460880 B/op 105127119 allocs/op
// BenchmarkMemoryNonStreaming/20k-22                     1        4973329348 ns/op               322.6 MB_alloc         7310 MB_sys            20000 items        338232552 B/op   1973396 allocs/op
// BenchmarkMemoryNonStreaming/100k-22                    1        23012171441 ns/op             1523 MB_alloc           7310 MB_sys           100000 items        1597208888 B/op 10482437 allocs/op
// BenchmarkMemoryNonStreaming/1M-22                      1        288301830282 ns/op           19508 MB_alloc           8712 MB_sys          1000000 items        20455808736 B/op        103948066 allocs/op
// BenchmarkMemoryStreaming/20k-22                        1        4396857253 ns/op                78.92 MB_alloc        8712 MB_sys            20000 items        82752320 B/op    1996759 allocs/op
// BenchmarkMemoryStreaming/100k-22                       1        19480238750 ns/op              373.6 MB_alloc         8712 MB_sys           100000 items        391758696 B/op  10598426 allocs/op
// BenchmarkMemoryStreaming/1M-22                         1        202891020870 ns/op            3744 MB_alloc           8712 MB_sys          1000000 items        3925809696 B/op 105090385 allocs/op
// BenchmarkTimeToFirstItem/NonStreaming/20k-22                   1        5074887860 ns/op              5074 ms/first_item        313090040 B/op   1972508 allocs/op
// BenchmarkTimeToFirstItem/Streaming/20k-22                      1        4588397369 ns/op               877.0 ms/first_item      82202784 B/op    1992641 allocs/op
// BenchmarkTimeToFirstItem/NonStreaming/100k-22                  1        24627578986 ns/op            24627 ms/first_item        1472876416 B/op 10482250 allocs/op
// BenchmarkTimeToFirstItem/Streaming/100k-22                     1        22736601083 ns/op              879.0 ms/first_item      395459544 B/op  10594575 allocs/op
// BenchmarkTimeToFirstItem/NonStreaming/1M-22                    1        22340813667 ns/op            22340 ms/first_item        1472728624 B/op 10482367 allocs/op
// BenchmarkTimeToFirstItem/Streaming/1M-22                       1        22876171710 ns/op             1080 ms/first_item        393312472 B/op  10598566 allocs/op
// PASS
// ok      open-api-streaming/bench        1137.372s
package bench

import (
	"context"
	"fmt"
	"log"
	"net/http"
	"runtime"
	"testing"
	"time"

	"open-api-streaming/adapters"
)

const (
  DetailedReportByPeriodUrl = "" // todo: add
	maxLimit                  = 100_000
)

var (
	DetailedReportByPeriodHeaders = map[string]string{
    // todo: add
	}
	HttpClient = &http.Client{
		Timeout: 0,
	}
)

func init() {
	log.SetFlags(log.LstdFlags | log.Lshortfile)
}

type benchCase struct {
	name    string
	target  int
	timeout time.Duration
}

var benchCases = []benchCase{
	{name: "20k", target: 20_000, timeout: 30 * time.Second},
	{name: "100k", target: 100_000, timeout: 2 * time.Minute},
	{name: "1M", target: 1_000_000, timeout: 10 * time.Minute},
}

func pagesFor(target int) (pages int, limit int) {
	if target <= maxLimit {
		return 1, target
	}
	return (target + maxLimit - 1) / maxLimit, maxLimit
}

func fetchAllNonStreaming(
	ctx context.Context,
	frApi adapters.FinancialReportAPI,
	bc benchCase,
) ([]adapters.ReportDetail, error) {
	pages, limit := pagesFor(bc.target)
	var all []adapters.ReportDetail
	rrdID := 0

	for page := 0; page < pages; page++ {
		req := adapters.DetailedReportByPeriodRequest{
			DateFrom: "2025-01-01",
			DateTo:   "2026-01-01",
			Limit:    limit,
			RrdID:    rrdID,
			Period:   "daily",
		}
		details, err := frApi.GetDetailsByPeriod(ctx, req)
		if err != nil {
			return nil, fmt.Errorf("page %d: %w", page, err)
		}
		if len(details) == 0 {
			break
		}
		all = append(all, details...)
		rrdID = int(details[len(details)].RrdID)
	}
	return all, nil
}

func fetchAllStreaming(
	ctx context.Context,
	frApi adapters.FinancialReportAPI,
	bc benchCase,
) (int, error) {
	pages, limit := pagesFor(bc.target)
	totalCount := 0
	rrdID := 0

	for page := 0; page < pages; page++ {
		req := adapters.DetailedReportByPeriodRequest{
			DateFrom: "2025-01-01",
			DateTo:   "2026-01-01",
			Limit:    limit,
			RrdID:    rrdID,
			Period:   "daily",
		}
		ch, err := frApi.GetDetailsByPeriodChannel(ctx, req)
		if err != nil {
			return 0, fmt.Errorf("page %d: %w", page, err)
		}
		var lastRrdID int64
		count := 0
		for detail := range ch {
			lastRrdID = detail.RrdID
			count++
		}
		if count == 0 {
			break
		}
		totalCount += count
		rrdID = int(lastRrdID)
	}
	return totalCount, nil
}

func BenchmarkNonStreaming(b *testing.B) {
	frApi := adapters.NewFinancialReportAPI(HttpClient, DetailedReportByPeriodUrl, DetailedReportByPeriodHeaders)

	for _, bc := range benchCases {
		b.Run(bc.name, func(b *testing.B) {
			b.ResetTimer()
			for i := 0; i < b.N; i++ {
				ctx, cancel := context.WithTimeout(context.Background(), bc.timeout)
				details, err := fetchAllNonStreaming(ctx, frApi, bc)
				cancel()
				if err != nil {
					b.Fatalf("failed: %v", err)
				}
				b.ReportMetric(float64(len(details)), "items")
			}
		})
	}
}

func BenchmarkStreaming(b *testing.B) {
	frApi := adapters.NewFinancialReportAPI(HttpClient, DetailedReportByPeriodUrl, DetailedReportByPeriodHeaders)

	for _, bc := range benchCases {
		b.Run(bc.name, func(b *testing.B) {
			b.ResetTimer()
			for i := 0; i < b.N; i++ {
				ctx, cancel := context.WithTimeout(context.Background(), bc.timeout)
				count, err := fetchAllStreaming(ctx, frApi, bc)
				cancel()
				if err != nil {
					b.Fatalf("failed: %v", err)
				}
				if count == 0 {
					b.Fatal("no items received")
				}
				b.ReportMetric(float64(count), "items")
			}
		})
	}
}

func BenchmarkMemoryNonStreaming(b *testing.B) {
	frApi := adapters.NewFinancialReportAPI(HttpClient, DetailedReportByPeriodUrl, DetailedReportByPeriodHeaders)

	for _, bc := range benchCases {
		b.Run(bc.name, func(b *testing.B) {
			b.ResetTimer()
			for i := 0; i < b.N; i++ {
				runtime.GC()
				var before runtime.MemStats
				runtime.ReadMemStats(&before)

				ctx, cancel := context.WithTimeout(context.Background(), bc.timeout)
				details, err := fetchAllNonStreaming(ctx, frApi, bc)
				cancel()
				if err != nil {
					b.Fatalf("failed: %v", err)
				}

				var after runtime.MemStats
				runtime.ReadMemStats(&after)

				b.ReportMetric(float64(after.TotalAlloc-before.TotalAlloc)/1024/1024, "MB_alloc")
				b.ReportMetric(float64(after.Sys)/1024/1024, "MB_sys")
				b.ReportMetric(float64(len(details)), "items")

				runtime.GC()
			}
		})
	}
}

func BenchmarkMemoryStreaming(b *testing.B) {
	frApi := adapters.NewFinancialReportAPI(HttpClient, DetailedReportByPeriodUrl, DetailedReportByPeriodHeaders)

	for _, bc := range benchCases {
		b.Run(bc.name, func(b *testing.B) {
			b.ResetTimer()
			for i := 0; i < b.N; i++ {
				runtime.GC()
				var before runtime.MemStats
				runtime.ReadMemStats(&before)

				ctx, cancel := context.WithTimeout(context.Background(), bc.timeout)
				count, err := fetchAllStreaming(ctx, frApi, bc)
				cancel()
				if err != nil {
					b.Fatalf("failed: %v", err)
				}

				var after runtime.MemStats
				runtime.ReadMemStats(&after)

				b.ReportMetric(float64(after.TotalAlloc-before.TotalAlloc)/1024/1024, "MB_alloc")
				b.ReportMetric(float64(after.Sys)/1024/1024, "MB_sys")
				b.ReportMetric(float64(count), "items")

				runtime.GC()
			}
		})
	}
}

func BenchmarkTimeToFirstItem(b *testing.B) {
	frApi := adapters.NewFinancialReportAPI(HttpClient, DetailedReportByPeriodUrl, DetailedReportByPeriodHeaders)

	for _, bc := range benchCases {
		_, limit := pagesFor(bc.target)
		req := adapters.DetailedReportByPeriodRequest{
			DateFrom: "2025-01-01",
			DateTo:   "2026-01-01",
			Limit:    limit,
			RrdID:    0,
			Period:   "daily",
		}

		b.Run(fmt.Sprintf("NonStreaming/%s", bc.name), func(b *testing.B) {
			b.ResetTimer()
			for i := 0; i < b.N; i++ {
				ctx, cancel := context.WithTimeout(context.Background(), bc.timeout)
				start := time.Now()
				details, err := frApi.GetDetailsByPeriod(ctx, req)
				ttfi := time.Since(start)
				cancel()
				if err != nil {
					b.Fatalf("failed: %v", err)
				}
				if len(details) == 0 {
					b.Fatal("no items received")
				}
				b.ReportMetric(float64(ttfi.Milliseconds()), "ms/first_item")
			}
		})

		b.Run(fmt.Sprintf("Streaming/%s", bc.name), func(b *testing.B) {
			b.ResetTimer()
			for i := 0; i < b.N; i++ {
				ctx, cancel := context.WithTimeout(context.Background(), bc.timeout)
				start := time.Now()
				ch, err := frApi.GetDetailsByPeriodChannel(ctx, req)
				if err != nil {
					cancel()
					b.Fatalf("failed: %v", err)
				}
				_, ok := <-ch
				ttfi := time.Since(start)
				if !ok {
					cancel()
					b.Fatal("no items received from stream")
				}
				b.ReportMetric(float64(ttfi.Milliseconds()), "ms/first_item")
				for range ch {
				}
				cancel()
			}
		})
	}
}