pool/scenario_test.go

package pool_test

import (
	"code.squareroundforest.org/arpio/pool"
	"code.squareroundforest.org/arpio/times"
	"math"
	"math/rand"
	"strings"
	"testing"
	"time"
)

type scenarioOptions struct {
	algo        pool.Algo
	initial     int
	ops         int
	deviation   int
	changeRate  int
	minDelay    time.Duration
	maxDelay    time.Duration
	concurrency int
	plot        bool
	exclude     []string
}

type (
	resource[T any] chan T
	active          resource[[][]byte]
	stats           resource[[]pool.Stats]
	scenarioStep    func(scenarioOptions, *rand.Rand, int, int, func(), func())
	verifyScenario  func(*testing.T, scenarioOptions, []pool.Stats)
)

func initResource[T any]() resource[T] {
	var zero T
	r := make(resource[T], 1)
	r <- zero
	return r
}

func (r resource[T]) apply(f func(T) T) {
	v := <-r
	defer func() {
		r <- v
	}()

	v = f(v)
}

func (a active) count() int {
	var c int
	resource[[][]byte](a).apply(func(a [][]byte) [][]byte {
		c = len(a)
		return a
	})

	return c
}

func (a active) shift() []byte {
	var b []byte
	resource[[][]byte](a).apply(func(a [][]byte) [][]byte {
		if len(a) == 0 {
			return a
		}

		b, a = a[len(a)-1], a[:len(a)-1]
		return a
	})

	return b
}

func (a active) push(b []byte) {
	resource[[][]byte](a).apply(func(a [][]byte) [][]byte {
		return append(a, b)
	})
}

func (s stats) get() []pool.Stats {
	var a []pool.Stats
	resource[[]pool.Stats](s).apply(func(r []pool.Stats) []pool.Stats {
		a = r
		return r
	})

	return a
}

func (s stats) push(a pool.Stats) {
	resource[[]pool.Stats](s).apply(func(r []pool.Stats) []pool.Stats {
		return append(r, a)
	})
}

func testScenario(t *testing.T, o scenarioOptions, step scenarioStep, verify verifyScenario) {
	for _, n := range o.exclude {
		if strings.HasSuffix(t.Name(), n) {
			t.Skip()
		}
	}

	var (
		testClock times.TestClock
		clock     times.Clock
	)

	if o.concurrency <= 0 {
		o.concurrency = 1
	}

	o.initial *= o.concurrency
	if o.minDelay > 0 {
		if o.maxDelay < o.minDelay {
			o.maxDelay = o.minDelay
		}

		testClock = times.Test()
		clock = testClock
	}

	alloc := func() ([]byte, error) { return make([]byte, 1<<9), nil }
	po := pool.Options{
		Algo:  o.algo,
		Clock: clock,
	}

	p := pool.Make(alloc, nil, po)
	active := active(initResource[[][]byte]())
	stats := stats(initResource[[]pool.Stats]())
	get := func() {
		b, err := p.Get()
		if err != nil {
			t.Fatal(err)
		}

		active.push(b)
	}

	put := func() {
		b := active.shift()
		if len(b) == 0 {
			return
		}

		p.Put(b)
	}

	for i := 0; i < o.initial; i++ {
		get()
	}

	stats.push(p.Stats())
	rnd := rand.New(rand.NewSource(0))
	c := make(chan struct{}, o.concurrency)
	iter := func(i int, localClock times.TestClock) {
		if o.minDelay > 0 {
			d := o.minDelay
			if o.maxDelay > o.minDelay {
				diff := o.maxDelay - o.minDelay
				rdiff := rand.Intn(int(diff))
				d += time.Duration(rdiff)
			}

			localClock.Pass(d)
			testClock.Jump(localClock.Now())
		}

		step(o, rnd, i, active.count(), get, put)
		stats.push(p.Stats())
	}

	for i := 0; i < o.concurrency; i++ {
		go func() {
			var localClock times.TestClock
			if o.minDelay > 0 {
				localClock = times.Test()
			}

			for i := 0; i < o.ops; i++ {
				iter(i, localClock)
			}

			c <- struct{}{}
		}()
	}

	for i := 0; i < o.concurrency; i++ {
		<-c
	}

	if o.plot {
		for i, s := range stats.get() {
			t.Log(i, s)
		}
	}

	if verify != nil {
		verify(t, o, stats.get())
	}
}

