orchard

  1package main
  2
  3import (
  4	"context"
  5	"sync"
  6	"testing"
  7	"time"
  8)
  9
 10func TestQueueSerialises(t *testing.T) {
 11	q := NewQueue()
 12	var (
 13		mu      sync.Mutex
 14		running int
 15		peak    int
 16		order   []int
 17	)
 18
 19	var wg sync.WaitGroup
 20	for i := 0; i < 5; i++ {
 21		wg.Add(1)
 22		go func(i int) {
 23			defer wg.Done()
 24			release, ok := q.Enter(context.Background(), nil)
 25			if !ok {
 26				t.Errorf("worker %d was refused", i)
 27				return
 28			}
 29			defer release()
 30
 31			mu.Lock()
 32			running++
 33			if running > peak {
 34				peak = running
 35			}
 36			order = append(order, i)
 37			mu.Unlock()
 38
 39			time.Sleep(20 * time.Millisecond)
 40
 41			mu.Lock()
 42			running--
 43			mu.Unlock()
 44		}(i)
 45		// Stagger so the order is deterministic enough to assert on.
 46		time.Sleep(5 * time.Millisecond)
 47	}
 48	wg.Wait()
 49
 50	if peak != 1 {
 51		t.Errorf("peak concurrency was %d, want 1", peak)
 52	}
 53	if len(order) != 5 {
 54		t.Errorf("only %d of 5 ran", len(order))
 55	}
 56}
 57
 58func TestQueueGivesUpOnCancel(t *testing.T) {
 59	q := NewQueue()
 60
 61	held, ok := q.Enter(context.Background(), nil)
 62	if !ok {
 63		t.Fatal("the first caller should run at once")
 64	}
 65
 66	// Second joins and then leaves before its turn.
 67	ctx, cancel := context.WithCancel(context.Background())
 68	left := make(chan bool, 1)
 69	go func() {
 70		_, ok := q.Enter(ctx, nil)
 71		left <- ok
 72	}()
 73	time.Sleep(30 * time.Millisecond)
 74
 75	if n, _ := q.Depth(); n != 1 {
 76		t.Fatalf("depth = %d, want 1 waiting", n)
 77	}
 78	cancel()
 79	if ok := <-left; ok {
 80		t.Error("a cancelled caller should not be given the slot")
 81	}
 82
 83	// A third must still get through once the first releases.
 84	held()
 85	done := make(chan bool, 1)
 86	go func() {
 87		r, ok := q.Enter(context.Background(), nil)
 88		if ok {
 89			r()
 90		}
 91		done <- ok
 92	}()
 93	select {
 94	case ok := <-done:
 95		if !ok {
 96			t.Error("the third caller was refused")
 97		}
 98	case <-time.After(2 * time.Second):
 99		t.Fatal("the queue stalled after a cancellation")
100	}
101}
102
103func TestQueueReportsPosition(t *testing.T) {
104	q := NewQueue()
105	release, _ := q.Enter(context.Background(), nil)
106
107	seen := make(chan QueueState, 4)
108	go func() {
109		r, ok := q.Enter(context.Background(), func(s QueueState) { seen <- s })
110		if ok {
111			r()
112		}
113	}()
114
115	select {
116	case s := <-seen:
117		if s.Position != 1 || s.Ahead != 0 {
118			t.Errorf("first waiter reported %+v, want position 1 ahead 0", s)
119		}
120	case <-time.After(2 * time.Second):
121		t.Fatal("no position was reported")
122	}
123	release()
124}