fixed_stack_test.go

// ©Hayabusa Cloud Co., Ltd. 2023. All rights reserved.
// Use of this source code is governed by a MIT-style
// license that can be found in the LICENSE file.

package sox_test

import (
	"hybscloud.com/sox"
	"io"
	"math"
	"sync"
	"testing"
)

func TestFixedStack_Series(t *testing.T) {
	t.Run("a little serial ops", func(t *testing.T) {
		s, err := sox.NewFixedStack[uintptr](func(options *sox.FixedStackOptions) {
			options.Capacity = 0x3
			options.Concurrent = false
			options.Nonblocking = true
		})
		if err != nil {
			t.Errorf("fixed stack new: %v", err)
			return
		}
		testFixedStackNonblocking(t, s)
	})

	t.Run("parallel produce and consume", func(t *testing.T) {
		s, err := sox.NewFixedStack[int](func(options *sox.FixedStackOptions) {
			options.Concurrent = false
			options.Nonblocking = false
		})
		if err != nil {
			t.Errorf("fixed stack new: %v", err)
			return
		}
		go func() {
			for i := range 1 << 16 {
				err := s.Push(i)
				if err != nil {
					t.Errorf("fixed stack push: %v", err)
					break
				}
			}
			err := s.Close()
			if err != nil {
				t.Errorf("fixed stack close: %v", err)
				return
			}
		}()
		stack, index, next := make([]int, 0), make([]int, 1<<16), 0
		for i := range 1 << 16 {
			index[i] = -1
		}
		for range 1 << 16 {
			item, err := s.Pop()
			if err != nil {
				t.Errorf("fixed stack pop: %v", err)
				return
			}
			for j := next; j <= item; j++ {
				if index[j] >= 0 && stack[len(stack)-1] != j {
					t.Errorf("fixed stack bad result")
					return
				}
				index[j] = len(stack)
				stack = append(stack, j)
			}
			if next < item+1 {
				next = item + 1
			}
			stack = stack[:len(stack)-1]
		}
		_, err = s.Pop()
		if err != nil && err != io.EOF {
			t.Errorf("fixed stack pop: %v", err)
			return
		}
	})
}

func BenchmarkFixedStack_Parallel(b *testing.B) {
	s, err := sox.NewFixedStack[int](func(options *sox.FixedStackOptions) {
		options.Concurrent = false
		options.Nonblocking = false
	})
	if err != nil {
		b.Errorf("fixed stack new: %v", err)
		return
	}
	b.ResetTimer()
	go func() {
		for i := range b.N {
			err := s.Push(i)
			if err != nil {
				b.Errorf("fixed stack push: %v", err)
				break
			}
		}
		err := s.Close()
		if err != nil {
			b.Errorf("fixed stack close: %v", err)
			return
		}
	}()
	for range b.N {
		_, err = s.Pop()
		if err != nil {
			b.Errorf("fixed stack pop: %v", err)
			return
		}
	}
}

func TestFixedStack_Concurrent(t *testing.T) {
	t.Run("a little ops", func(t *testing.T) {
		s, err := sox.NewFixedStack[uintptr](func(options *sox.FixedStackOptions) {
			options.Capacity = 0x3
			options.Concurrent = true
			options.Nonblocking = true
		})
		if err != nil {
			t.Errorf("fixed stack new: %v", err)
			return
		}
		testFixedStackNonblocking(t, s)
	})

	t.Run("4 push goroutines 4 pop goroutines 32k buffer", func(t *testing.T) {
		s, err := sox.NewFixedStack[int64](func(options *sox.FixedStackOptions) {
			options.Concurrent = true
			options.Nonblocking = false
		})
		if err != nil {
			t.Errorf("fixed stack new: %v", err)
			return
		}
		testFixedStackConcurrent(t, s, 0x04, 0x2000)
	})

	t.Run("16 push goroutines 16 pop goroutines 32k buffer", func(t *testing.T) {
		s, err := sox.NewFixedStack[int64](func(options *sox.FixedStackOptions) {
			options.Concurrent = true
			options.Nonblocking = false
		})
		if err != nil {
			t.Errorf("fixed stack new: %v", err)
			return
		}
		testFixedStackConcurrent(t, s, 0x10, 0x2000)
	})

	t.Run("64 push goroutines 64 pop goroutines 32k buffer", func(t *testing.T) {
		s, err := sox.NewFixedStack[int64](func(options *sox.FixedStackOptions) {
			options.Concurrent = true
			options.Nonblocking = false
		})
		if err != nil {
			t.Errorf("fixed stack new: %v", err)
			return
		}
		testFixedStackConcurrent(t, s, 0x40, 0x2000)
	})

}