func testCyclicScenario(t *testing.T, o scenarioOptions, step scenarioStep, verify verifyScenario) {
	cyclicStep := func(o scenarioOptions, rnd *rand.Rand, i, active int, get, put func()) {
		o.ops = o.ops / 4
		i = i % o.ops
		step(o, rnd, i, active, get, put)
	}

	testScenario(t, o, cyclicStep, verify)
}

func steadyStep(o scenarioOptions, rnd *rand.Rand, i, active int, get, put func(), target int) {
	switch {
	case active > target:
		put()
	case active < target-o.deviation:
		get()
	default:
		if rnd.Intn(2) == 1 {
			get()
		} else {
			put()
		}
	}
}

func steady(o scenarioOptions, rnd *rand.Rand, i, active int, get, put func()) {
	steadyStep(o, rnd, i, active, get, put, o.initial)
}

func jumpStep(o scenarioOptions, rnd *rand.Rand, i, active int, get, put func()) {
	get()
}

func dropStep(o scenarioOptions, rnd *rand.Rand, i, active int, get, put func()) {
	put()
}

func steadyStepUp(o scenarioOptions, rnd *rand.Rand, i, active int, get, put func()) {
	switch {
	case i < o.ops/3:
		steadyStep(o, rnd, i, active, get, put, o.initial)
	case i >= o.ops/3 && i < 2*o.ops/3:
		jumpStep(o, rnd, i, active, get, put)
	default:
		steadyStep(o, rnd, i, active, get, put, o.initial/3)
	}
}

func steadyStepDown(o scenarioOptions, rnd *rand.Rand, i, active int, get, put func()) {
	switch {
	case i < o.ops/3:
		steadyStep(o, rnd, i, active, get, put, o.initial)
	case i >= o.ops/3 && i < 2*o.ops/3:
		dropStep(o, rnd, i, active, get, put)
	default:
		steadyStep(o, rnd, i, active, get, put, o.initial/3)
	}
}

func steadyStepDownToZero(o scenarioOptions, rnd *rand.Rand, i, active int, get, put func()) {
	switch {
	case i < o.ops/3:
		steadyStep(o, rnd, i, active, get, put, o.initial)
	case i >= o.ops/3 && i < 2*o.ops/3:
		dropStep(o, rnd, i, active, get, put)
	default:
		steadyStep(o, rnd, i, active, get, put, 0)
	}
}

func change(o scenarioOptions, rnd *rand.Rand, i, active int, get, put func()) {
	cr := o.changeRate
	inc := get
	dec := put
	if cr < 0 {
		cr = 0 - cr
		inc, dec = put, get
	}

	v := rnd.Intn(cr + 2)
	if v == 0 {
		dec()
		return
	}

	inc()
}

func changeAndJump(o scenarioOptions, rnd *rand.Rand, i, active int, get, put func()) {
	var f scenarioStep
	switch {
	case 3*i < 2*o.ops:
		f = change
	default:
		f = jumpStep
	}

	f(o, rnd, i, active, get, put)
}

func cyclicSpikes(o scenarioOptions, rnd *rand.Rand, i, active int, get, put func()) {
	var f scenarioStep
	switch {
	case 3*i < o.ops:
		o.initial = active
		f = steady
	case 3*i < 2*o.ops:
		f = jumpStep
	default:
		f = dropStep
	}

	f(o, rnd, i, active, get, put)
}

func cyclicDrops(o scenarioOptions, rnd *rand.Rand, i, active int, get, put func()) {
	var f scenarioStep
	switch {
	case 3*i < o.ops:
		o.initial = active
		f = steady
	case 3*i < 2*o.ops:
		f = dropStep
	default:
		f = jumpStep
	}

	f(o, rnd, i, active, get, put)
}

func sigmaSteps(o scenarioOptions, rnd *rand.Rand, i, active int, get, put func()) {
	var f scenarioStep
	switch {
	case 6*i < o.ops:
		f = jumpStep
	case 2*i < o.ops:
		o.initial = active
		f = steady
	case 3*i < 2*o.ops:
		f = dropStep
	default:
		o.initial = active
		f = steady
	}

	f(o, rnd, i, active, get, put)
}

