priorityq/pq/lib_test.go

package pq_test

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

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

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

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

func TestRecvClosed(t *testing.T) {
	t.Parallel()
	q := pq.Make[int, int](4)
	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 := pq.Make[int, int](4)
	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 := pq.Make[int, int](4)
	assumeSendOk := func(n int) {
		err := q.TrySend(n, 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)
		}
	}
	assumeSendOk(1)
	assumeSendOk(2)
	assumeSendOk(3)
	assumeSendOk(4)
	err := q.TrySend(5, 5)
	if err == nil {
		t.Errorf("expected queue to be full")
	}
	assumeRecvOk(4)
	assumeRecvOk(3)
	assumeRecvOk(2)
	assumeRecvOk(1)

	_, _, 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, 1)
	if err != priorityq.ErrClosed {
		t.Errorf("expected queue to be closed ")
	}
}

func TestConcProducerConsumer(t *testing.T) {
	t.Parallel()
	q := pq.Make[int, int](4)
	var wg sync.WaitGroup
	produceDone := make(chan struct{})
	wg.Add(2)
	go func() {
		for i := 0; i < 10000; i++ {
			q.Send(rand.Int(), 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()
}

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

func BenchmarkRecv(b *testing.B) {
	q := pq.Make[int, int](b.N)
	// randomize priorities to get amortized cost per op
	for i := 0; i < b.N; i++ {
		q.Send(rand.Int(), i)
	}
	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		q.Recv()
	}
}

func BenchmarkConcSendRecv(b *testing.B) {
	q := pq.Make[int, int](b.N)
	// randomize priorities to get amortized cost per op
	ps := make([]int, b.N)
	for i := 0; i < b.N; i++ {
		ps[i] = rand.Int()
	}
	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 := pq.Make[int, int](b.N)
	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 := 0; i < sendsPerProducer; i++ {
				ps[i] = rand.Int()
			}
			<-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
}