func BenchmarkFixedStackConcurrent(b *testing.B) {
	b.Run("1 push goroutine 1 pop goroutine", func(b *testing.B) {
		s, err := sox.NewFixedStack[int](func(options *sox.FixedStackOptions) {
			options.Concurrent = true
			options.Nonblocking = false
		})
		if err != nil {
			b.Errorf("fixed stack new: %v", err)
			return
		}
		b.ResetTimer()
		benchmarkFixedStackConcurrent(b, s, 1)
	})

	b.Run("4 push goroutines 4 pop goroutines", func(b *testing.B) {
		s, err := sox.NewFixedStack[int](func(options *sox.FixedStackOptions) {
			options.Concurrent = true
			options.Nonblocking = false
		})
		if err != nil {
			b.Errorf("fixed stack new: %v", err)
			return
		}
		b.ResetTimer()
		benchmarkFixedStackConcurrent(b, s, 4)
	})

	b.Run("16 push goroutines 16 pop goroutines", func(b *testing.B) {
		s, err := sox.NewFixedStack[int](func(options *sox.FixedStackOptions) {
			options.Concurrent = true
			options.Nonblocking = false
		})
		if err != nil {
			b.Errorf("fixed stack new: %v", err)
			return
		}
		b.ResetTimer()
		benchmarkFixedStackConcurrent(b, s, 16)
	})

	b.Run("64 push goroutines 64 pop goroutines", func(b *testing.B) {
		s, err := sox.NewFixedStack[int](func(options *sox.FixedStackOptions) {
			options.Concurrent = true
			options.Nonblocking = false
		})
		if err != nil {
			b.Errorf("fixed stack new: %v", err)
			return
		}
		b.ResetTimer()
		benchmarkFixedStackConcurrent(b, s, 64)
	})
}