func chainSaw(o scenarioOptions, rnd *rand.Rand, i, active int, get, put func()) {
	var f scenarioStep
	switch {
	case 3*i < o.ops:
		f = change
	case 2*i < o.ops:
		f = dropStep
	case 6*i < 5*o.ops:
		f = change
	default:
		f = dropStep
	}

	f(o, rnd, i, active, get, put)
}

func inverseChainSaw(o scenarioOptions, rnd *rand.Rand, i, active int, get, put func()) {
	var f scenarioStep
	switch {
	case 6*i < o.ops:
		f = jumpStep
	case 2*i < o.ops:
		f = change
	case 3*i < 2*o.ops:
		f = jumpStep
	default:
		f = change
	}

	f(o, rnd, i, active, get, put)
}

func sinusStep(o scenarioOptions, rnd *rand.Rand, i, active int, get, put func()) {
	cycle := float64(o.ops)
	amp := cycle / 4
	x := float64(i)
	sin := amp * math.Sin(x*2*math.Pi/cycle)
	target := o.initial + int(sin)
	if target < 0 {
		target = 0
	}

	delta := target - active
	absDelta := delta
	if absDelta < 0 {
		absDelta = 0 - absDelta
	}

	absR := rnd.Intn(o.deviation + absDelta)
	r := absR - o.deviation

	var grow bool
	switch {
	case delta >= 0 && r >= 0:
		grow = true
	case delta <= 0 && r < 0:
		grow = true
	}

	f := put
	if grow {
		f = get
	}

	f()
}

func noopVerify(*testing.T, scenarioOptions, []pool.Stats) {}

func verifySteady(t *testing.T, o scenarioOptions, s []pool.Stats) {
	for i, stats := range s {
		if 2*stats.Idle > 3*(o.deviation+1)*o.concurrency {
			t.Fatal(i, stats)
		}
	}
}

func verifyAllocRate(t *testing.T, o scenarioOptions, s []pool.Stats) {
	if len(s) == 0 {
		t.Fatal("no stats")
	}

	last := s[len(s)-1]
	if (last.Alloc-o.initial)*3 > last.Get*2 {
		t.Fatal("too many allocations", last)
	}
}

func verifyAllocRateLax(t *testing.T, o scenarioOptions, s []pool.Stats) {
	if len(s) == 0 {
		t.Fatal("no stats")
	}

	last := s[len(s)-1]
	if (last.Alloc-o.initial)*10 > last.Get*9 {
		t.Fatal("too many allocations", last)
	}
}

func verifyDealloc(t *testing.T, o scenarioOptions, s []pool.Stats) {
	if len(s) == 0 {
		t.Fatal("no stats")
	}

	last := s[len(s)-1]

	if last.Free > o.deviation*o.concurrency {
		t.Fatal("too many deallocations", last)
	}
}

func testSteadyUsage(t *testing.T, o scenarioOptions) {
	testScenario(t, o, steady, verifySteady)
}

func testSteadyStepUp(t *testing.T, o scenarioOptions) {
	testScenario(t, o, steadyStepUp, verifyAllocRate)
}

func testSteadyStepDown(t *testing.T, o scenarioOptions) {
	testScenario(t, o, steadyStepDown, verifyAllocRate)
}

func testSteadyStepDownToZero(t *testing.T, o scenarioOptions) {
	testScenario(t, o, steadyStepDownToZero, verifyAllocRate)
}

func testChange(t *testing.T, o scenarioOptions) {
	verify := verifyDealloc
	if o.changeRate < 0 {
		verify = verifyAllocRate
	}

	testScenario(t, o, change, verify)
}

func testChangeAndJump(t *testing.T, o scenarioOptions) {
	testScenario(t, o, changeAndJump, verifyAllocRate)
}

func testCyclicSpikes(t *testing.T, o scenarioOptions) {
	verify := noopVerify
	if o.initial > 0 {
		verify = verifyAllocRate
	}

	testCyclicScenario(t, o, cyclicSpikes, verify)
}

