priorityq/npq/lib_test.go

package npq_test

import (
	"math/rand"
	"runtime"
	"sync"
	"testing"

	"gogs.humancabbage.net/sam/priorityq"
	"gogs.humancabbage.net/sam/priorityq/npq"
)

func TestRecvHighFirst(t *testing.T) {
	t.Parallel()
	q := npq.Make[int, int](4, 1, 2)
	q.Send(1, 1)
	q.Send(1, 2)
	q.Send(1, 3)
	q.Send(1, 4)
	q.Send(2, 5)
	q.Send(2, 6)
	q.Send(2, 7)
	q.Send(2, 8)
	checkRecv := func(n int) {
		if v, _ := q.Recv(); v != n {
			t.Errorf("popped %d, expected %d", v, n)
		}
	}
	checkRecv(5)
	checkRecv(6)
	checkRecv(7)
	checkRecv(8)
	checkRecv(1)
	checkRecv(2)
	checkRecv(3)
	checkRecv(4)
}

func TestSendClosedPanic(t *testing.T) {
	t.Parallel()
	defer func() {
		if r := recover(); r == nil {
			t.Errorf("sending to closed queue did not panic")
		}
	}()
	q := npq.Make[int, int](4, 1, 2)
	q.Close()
	q.Send(1, 1)
}

func TestTooLowPriorityPanic(t *testing.T) {
	testInvalidPriorityPanic(t, 0)
}

func TestTooHighPriorityPanic(t *testing.T) {
	testInvalidPriorityPanic(t, 3)
}

func testInvalidPriorityPanic(t *testing.T, invalid int) {
	t.Parallel()
	defer func() {
		if r := recover(); r == nil {
			t.Errorf("sending with invalid priority did not panic")
		}
	}()
	q := npq.Make[int, int](4, 1, 2)
	q.Send(invalid, 1)

}

func TestRecvClosed(t *testing.T) {
	t.Parallel()
	q := npq.Make[int, int](4, 1, 2)
	q.Send(1, 1)
	q.Close()
	_, ok := q.Recv()
	if !ok {
		t.Errorf("queue should have item to receive")
	}
	_, ok = q.Recv()
	if ok {
		t.Errorf("queue should be closed")
	}
}

func TestDoubleClose(t *testing.T) {
	t.Parallel()
	q := npq.Make[int, int](4, 1, 2)
	defer func() {
		if r := recover(); r == nil {
			t.Errorf("closing a closed queue did not panic")
		}
	}()
	q.Close()
	q.Close()
}

func TestTrySendRecv(t *testing.T) {
	t.Parallel()
	q := npq.Make[int, int](4, 1, 2)
	assumeSendOk := func(n int, f func(int) error) {
		err := f(n)
		if err != nil {
			t.Errorf("expected to be able to send")
		}
	}
	assumeRecvOk := func(expected int) {
		actual, err := q.TryRecv()
		if err != nil {
			t.Errorf("expected to be able to receive")
		}
		if actual != expected {
			t.Errorf("expected %d, got %d", expected, actual)
		}
	}
	trySendLow := func(n int) error {
		return q.TrySend(1, n)
	}
	trySendHigh := func(n int) error {
		return q.TrySend(2, n)
	}
	assumeSendOk(1, trySendLow)
	assumeSendOk(2, trySendLow)
	assumeSendOk(3, trySendLow)
	assumeSendOk(4, trySendLow)
	err := trySendLow(5)
	if err == nil {
		t.Errorf("expected low buffer to be full")
	}
	assumeRecvOk(1)
	assumeRecvOk(2)
	assumeRecvOk(3)
	assumeRecvOk(4)

	assumeSendOk(5, trySendHigh)
	assumeSendOk(6, trySendHigh)
	assumeSendOk(7, trySendHigh)
	assumeSendOk(8, trySendHigh)
	err = trySendHigh(5)
	if err == nil {
		t.Errorf("expected high buffer to be full")
	}
	assumeRecvOk(5)
	assumeRecvOk(6)
	assumeRecvOk(7)
	assumeRecvOk(8)

	_, err = q.TryRecv()
	if err != priorityq.ErrEmpty {
		t.Errorf("expected queue to be empty")
	}
	q.Close()
	_, err = q.TryRecv()
	if err != priorityq.ErrClosed {
		t.Errorf("expected queue to be closed ")
	}
	err = q.TrySend(1, 5)
	if err != priorityq.ErrClosed {
		t.Errorf("expected queue to be closed ")
	}
}

func TestConcProducerConsumer(t *testing.T) {
	t.Parallel()
	q := npq.Make[int, int](4, 1, 2)
	var wg sync.WaitGroup
	produceDone := make(chan struct{})
	wg.Add(2)
	go func() {
		for i := 0; i < 10000; i++ {
			q.Send(rand.Intn(2)+1, i)
		}
		close(produceDone)
		wg.Done()
	}()
	go func() {
		ok := true
		for ok {
			_, ok = q.Recv()
		}
		wg.Done()
	}()
	<-produceDone
	t.Logf("producer done, closing channel")
	q.Close()
	wg.Wait()
}

const highPriority = 2

func BenchmarkSend(b *testing.B) {
	q := npq.Make[int, int](b.N, 1, highPriority)
	// randomize priorities to get amortized cost per op
	ps := make([]int, b.N)
	for i := 0; i < b.N; i++ {
		ps[i] = rand.Intn(highPriority) + 1
	}
	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		q.Send(ps[i], i)
	}
}

func BenchmarkRecv(b *testing.B) {
	q := npq.Make[int, int](b.N, 1, highPriority)
	for i := 0; i < b.N; i++ {
		q.Send(rand.Intn(highPriority)+1, i)
	}
	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		q.Recv()
	}
}

func BenchmarkConcSendRecv(b *testing.B) {
	q := npq.Make[int, int](b.N, 1, highPriority)
	// randomize priorities to get amortized cost per op
	ps := make([]int, b.N)
	for i := 0; i < b.N; i++ {
		ps[i] = rand.Intn(highPriority) + 1
	}
	var wg sync.WaitGroup
	wg.Add(2)
	start := make(chan struct{})
	go func() {
		<-start
		for i := 0; i < b.N; i++ {
			q.Send(ps[i], i)
		}
		wg.Done()
	}()
	go func() {
		<-start
		for i := 0; i < b.N; i++ {
			q.Recv()
		}
		wg.Done()
	}()
	b.ResetTimer()
	close(start)
	wg.Wait()
}

func BenchmarkHighContention(b *testing.B) {
	q := npq.Make[int, int](b.N, 1, highPriority)
	var wg sync.WaitGroup
	start := make(chan struct{})
	done := make(chan struct{})
	numProducers := runtime.NumCPU()
	sendsPerProducer := b.N / numProducers
	wg.Add(numProducers)
	for i := 0; i < numProducers; i++ {
		go func() {
			ps := make([]int, sendsPerProducer)
			for i := range ps {
				ps[i] = rand.Intn(highPriority) + 1
			}
			<-start
			for i := 0; i < sendsPerProducer; i++ {
				q.Send(ps[i], 1)
			}
			wg.Done()
		}()
	}
	go func() {
		ok := true
		for ok {
			_, ok = q.Recv()
		}
		close(done)
	}()
	b.ResetTimer()
	close(start)
	wg.Wait()
	q.Close()
	<-done
}