func testFixedStackNonblocking(t *testing.T, s sox.Stack[uintptr]) {
	item, err := s.Pop()
	if err != sox.ErrTemporarilyUnavailable {
		t.Errorf("fixed stack pop expected ErrTemporarilyUnavailable but got %v %v", item, err)
		return
	}
	err = s.Push(1)
	if err != nil {
		t.Errorf("fixed stack push: %v", err)
		return
	}
	item, err = s.Pop()
	if err != nil {
		t.Errorf("fixed stack pop: %v", err)
		return
	}
	if item != 1 {
		t.Errorf("fixed stack pop item expected %d but got %d", 1, item)
		return
	}
	item, err = s.Pop()
	if err != sox.ErrTemporarilyUnavailable {
		t.Errorf("fixed stack expected ErrTemporarilyUnavailable but got %v %v", item, err)
		return
	}
	err = s.Push(2)
	if err != nil {
		t.Errorf("fixed stack push: %v", err)
		return
	}
	err = s.Push(3)
	if err != nil {
		t.Errorf("fixed stack push: %v", err)
		return
	}
	err = s.Push(4)
	if err != nil {
		t.Errorf("fixed stack push: %v", err)
		return
	}
	item, err = s.Pop()
	if err != nil {
		t.Errorf("fixed stack pop: %v", err)
		return
	}
	if item != 4 {
		t.Errorf("fixed stack pop item expected %d but got %d", 4, item)
		return
	}
	err = s.Push(5)
	if err != nil {
		t.Errorf("fixed stack push: %v", err)
		return
	}
	err = s.Push(6)
	if err != sox.ErrTemporarilyUnavailable {
		t.Errorf("fixed stack expected ErrTemporarilyUnavailable but got %v", err)
		return
	}
	item, err = s.Pop()
	if err != nil {
		t.Errorf("fixed stack pop: %v", err)
		return
	}
	if item != 5 {
		t.Errorf("fixed stack pop item expected %d but got %d", 5, item)
		return
	}
	item, err = s.Pop()
	if err != nil {
		t.Errorf("fixed stack pop: %v", err)
		return
	}
	if item != 3 {
		t.Errorf("fixed stack pop item expected %d but got %d", 3, item)
		return
	}
	err = s.Push(7)
	if err != nil {
		t.Errorf("fixed stack push: %v", err)
		return
	}
	item, err = s.Pop()
	if err != nil {
		t.Errorf("fixed stack pop: %v", err)
		return
	}
	if item != 7 {
		t.Errorf("fixed stack pop item expected %d but got %d", 7, item)
		return
	}
	item, err = s.Pop()
	if err != nil {
		t.Errorf("fixed stack pop: %v", err)
		return
	}
	if item != 2 {
		t.Errorf("fixed stack pop item expected %d but got %d", 2, item)
		return
	}
	item, err = s.Pop()
	if err != sox.ErrTemporarilyUnavailable {
		t.Errorf("fixed stack pop expected %v but got %v %v", sox.ErrTemporarilyUnavailable, item, err)
		return
	}
	err = s.Close()
	if err != nil {
		t.Errorf("fixed stack pop: %v", err)
		return
	}
	item, err = s.Pop()
	if err != io.EOF {
		t.Errorf("fixed stack pop expected %v but got %v %v", io.EOF, item, err)
		return
	}
}

func testFixedStackConcurrent(t *testing.T, s sox.Stack[int64], m int, n int) {
	stack, index, next := make([][]int64, m), make([][]int, m), make([]int64, m)
	locks := make([]sync.Mutex, m)
	for i := range m {
		stack[i] = make([]int64, 0, n+1)
		index[i] = make([]int, n)
		for j := range n {
			index[i][j] = -1
		}
		next[i] = 0
	}
	wg := sync.WaitGroup{}
	for range m {
		wg.Add(1)
		go func() {
			for range n {
				item, err := s.Pop()
				if err != nil {
					t.Errorf("fixed stack pop: %v", err)
					return
				}
				high, low := item>>32, item&math.MaxUint32
				locks[high].Lock()
				st, idx, val := stack[high], index[high], low
				for j := next[high]; j <= val; j++ {
					if idx[j] >= 0 && st[len(st)-1] != j {
						t.Errorf("fixed stack bad result")
						return
					}
					idx[j] = len(st)
					st = append(st, j)
				}
				if next[high] < val+1 {
					next[high] = val + 1
				}
				st = st[:len(st)-1]
				stack[high] = st
				locks[high].Unlock()
			}
			wg.Done()
		}()
	}
	for i := range m {
		go func(i int) {
			for j := range n {
				err := s.Push(int64(i<<32) | int64(j))
				if err != nil {
					t.Errorf("fixed stack push: %v", err)
					return
				}
			}
		}(i)
	}
	wg.Wait()
}

func benchmarkFixedStackConcurrent(b *testing.B, s sox.Stack[int], m int) {
	for i := range m {
		go func(i int) {
			for j := range b.N/m + 1 {
				err := s.Push(j)
				if err != nil {
					b.Errorf("fixed stack push: %v", err)
					return
				}
			}
		}(i)
	}
	wg := sync.WaitGroup{}
	for i := range m {
		wg.Add(1)
		go func(i int) {
			for range b.N / m {
				_, err := s.Pop()
				if err != nil {
					b.Errorf("fixed stack pop: %v", err)
					return
				}
			}
			wg.Done()
		}(i)
	}
	wg.Wait()
}