func testCyclicDrops(t *testing.T, o scenarioOptions) {
	testCyclicScenario(t, o, cyclicDrops, verifyAllocRate)
}

func testCyclicSigmaSteps(t *testing.T, o scenarioOptions) {
	testCyclicScenario(t, o, sigmaSteps, verifyAllocRate)
}

func testCyclicChainSaw(t *testing.T, o scenarioOptions) {
	testCyclicScenario(t, o, chainSaw, verifyAllocRate)
}

func testCyclicInverseChainSaw(t *testing.T, o scenarioOptions) {
	testCyclicScenario(t, o, inverseChainSaw, verifyAllocRateLax)
}

func testCyclicSinus(t *testing.T, o scenarioOptions) {
	o.initial += o.ops / 16 // a single cycle is o.ops / 4, the amp in sinusStep is cycle ops / 4
	testCyclicScenario(t, o, sinusStep, verifyAllocRate)
}

func testBasicSet(t *testing.T, base scenarioOptions) {
	t.Run("steady minimal", func(t *testing.T) {
		o := base
		o.initial = 1
		o.ops = 60
		o.deviation = 1
		testSteadyUsage(t, o)
	})

	t.Run("steady small", func(t *testing.T) {
		o := base
		o.initial = 8
		o.ops = 60
		o.deviation = 2
		testSteadyUsage(t, o)
	})

	t.Run("steady large", func(t *testing.T) {
		o := base
		o.initial = 60
		o.ops = 1200
		o.deviation = 10
		testSteadyUsage(t, o)
	})

	t.Run("steady step up small", func(t *testing.T) {
		o := base
		o.initial = 8
		o.ops = 60
		o.deviation = 2
		testSteadyStepUp(t, o)
	})

	t.Run("steady step up large", func(t *testing.T) {
		o := base
		o.initial = 60
		o.ops = 1200
		o.deviation = 10
		testSteadyStepUp(t, o)
	})

	t.Run("steady step down small", func(t *testing.T) {
		o := base
		o.initial = 20
		o.ops = 60
		o.deviation = 2
		testSteadyStepDown(t, o)
	})

	t.Run("steady step down large", func(t *testing.T) {
		o := base
		o.initial = 450
		o.ops = 1200
		o.deviation = 10
		testSteadyStepDown(t, o)
	})

	t.Run("steady step down small to zero", func(t *testing.T) {
		o := base
		o.initial = 20
		o.ops = 60
		o.deviation = 2
		testSteadyStepDownToZero(t, o)
	})

	t.Run("steady step down large to zero", func(t *testing.T) {
		o := base
		o.initial = 450
		o.ops = 1200
		o.deviation = 10
		testSteadyStepDownToZero(t, o)
	})

	t.Run("slow rise from zero small", func(t *testing.T) {
		o := base
		o.ops = 60
		o.changeRate = 1
		o.deviation = 3
		testChange(t, o)
	})

	t.Run("slow rise from zero large", func(t *testing.T) {
		o := base
		o.ops = 1200
		o.changeRate = 1
		o.deviation = 10
		testChange(t, o)
	})

	t.Run("slow decrease to zero small", func(t *testing.T) {
		o := base
		o.initial = 15
		o.ops = 60
		o.changeRate = -1
		o.deviation = 3
		testChange(t, o)
	})

	t.Run("slow decrease to zero large", func(t *testing.T) {
		o := base
		o.initial = 300
		o.ops = 1200
		o.changeRate = -1
		o.deviation = 10
		testChange(t, o)
	})

	t.Run("slow decrease and jump small", func(t *testing.T) {
		o := base
		o.initial = 15
		o.ops = 60
		o.changeRate = -2
		o.deviation = 3
		testChangeAndJump(t, o)
	})

	t.Run("slow decrease and jump large", func(t *testing.T) {
		o := base
		o.initial = 300
		o.ops = 1200
		o.changeRate = -2
		o.deviation = 10
		testChangeAndJump(t, o)
	})
}

func testCyclicSet(t *testing.T, base scenarioOptions) {
	t.Run("cyclic spikes from zero small", func(t *testing.T) {
		o := base
		o.ops = 300
		o.deviation = 3
		testCyclicSpikes(t, o)
	})

	t.Run("cyclic spikes from zero large", func(t *testing.T) {
		o := base
		o.ops = 6000
		o.deviation = 10
		testCyclicSpikes(t, o)
	})

	t.Run("steady and cyclic spikes small", func(t *testing.T) {
		o := base
		o.initial = 30
		o.ops = 300
		o.deviation = 3
		testCyclicSpikes(t, o)
	})

	t.Run("steady and cyclic spikes large", func(t *testing.T) {
		o := base
		o.initial = 300
		o.ops = 6000
		o.deviation = 10
		testCyclicSpikes(t, o)
	})

	t.Run("steady and cyclic drops to zero small", func(t *testing.T) {
		o := base
		o.initial = 30
		o.ops = 300
		o.deviation = 3
		testCyclicDrops(t, o)
	})

	t.Run("steady and cyclic drops to zero large", func(t *testing.T) {
		o := base
		o.initial = 420
		o.ops = 6000
		o.deviation = 10
		testCyclicDrops(t, o)
	})

	t.Run("sigma steps from zero small", func(t *testing.T) {
		o := base
		o.ops = 300
		o.deviation = 3
		testCyclicSigmaSteps(t, o)
	})

	t.Run("sigma steps from zero large", func(t *testing.T) {
		o := base
		o.ops = 6000
		o.deviation = 10
		testCyclicSigmaSteps(t, o)
	})

	t.Run("sigma steps small", func(t *testing.T) {
		o := base
		o.initial = 60
		o.ops = 300
		o.deviation = 3
		testCyclicSigmaSteps(t, o)
	})

	t.Run("sigma steps large", func(t *testing.T) {
		o := base
		o.initial = 600
		o.ops = 6000
		o.deviation = 10
		testCyclicSigmaSteps(t, o)
	})

	t.Run("chain saw from zero small", func(t *testing.T) {
		o := base
		o.ops = 300
		o.deviation = 3
		o.changeRate = 2
		testCyclicChainSaw(t, o)
	})

	t.Run("chain saw from zero large", func(t *testing.T) {
		o := base
		o.ops = 6000
		o.deviation = 3
		o.changeRate = 2
		testCyclicChainSaw(t, o)
	})

	t.Run("chain saw small", func(t *testing.T) {
		o := base
		o.initial = 60
		o.ops = 300
		o.deviation = 3
		o.changeRate = 2
		testCyclicChainSaw(t, o)
	})

	t.Run("chain saw large", func(t *testing.T) {
		o := base
		o.initial = 600
		o.ops = 6000
		o.deviation = 3
		o.changeRate = 2
		testCyclicChainSaw(t, o)
	})

	t.Run("inverse chain saw from zero small", func(t *testing.T) {
		o := base
		o.ops = 300
		o.deviation = 3
		o.changeRate = -3
		testCyclicInverseChainSaw(t, o)
	})

	t.Run("inverse chain saw from zero large", func(t *testing.T) {
		o := base
		o.ops = 6000
		o.deviation = 3
		o.changeRate = -3
		testCyclicInverseChainSaw(t, o)
	})

	t.Run("inverse chain saw small", func(t *testing.T) {
		o := base
		o.initial = 60
		o.ops = 300
		o.deviation = 3
		o.changeRate = -3
		testCyclicInverseChainSaw(t, o)
	})

	t.Run("inverse chain saw large", func(t *testing.T) {
		o := base
		o.initial = 600
		o.ops = 6000
		o.deviation = 3
		o.changeRate = -3
		testCyclicInverseChainSaw(t, o)
	})

	t.Run("sinus to zero small", func(t *testing.T) {
		o := base
		o.ops = 300
		o.deviation = 3
		testCyclicSinus(t, o)
	})

	t.Run("sinus to zero large", func(t *testing.T) {
		o := base
		o.ops = 6000
		o.deviation = 10
		testCyclicSinus(t, o)
	})

	t.Run("sinus small", func(t *testing.T) {
		o := base
		o.initial = 15
		o.ops = 300
		o.deviation = 3
		testCyclicSinus(t, o)
	})

	t.Run("sinus large", func(t *testing.T) {
		o := base
		o.initial = 60
		o.ops = 6000
		o.deviation = 10
		testCyclicSinus(t, o)
	})
}